A background task that updates the state.
(self)
| 1693 | |
| 1694 | @xt.background |
| 1695 | async def background_task(self): |
| 1696 | """A background task that updates the state.""" |
| 1697 | async with self: |
| 1698 | assert not self.order |
| 1699 | self.order.append("background_task:start") |
| 1700 | |
| 1701 | assert isinstance(self, StateProxy) |
| 1702 | with pytest.raises(ImmutableStateError): |
| 1703 | self.order.append("bad idea") |
| 1704 | |
| 1705 | with pytest.raises(ImmutableStateError): |
| 1706 | # Even nested access to mutables raises an exception. |
| 1707 | self.dict_list["foo"].append(42) |
| 1708 | |
| 1709 | with pytest.raises(ImmutableStateError): |
| 1710 | # Direct calling another handler that modifies state raises an exception. |
| 1711 | self.other() |
| 1712 | |
| 1713 | with pytest.raises(ImmutableStateError): |
| 1714 | # Calling other methods that modify state raises an exception. |
| 1715 | self._private_method() |
| 1716 | |
| 1717 | # wait for some other event to happen |
| 1718 | while len(self.order) == 1: |
| 1719 | await asyncio.sleep(0.01) |
| 1720 | async with self: |
| 1721 | pass # update proxy instance |
| 1722 | |
| 1723 | async with self: |
| 1724 | # Methods on ImmutableMutableProxy should return their wrapped return value. |
| 1725 | assert self.dict_list.pop("foo") == [1, 2, 3] |
| 1726 | |
| 1727 | self.order.append("background_task:stop") |
| 1728 | self.other() # direct calling event handlers works in context |
| 1729 | self._private_method() |
| 1730 | |
| 1731 | @xt.background |
| 1732 | async def background_task_reset(self): |
no test coverage detected