diff --git a/docs/async.md b/docs/async.md index fbc18992..07427e40 100644 --- a/docs/async.md +++ b/docs/async.md @@ -152,7 +152,8 @@ processing loop at a time. The sync engine does not support this pattern. Under the hood, the async engine attaches an `asyncio.Future` to each externally enqueued event. The coroutine that acquires the processing lock resolves each event's future as it processes the queue. Callers that didn't -acquire the lock simply `await` their future. +acquire the lock `await` their future, or the release of the lock, to take +the loop over if the coroutine holding it was cancelled. ```{note} Futures are created for every event sent from outside the processing loop, @@ -165,9 +166,18 @@ model: they are enqueued and processed by the running loop. If an exception occurs during processing (with `catch_errors_as_events=False`), the exception is routed to the caller whose event raised it. Other callers whose events were still pending will also receive the exception, since the -processing loop clears the queue on failure. An exception raised by the -internal events or eventless transitions that follow an event goes only to -the callers still pending: the caller of that event already has its result. +processing loop clears the queue on failure. A caller is answered once the +macrostep of its event ends, so an exception raised by the internal events or +eventless transitions that follow its event reaches it too. When no coroutine +awaits that event, because a callback sent it or its caller was cancelled while +awaiting it, the exception goes to the coroutine running the processing loop. + +If the coroutine running the processing loop is cancelled, only that coroutine +stops: a coroutine awaiting a queued event takes the loop over, and returns +once the queue is empty. With no coroutine waiting, the events left in the +queue run at the next `send()`. The caller of the event being processed when +the cancellation hit gets a `StateMachineError`, since its callbacks were +interrupted. (syncengine)= diff --git a/docs/releases/3.2.2.md b/docs/releases/3.2.2.md index 64c40671..eed34b1e 100644 --- a/docs/releases/3.2.2.md +++ b/docs/releases/3.2.2.md @@ -74,6 +74,9 @@ fail after upgrading: transition. Internal events raised before a callback fails, and events sent before a failed lazy initial activation on the async engine, are dropped too, instead of running at the next event. See {ref}`release-3.2.2-macrostep`. +- On the async engine, with `catch_errors_as_events=False`, `await send()` raises an exception + that escapes the internal events or eventless transitions following its event. It used to + return the event's result, and the exception was lost. See {ref}`release-3.2.2-async-loop`. - An attribute named with the `error_`, `done_state_` or `done_invoke_` prefix now registers two events, so `events`, `allowed_events` and `enabled_events()` list both ids. See {ref}`release-3.2.2-prefixed-events`. @@ -377,18 +380,82 @@ dropped event gets that exception. Any exception escapes with `catch_errors_as_e as on a `StateMachine`; on a `StateChart`, one from a validator or `prepare_event` does. On 3.2.1 the queued external events ran before those internal events and eventless transitions, and their callers got their own results; the exception was then raised by the sync `send()` -that started processing, and on the async engine it reached no caller. On the async engine, as -on 3.2.1, the coroutine whose event enabled the failing transitions gets its own event's -result. Internal events raised by a callback before it fails are dropped too: they used to run -at the next event. The same applies when the lazy initial activation fails on the async engine, -where a second coroutine awaiting an event sent before it used to wait forever, and the events -sent before it, which used to run at the next `send()`, are dropped. +that started processing, and on the async engine it reached no caller. The coroutine whose +event enabled the failing transitions also gets the exception, see +{ref}`release-3.2.2-async-loop`. Internal events raised by a callback before it fails are +dropped too: they used to run at the next event. The same applies when the lazy initial +activation fails on the async engine, where a second coroutine awaiting an event sent before it +used to wait forever, and the events sent before it, which used to run at the next `send()`, +are dropped. [#668](https://github.com/fgmacedo/python-statemachine/pull/668), [#678](https://github.com/fgmacedo/python-statemachine/issues/678), [#686](https://github.com/fgmacedo/python-statemachine/issues/686), [#685](https://github.com/fgmacedo/python-statemachine/pull/685). +(release-3.2.2-async-loop)= + +#### Async engine: an exception after the event, and a cancelled processing loop + +**An exception raised after a coroutine's event was lost.** With +`catch_errors_as_events=False`, as on a `StateMachine`, an exception raised by the internal +events or eventless transitions that follow an external event reached no caller on the async +engine: `await send()` returned the event's result, and the exception was neither raised nor +logged. The sync engine raises it from `send()`. A coroutine is now answered once the +macrostep of its event ends, so the exception reaches it: + +```py +>>> import asyncio + +>>> class Alarm(StateMachine): +... idle = State(initial=True) +... armed = State() +... ringing = State() +... +... arm = idle.to(armed) +... trip = armed.to(ringing) +... reset = ringing.to(idle) +... +... async def on_enter_armed(self): +... await self.raise_("trip") +... +... async def on_trip(self): +... raise ValueError("sensor failure") + +>>> async def arm(): +... sm = Alarm() +... await sm.activate_initial_state() +... await sm.send("arm") + +>>> asyncio.run(arm()) +Traceback (most recent call last): +... +ValueError: sensor failure + +``` + +When no coroutine awaits the event whose macrostep failed, because a callback sent it or its +caller was cancelled while awaiting it, the coroutine running the processing loop gets the +exception, as the +sync engine raises it from the `send()` running the loop. The callers of the events still +queued get the exception as before. A `StateChart` is unchanged: the error becomes +`error.execution`. + +**Cancelling the coroutine running the processing loop left the others waiting.** When the +coroutine holding the processing loop was cancelled, as `asyncio.wait_for()` or a request +timeout does, the coroutines awaiting the `send()` of events queued behind it were not +answered until a later `send()` ran the loop again: `CancelledError` is not an `Exception`, so +the loop's error handling did not run. The cancellation now stops only that coroutine. The +internal events left by its macrostep are dropped, and a coroutine awaiting a queued event +takes the processing loop over, so each gets its own result, delayed events included. Like any +coroutine running the loop, the one taking over returns once the queue is empty. A coroutine +awaiting the event being processed when the cancellation hit, whose callbacks were interrupted +halfway, gets a `StateMachineError` saying so. With no coroutine waiting, the events left in +the queue, the cancelled coroutine's own included, run at the next `send()`. + +[#689](https://github.com/fgmacedo/python-statemachine/issues/689), +[#690](https://github.com/fgmacedo/python-statemachine/issues/690). + (release-3.2.2-prefixed-events)= #### Prefixed events were registered under a glued id diff --git a/statemachine/engines/async_.py b/statemachine/engines/async_.py index c2f9c09f..81675b03 100644 --- a/statemachine/engines/async_.py +++ b/statemachine/engines/async_.py @@ -7,7 +7,9 @@ from ..event_data import EventData from ..event_data import TriggerData from ..exceptions import InvalidDefinition +from ..exceptions import StateMachineError from ..exceptions import TransitionNotAllowed +from ..i18n import _ from ..orderedset import OrderedSet from ..state import State from .base import _ERROR_EXECUTION @@ -16,9 +18,42 @@ from .base import _running_engine if TYPE_CHECKING: + from ..statemachine import StateChart from ..transition import Transition +class _MacrostepCaller: + """The caller of the external event whose macrostep is running, answered once the + macrostep ends, so that an exception raised by its internal events or eventless + transitions still reaches it.""" + + def __init__(self): + self.future: "asyncio.Future[object] | None" = None + self.result: object = None + + def answer(self): + AsyncEngine._resolve_future(self.future, self.result) + self.future = None + self.result = None + + def reject(self, exc: Exception) -> bool: + """Reject the caller with ``exc``; returns whether a caller was still waiting.""" + if self.future is None or self.future.done(): + return False + self.future.set_exception(exc) + return True + + def interrupt(self): + self.reject( + StateMachineError( + _( + "The task running the processing loop was cancelled while it processed " + "this event." + ) + ) + ) + + class AsyncEngine(BaseEngine): """Async engine with full StateChart support. @@ -26,6 +61,10 @@ class AsyncEngine(BaseEngine): All pure-computation helpers are inherited from :class:`BaseEngine`. """ + def __init__(self, sm: "StateChart"): + super().__init__(sm) + self._lock_waiters: "set[asyncio.Future[object]]" = set() + def put(self, trigger_data: TriggerData, internal: bool = False): """Override to attach an asyncio.Future to a trigger put from outside the loop. @@ -50,12 +89,6 @@ def _resolve_future(future: "asyncio.Future[object] | None", result): if future is not None and not future.done(): future.set_result(result) - @staticmethod - def _reject_future(future: "asyncio.Future[object] | None", exc: Exception): - """Reject a future with the given exception, if present and not yet done.""" - if future is not None and not future.done(): - future.set_exception(exc) - def _reject_pending_futures(self, exc: Exception): """Reject all unresolved futures in the external queue.""" self.external_queue.reject_futures(exc) @@ -340,17 +373,45 @@ async def processing_loop(self, caller_future: "asyncio.Future[object] | None" = When ``caller_future`` is provided, the caller can ``await`` it to receive its own event's result — even if another coroutine holds the - processing lock. + processing lock. If that coroutine is cancelled, the caller takes the loop over. """ - if not self._processing.acquire(blocking=False): - # Another coroutine holds the lock and will process our event. - # Await the caller's future so we get our own result back. - if caller_future is not None: - return await caller_future - return None + while not self._processing.acquire(blocking=False): + if caller_future is None: + return None + await self._wait_for_result_or_release(caller_future) + if caller_future.done(): + return caller_future.result() + + try: + return await self._run_processing_loop(caller_future) + finally: + self._processing.release() + self._wake_lock_waiters() + + async def _wait_for_result_or_release(self, caller_future: "asyncio.Future[object]"): + """Wait until the caller's event is answered or the processing lock is released. + A caller cancelled here cancels its future, so its event, still processed, answers + nobody. + """ + released = caller_future.get_loop().create_future() + self._lock_waiters.add(released) + try: + await asyncio.wait((caller_future, released), return_when=asyncio.FIRST_COMPLETED) + except BaseException: + caller_future.cancel() + raise + finally: + self._lock_waiters.discard(released) + + def _wake_lock_waiters(self): + for released in self._lock_waiters: + self._resolve_future(released, None) + + async def _run_processing_loop(self, caller_future: "asyncio.Future[object] | None"): self._debug("%s Processing loop started: %s", self._log_id, self.sm.current_state_value) first_result = self._sentinel + macrostep = _MacrostepCaller() try: with _processing_step(self): took_events = True @@ -360,37 +421,35 @@ async def processing_loop(self, caller_future: "asyncio.Future[object] | None" = # Spawn invoke handlers for states entered during this macrostep. await self._invoke_manager.spawn_pending_async() + macrostep.answer() self._check_root_final_state() if not self.running: break took_external_events, first_result = await self._take_due_external_event( - first_result + first_result, macrostep ) took_events = took_events or took_external_events + macrostep.answer() if not self.running: self._drop_external_events() except Exception as exc: + answered = caller_future is None or caller_future.done() self._reject_pending_futures(exc) self.clear() - if caller_future is not None: - # Route the exception to the caller's future if still pending. - # If already resolved (caller's own event succeeded before a - # later event failed), suppress the exception — the caller will - # get their successful result via ``await future`` below, and - # the failing event's exception was already routed to *its* - # caller's future by ``_reject_future(event_future, ...)``. - self._reject_future(caller_future, exc) - else: + if not macrostep.reject(exc) and answered: raise - finally: - self._processing.release() + except BaseException: + self.internal_queue.clear() + macrostep.interrupt() + if caller_future is not None: + caller_future.cancel() + raise self._debug("%s Processing loop ended", self._log_id) result = first_result if first_result is not self._sentinel else None - # If the caller has a future, await it (already resolved by now). if caller_future is not None: # Resolve the future if it wasn't processed (e.g. machine terminated). self._resolve_future(caller_future, result) @@ -424,11 +483,12 @@ async def _run_macrostep(self) -> bool: took_events = True await self.microstep(list(enabled_transitions), internal_event) - async def _take_due_external_event(self, first_result): + async def _take_due_external_event(self, first_result, macrostep: "_MacrostepCaller"): """Take the external event at the head of the queue, if it is due. One at a time, so the macrostep that follows runs the internal events and eventless - transitions it enabled before the next external event is taken. + transitions it enabled before the next external event is taken. The event's caller is + left in ``macrostep``, to be answered once that macrostep ends. Returns whether the queue held an event, and ``first_result`` updated with the result of the first event that enabled a transition. @@ -458,35 +518,29 @@ async def _take_due_external_event(self, first_result): await self._enter_states(transitions, external_event, OrderedSet(), OrderedSet()) return True, first_result + macrostep.future = external_event.future result = await self._take_external_event(external_event) + if result is not self._sentinel: + macrostep.result = result if first_result is self._sentinel: first_result = result return True, first_result async def _take_external_event(self, external_event: TriggerData): - """Run one external event and settle its future. + """Run one external event. Returns the event's microstep result, or the sentinel when it enabled no transition. """ # Finalize + autoforward for active invocations self._invoke_manager.handle_external_event(external_event) - event_future = external_event.future - try: - enabled_transitions = await self.select_transitions(external_event) - self._debug("%s Enabled transitions: %s", self._log_id, enabled_transitions) - if enabled_transitions: - result = await self.microstep(list(enabled_transitions), external_event) - self._resolve_future(event_future, result) - return result - if not self.sm.allow_event_without_transition: - raise TransitionNotAllowed(external_event.event, self.sm.configuration) - # Event allowed but no transition — resolve with None - self._resolve_future(event_future, None) - return self._sentinel - except Exception as exc: - self._reject_future(event_future, exc) - raise + enabled_transitions = await self.select_transitions(external_event) + self._debug("%s Enabled transitions: %s", self._log_id, enabled_transitions) + if enabled_transitions: + return await self.microstep(list(enabled_transitions), external_event) + if not self.sm.allow_event_without_transition: + raise TransitionNotAllowed(external_event.event, self.sm.configuration) + return self._sentinel async def enabled_events(self, *args, **kwargs): sm = self.sm diff --git a/tests/test_async_futures.py b/tests/test_async_futures.py index e9eacf8a..cac60016 100644 --- a/tests/test_async_futures.py +++ b/tests/test_async_futures.py @@ -12,6 +12,7 @@ import pytest from statemachine.engines.base import EventQueue from statemachine.event_data import TriggerData +from statemachine.exceptions import StateMachineError from statemachine.exceptions import TransitionNotAllowed from statemachine import Event @@ -681,7 +682,8 @@ async def test_statemachine_callers_get_the_exception(self, failing, eventless): sm = gated(khazad_dum(StateMachine, failing, eventless))() holder, waiters = await cross_with_callers_queued(sm, *THREE_FLIGHTS) - await holder + with pytest.raises(ValueError, match=f"{failing} failed"): + await holder results = await asyncio.gather(*waiters, return_exceptions=True) assert [str(result) for result in results] == [f"{failing} failed"] * 3 assert sm.flights == 0 @@ -706,8 +708,7 @@ async def test_statechart_callers_get_their_results(self, failing, eventless): async def test_machine_takes_new_events_afterwards(self): sm = gated(khazad_dum(StateMachine, "on_falling"))() holder, waiters = await cross_with_callers_queued(sm, *THREE_FLIGHTS) - await holder - await asyncio.gather(*waiters, return_exceptions=True) + await asyncio.gather(holder, *waiters, return_exceptions=True) assert await sm.send("flee") == 1 assert sm.configuration_values == {"bridge"} @@ -725,9 +726,9 @@ async def test_delayed_and_cancelled_callers_are_answered(self): ) sm.cancel_event("horn") - await holder - results = await asyncio.gather(delayed, cancelled, flight, return_exceptions=True) + results = await asyncio.gather(holder, delayed, cancelled, flight, return_exceptions=True) assert [str(result) for result in results] == [ + "on_falling failed", "on_falling failed", "None", "on_falling failed", @@ -737,7 +738,7 @@ async def test_delayed_and_cancelled_callers_are_answered(self): @pytest.mark.timeout(5) @pytest.mark.asyncio() - async def test_failure_after_a_queued_event_answers_only_the_callers_behind_it(self): + async def test_failure_after_a_queued_event_answers_its_caller_and_those_behind_it(self): sm = gated(khazad_dum(StateMachine, "on_falling"))() holder, waiters = await cross_with_callers_queued( sm, ("flee", {}), ("retreat", {}), ("cross", {}), *THREE_FLIGHTS[:2], awake=False @@ -745,8 +746,8 @@ async def test_failure_after_a_queued_event_answers_only_the_callers_behind_it(s assert await holder == "crossed" results = await asyncio.gather(*waiters, return_exceptions=True) - assert results[:3] == [1, None, "crossed"] - assert [str(result) for result in results[3:]] == ["on_falling failed"] * 2 + assert results[:2] == [1, None] + assert [str(result) for result in results[2:]] == ["on_falling failed"] * 3 assert sm.flights == 1 @pytest.mark.timeout(5) @@ -755,9 +756,8 @@ async def test_statechart_validator_failure_answers_the_callers(self): sm = gated(khazad_dum(StateChart, "validate_falling"))() holder, waiters = await cross_with_callers_queued(sm, *THREE_FLIGHTS) - await holder - results = await asyncio.gather(*waiters, return_exceptions=True) - assert [str(result) for result in results] == ["validate_falling failed"] * 3 + results = await asyncio.gather(holder, *waiters, return_exceptions=True) + assert [str(result) for result in results] == ["validate_falling failed"] * 4 assert sm.flights == 0 @pytest.mark.timeout(5) @@ -832,6 +832,251 @@ def test_sync_engine_drops_the_events_queued_behind_the_failure(self, failing, e assert sm.send("flee") == 1 +class FlickeringLamp(StateMachine): + """Its initial state raises an internal event whose transition fails.""" + + off = State(initial=True) + lit = State(final=True) + + switch = off.to(lit) + flicker = Event(off.to.itself(internal=True), internal=True) + + async def on_enter_off(self): + self.flicker.put() + + def on_flicker(self): + raise ValueError("short circuit") + + +async def cancel_and_wait(task): + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + +class TestMacrostepErrorReachesTheCallerOfItsEvent: + """An exception escaping the internal events or eventless transitions that follow an + external event reaches the caller of that event, on both engines. When no caller awaits + that event, because a callback sent it, the exception reaches the caller running the loop. + + See: https://github.com/fgmacedo/python-statemachine/issues/689 + """ + + @pytest.mark.timeout(5) + @pytest.mark.parametrize("eventless", [False, True], ids=["internal", "eventless"]) + @pytest.mark.parametrize("failing", FAILING_CALLBACKS) + async def test_caller_gets_the_exception(self, sm_runner, failing, eventless): + sm = await sm_runner.start(khazad_dum(StateMachine, failing, eventless)) + + with pytest.raises(ValueError, match=f"{failing} failed"): + await sm_runner.send(sm, "cross") + + assert await sm_runner.send(sm, "flee") == 1 + + @pytest.mark.timeout(5) + async def test_statechart_caller_gets_its_result(self, sm_runner): + sm = await sm_runner.start(khazad_dum(StateChart, "on_falling")) + + assert await sm_runner.send(sm, "cross") == "crossed" + + @pytest.mark.timeout(5) + async def test_event_sent_by_a_callback_fails_on_the_caller_running_the_loop(self, sm_runner): + sm = await sm_runner.start(khazad_dum(StateMachine, "on_falling")) + + with pytest.raises(ValueError, match="on_falling failed"): + await sm_runner.send(sm, "cross", awake=False, queued=("retreat", "cross")) + + assert sm.configuration_values == {"bridge"} + + @pytest.mark.timeout(5) + @pytest.mark.asyncio() + async def test_failure_of_a_cancelled_callers_event_reaches_the_caller_running_the_loop( + self, + ): + sm = gated(khazad_dum(StateMachine, "on_falling"))() + holder, (retreat, gives_up, flee) = await cross_with_callers_queued( + sm, ("retreat", {}), ("cross", {}), ("flee", {}), awake=False + ) + await cancel_and_wait(gives_up) + + with pytest.raises(ValueError, match="on_falling failed"): + await holder + assert await retreat is None + with pytest.raises(ValueError, match="on_falling failed"): + await flee + + @pytest.mark.timeout(5) + @pytest.mark.asyncio() + async def test_failure_after_the_lazy_initial_activation_reaches_every_caller(self): + sm = FlickeringLamp() + results = await asyncio.gather( + sm.send("switch"), sm.send("switch"), return_exceptions=True + ) + + assert [str(result) for result in results] == ["short circuit", "short circuit"] + assert sm.configuration_values == {"off"} + + @pytest.mark.timeout(5) + @pytest.mark.asyncio() + async def test_caller_running_the_loop_retrieves_its_rejected_future(self): + sm = FlickeringLamp() + trigger = sm.switch.put() + + with pytest.raises(ValueError, match="short circuit"): + await sm._processing_loop(trigger.future) + assert not trigger.future._log_traceback + + +def stalled_road(base=StateMachine): + """``go`` and ``stall`` hold the processing lock until cancelled, ``go`` until ``gate`` + is set first; ``step`` counts the steps taken.""" + + class StalledRoad(base): + start = State(initial=True) + road = State() + + go = start.to(road) + stall = start.to.itself(internal=True) | road.to.itself(internal=True) + step = start.to.itself(internal=True) | road.to.itself(internal=True) + hop = Event(road.to.itself(internal=True), internal=True) + + def __init__(self, *args, **kwargs): + self.gate = asyncio.Event() + self.stalling = asyncio.Event() + self.steps = 0 + self.hops = 0 + super().__init__(*args, **kwargs) + + async def on_go(self): + self.hop.put() + await self.gate.wait() + return "went" + + async def on_stall(self): + self.stalling.set() + await asyncio.Event().wait() + + def on_step(self): + self.steps += 1 + return self.steps + + def on_hop(self): + self.hops += 1 + + return StalledRoad + + +class TestCancelledLockHolder: + """Cancelling the coroutine that runs the processing loop cancels only that caller: a + coroutine awaiting a queued event takes the loop over, and the machine stays usable. + + See: https://github.com/fgmacedo/python-statemachine/issues/690 + """ + + @pytest.mark.timeout(5) + @pytest.mark.asyncio() + async def test_queued_callers_get_their_results(self): + sm = stalled_road()() + await sm.activate_initial_state() + holder = asyncio.ensure_future(sm.send("go")) + await asyncio.sleep(0) + waiters = await queue_behind(sm, ("step", {}), ("step", {"delay": 50}), ("step", {})) + + await cancel_and_wait(holder) + + assert await asyncio.gather(*waiters) == [1, 3, 2] + assert sm.configuration_values == {"start"} + assert sm.hops == 0 + assert not sm._engine._processing.locked() + + @pytest.mark.timeout(5) + @pytest.mark.asyncio() + async def test_machine_takes_new_events_afterwards(self): + sm = stalled_road()() + await sm.activate_initial_state() + holder = asyncio.ensure_future(sm.send("go")) + await asyncio.sleep(0) + + await cancel_and_wait(holder) + + assert not sm._engine._processing.locked() + assert sm._engine.internal_queue.is_empty() + assert await sm.send("step") == 1 + assert sm.hops == 0 + + @pytest.mark.timeout(5) + @pytest.mark.asyncio() + async def test_cancelled_while_waiting_for_a_delayed_event(self): + sm = stalled_road()() + await sm.activate_initial_state() + holder = asyncio.ensure_future(sm.send("step", delay=100)) + await asyncio.sleep(0) + (waiter,) = await queue_behind(sm, ("step", {"delay": 50})) + + await cancel_and_wait(holder) + + assert await waiter == 1 + assert sm.steps == 2 + assert sm._engine.external_queue.is_empty() + assert not sm._engine._processing.locked() + + @pytest.mark.timeout(5) + @pytest.mark.asyncio() + async def test_caller_of_the_interrupted_event_gets_a_state_machine_error(self): + sm = stalled_road()() + await sm.activate_initial_state() + holder = asyncio.ensure_future(sm.send("go")) + await asyncio.sleep(0) + stalled, stepper = await queue_behind(sm, ("stall", {}), ("step", {})) + sm.gate.set() + await sm.stalling.wait() + + await cancel_and_wait(holder) + + with pytest.raises(StateMachineError, match="was cancelled while"): + await stalled + assert await stepper == 1 + assert sm.configuration_values == {"road"} + + @pytest.mark.timeout(5) + @pytest.mark.asyncio() + async def test_cancelled_waiter_leaves_its_event_queued(self): + sm = stalled_road()() + await sm.activate_initial_state() + holder = asyncio.ensure_future(sm.send("go")) + await asyncio.sleep(0) + gives_up, waiter = await queue_behind(sm, ("step", {}), ("step", {})) + + await cancel_and_wait(gives_up) + sm.gate.set() + + assert await holder == "went" + assert await waiter == 2 + assert sm.hops == 1 + + @pytest.mark.timeout(5) + @pytest.mark.asyncio() + async def test_cancelled_initial_activation_hands_the_loop_over(self): + class SlowStart(StateMachine): + booting = State(initial=True) + ready = State(final=True) + + boot = booting.to(ready) + + async def on_enter_booting(self): + await asyncio.Event().wait() + + sm = SlowStart() + activation = asyncio.ensure_future(sm.activate_initial_state()) + await asyncio.sleep(0) + (waiter,) = await queue_behind(sm, ("boot", {})) + + await cancel_and_wait(activation) + + assert await waiter is None + assert sm.configuration_values == {"ready"} + + class TestEventQueueRejectFutures: """Unit tests for EventQueue.reject_futures.""" diff --git a/tests/test_cancel_event.py b/tests/test_cancel_event.py index d7652270..3116461b 100644 --- a/tests/test_cancel_event.py +++ b/tests/test_cancel_event.py @@ -324,7 +324,7 @@ async def test_a_caller_awaiting_a_cancelled_event_gets_none(cancels, clock): for _ in range(cancels): sm.cancel_event("dawn") - await asyncio.sleep(0) + await asyncio.wait([waiting], timeout=1) assert (settled(dawn.future), settled(waiting)) == ("answered with None", "answered with None") clock.now += 1.0