From 656d39d0147eb27eab53eb847b95eee402c1b555 Mon Sep 17 00:00:00 2001 From: Sahil Rajput <90515841+SahilRajputdotexe@users.noreply.github.com> Date: Fri, 2 Oct 2026 11:49:04 +0530 Subject: [PATCH] feat: add event deferral to states A state can list events to defer with State(defer=[...]), and compound states defer for their descendants. An event that no transition handles and that every active atomic state defers is queued instead of dropped, and sent again, in order, once the configuration stops deferring it. Adds on_defer/on_replay callbacks, the deferred_events, pending_deferred_event_ids, is_event_deferred() and clear_deferred_queue() introspection API on StateChart, and an opt-in validate_deferred_events check. Both the sync and async engines support it. Signed-off-by: Sahil Rajput <90515841+SahilRajputdotexe@users.noreply.github.com> --- docs/actions.md | 2 + docs/deferral.md | 205 +++++++++++++ docs/index.md | 1 + docs/releases/3.3.0.md | 40 +++ docs/releases/index.md | 1 + docs/states.md | 3 + docs/validations.md | 26 +- statemachine/callbacks.py | 2 + statemachine/engines/async_.py | 35 ++- statemachine/engines/base.py | 76 +++++ statemachine/engines/sync.py | 4 +- statemachine/factory.py | 25 ++ statemachine/state.py | 39 +++ statemachine/statemachine.py | 22 ++ tests/test_statechart_deferral.py | 487 ++++++++++++++++++++++++++++++ 15 files changed, 962 insertions(+), 6 deletions(-) create mode 100644 docs/deferral.md create mode 100644 docs/releases/3.3.0.md create mode 100644 tests/test_statechart_deferral.py diff --git a/docs/actions.md b/docs/actions.md index e3c32137..b04aa4bf 100644 --- a/docs/actions.md +++ b/docs/actions.md @@ -529,6 +529,8 @@ The full list of generic callbacks: | `on_enter_state()` | Enter | Runs when entering any state. | | `on_invoke_state()` | Invoke | Runs when spawning invoke handlers for any state. See {ref}`invoke`. | | `after_transition()` | After | Runs after all state changes. | +| `on_defer_event()` | Defer | Runs when a state defers an event. See {ref}`deferral`. | +| `on_replay_event()` | Replay | Runs when a deferred event is sent again. See {ref}`deferral`. | ```{note} `prepare_event()` is also a generic callback, but it serves a special purpose — diff --git a/docs/deferral.md b/docs/deferral.md new file mode 100644 index 00000000..5812680e --- /dev/null +++ b/docs/deferral.md @@ -0,0 +1,205 @@ +(deferral)= + +# Event deferral + +```{versionadded} 3.3.0 +``` + +```{seealso} +See {ref}`processing-model` for how events are queued and consumed, and {ref}`events` +for how they are declared and sent. +``` + +By default, an event that no active transition handles is dropped (or rejected, depending on +`allow_event_without_transition`). Sometimes the event is not wrong, only early: the +machine just is not ready for it yet. A state can **defer** such events. They are kept in a +queue instead of being discarded, and sent again as soon as the machine reaches a +configuration that no longer defers them. + +This is the UML statechart notion of a *deferred event*. + +## Deferring events in a state + +Pass the ids of the events to the `defer` parameter of {ref}`State`: + +```py +>>> from statemachine import State, StateChart + +>>> class Council(StateChart): +... gathering = State(initial=True, defer=["speak"]) +... deliberating = State() +... decided = State(final=True) +... +... convene = gathering.to(deliberating) +... speak = deliberating.to(decided) + +>>> sm = Council() +>>> sm.send("speak") +>>> sm.configuration_values +OrderedSet(['gathering']) + +>>> sm.pending_deferred_event_ids +['speak'] + +``` + +The `speak` event arrived while the council was still `gathering`, which has no transition for it +and defers it. Nothing happened, but the event was remembered. Once the machine leaves +`gathering`, the queued event is put back at the end of the external queue and processed +like any other: + +```py +>>> sm.send("convene") +>>> sm.configuration_values +OrderedSet(['decided']) + +>>> sm.pending_deferred_event_ids +[] + +``` + +Some details worth knowing: + +- An event is only deferred when **no transition is enabled** for it. If the active state + has a transition for the event, the transition runs as usual. +- Deferred events are replayed in the order they were sent (first in, first out), with the + arguments they were sent with. +- An event that is replayed into a configuration that still defers it stays in the queue. +- Deferral takes precedence over `allow_event_without_transition`: a deferred event is queued + whether the machine would otherwise ignore it or raise `TransitionNotAllowed`. +- The `defer` list does not need to name events declared in the class (see + {ref}`validate_deferred_events `). +- A final state cannot defer events, since the machine does not process events after it + terminates. The queue is cleared at that point. + +## Deferral in compound and parallel states + +A compound state defers events for all of its descendants. Use the `defer` class keyword: + +```py +>>> class Journey(StateChart): +... class mordor(State.Compound, defer=["rest"]): +... gates = State(initial=True) +... mount_doom = State() +... +... climb = gates.to(mount_doom) +... +... shire = State() +... done = State(final=True) +... +... return_home = mordor.to(shire) +... rest = shire.to(done) + +>>> sm = Journey() +>>> sm.send("rest") +>>> sm.send("climb") +>>> sm.pending_deferred_event_ids +['rest'] + +>>> sm.send("return_home") +>>> sm.configuration_values +OrderedSet(['done']) + +``` + +A state defers an event if it lists the event or any of its ancestors does. +`State.deferred_events` holds what a state declares itself, and +`State.deferred_events_recursive` adds the ids declared by its ancestors. + +In a parallel state there are several active atomic states at once, and the event is deferred +**only if all of them defer it**. If any region can still react to the event, it is processed +as usual and nothing is queued. + +## Callbacks + +Two optional callbacks follow the same pattern as `on_enter` and `on_exit`: + +| Callback | When | +|---|---| +| `on_defer` | An event was added to the deferred queue. | +| `on_replay` | An event left the queue to be processed again. | + +Both can be given inline to the state, or found by naming convention: `on_defer_event()` and +`on_replay_event()` for every state, `on_defer_()` and `on_replay_()` for a +specific one. + +```py +>>> class Messenger(StateChart): +... riding = State(initial=True, defer=["deliver"]) +... arrived = State() +... delivered = State(final=True) +... +... arrive = riding.to(arrived) +... deliver = arrived.to(delivered) +... +... def on_defer_riding(self, event): +... print(f"{event} has to wait") +... +... def on_replay_event(self, event, state): +... print(f"{event} is back while in {state.id}") + +>>> sm = Messenger() +>>> sm.send("deliver") +deliver has to wait + +>>> sm.send("arrive") +deliver is back while in arrived + +``` + +The callbacks run once for each active state that declares them, and receive the usual +{ref}`dependency injection ` arguments: `machine`, `model`, `event`, +`state` and `source` (the same state) and `target`, which is always `None`. Arguments passed +to `send()` are available too. With the async engine the callbacks can be coroutines. + +## Inspecting the queue + +The {ref}`StateChart` instance exposes the queue: + +- `deferred_events`: a copy of the queue, as a list of `TriggerData`. +- `pending_deferred_event_ids`: the ids of the queued events, in order. +- `is_event_deferred(event_id)`: whether an event would be deferred in the current + configuration. +- `clear_deferred_queue()`: discards the queued events without replaying them. + +```py +>>> sm = Council() +>>> sm.is_event_deferred("speak") +True + +>>> sm.send("speak") +>>> [str(trigger.event) for trigger in sm.deferred_events] +['speak'] + +>>> sm.clear_deferred_queue() +>>> sm.send("convene") +>>> sm.configuration_values +OrderedSet(['deliberating']) + +``` + +The queue is runtime state: it is not part of the serialized machine, so a pickled or copied +machine starts with an empty queue. + +(validating-deferred-events)= + +## Validating deferred events + +A typo in a `defer` list is silent by default, because the event might be sent by code that is +not declared in the class. Set `validate_deferred_events = True` to require every deferred id +to be a declared event: + +```py +>>> from statemachine.exceptions import InvalidDefinition + +>>> try: +... class Lookout(StateChart): +... validate_deferred_events = True +... watching = State(initial=True, defer=["relieved"]) +... relieved = State(final=True) +... relieve = watching.to(relieved) +... except InvalidDefinition as e: +... print(e) +State 'watching' defers unknown event(s): ['relieved']. Declared events are: ['relieve'] + +``` diff --git a/docs/index.md b/docs/index.md index 06a69c73..f2c71c43 100644 --- a/docs/index.md +++ b/docs/index.md @@ -32,6 +32,7 @@ guards statechart processing_model +deferral error_handling async listeners diff --git a/docs/releases/3.3.0.md b/docs/releases/3.3.0.md new file mode 100644 index 00000000..3729a48e --- /dev/null +++ b/docs/releases/3.3.0.md @@ -0,0 +1,40 @@ +# StateChart 3.3.0 + +*Not released yet* + +## What's new in 3.3.0 + +### Event deferral + +A state can now **defer** events: when an event arrives and no transition handles it, the +engine keeps it in a queue instead of discarding it, and sends it again once the machine +reaches a configuration that no longer defers it. This is the UML statechart *deferred event*. + +```py +>>> from statemachine import State, StateChart + +>>> class Council(StateChart): +... gathering = State(initial=True, defer=["speak"]) +... deliberating = State() +... decided = State(final=True) +... +... convene = gathering.to(deliberating) +... speak = deliberating.to(decided) + +>>> sm = Council() +>>> sm.send("speak") +>>> sm.pending_deferred_event_ids +['speak'] + +>>> sm.send("convene") +>>> sm.configuration_values +OrderedSet(['decided']) + +``` + +Compound states defer events for their children (`class mordor(State.Compound, defer=["rest"])`), +and in parallel states an event is deferred only when every active atomic state defers it. +States also accept `on_defer` and `on_replay` callbacks, `StateChart` exposes +`deferred_events`, `pending_deferred_event_ids`, `is_event_deferred()` and +`clear_deferred_queue()`, and the opt-in `validate_deferred_events` flag checks the deferred +ids against the declared events. See {ref}`deferral`. diff --git a/docs/releases/index.md b/docs/releases/index.md index f62455f7..e93e0780 100644 --- a/docs/releases/index.md +++ b/docs/releases/index.md @@ -16,6 +16,7 @@ Requires Python 3.10+. ```{toctree} :maxdepth: 2 +3.3.0 3.2.2 3.2.1 3.2.0 diff --git a/docs/states.md b/docs/states.md index dba8633c..1bf109be 100644 --- a/docs/states.md +++ b/docs/states.md @@ -41,6 +41,9 @@ True | `enter` | `None` | Callback(s) to run when entering this state. See {ref}`state-actions`. | | `exit` | `None` | Callback(s) to run when leaving this state. See {ref}`state-actions`. | | `invoke` | `None` | Background work spawned on entry, cancelled on exit. See {ref}`invoke-actions`. | +| `defer` | `None` | Ids of events to keep in a queue, instead of discarding them, while this state is active. See {ref}`deferral`. | +| `on_defer` | `None` | Callback(s) to run when an event is deferred by this state. See {ref}`deferral`. | +| `on_replay` | `None` | Callback(s) to run when a deferred event is sent again. See {ref}`deferral`. | ```py >>> class CampaignMachine(StateChart): diff --git a/docs/validations.md b/docs/validations.md index 7bac5402..3af6d034 100644 --- a/docs/validations.md +++ b/docs/validations.md @@ -183,6 +183,26 @@ The `donedata` parameter can only be used on states marked as `final=True`: ``` +### Deferred events + +A final state cannot defer events, since a terminated machine processes none: + +```py +>>> try: +... class Bad(StateChart): +... a = State(initial=True) +... b = State(final=True, defer=["go"]) +... go = a.to(b) +... except InvalidDefinition as e: +... print(e) +Cannot defer events on final states. + +``` + +Optionally, set `validate_deferred_events = True` to also require every deferred event id to +be a declared event. See {ref}`validating-deferred-events`. + + ### Invalid listener entries Entries in the `listeners` class attribute must be classes, callables, or @@ -279,9 +299,11 @@ Expressions support `and`, `or`, `not`, and parentheses. See | Internal transition targets | Class definition| No | | Initial transitions have no cond | Class definition| No | | `donedata` on final states only | Class definition| No | +| Final states do not defer events | Class definition| No | +| Deferred events are declared | Class definition| `validate_deferred_events` (off by default) | | Invalid listener entries | Class definition| No | | Callback resolution | Instance creation | No | | Boolean expression parsing | Instance creation | No | -All configurable flags default to `True`. Set them to `False` on the class -to disable the corresponding check. +All configurable flags default to `True`, except `validate_deferred_events`, which is opt-in. +Set a flag to `False` on the class to disable the corresponding check. diff --git a/statemachine/callbacks.py b/statemachine/callbacks.py index e7c6da1a..30545667 100644 --- a/statemachine/callbacks.py +++ b/statemachine/callbacks.py @@ -41,6 +41,8 @@ class CallbackGroup(IntEnum): ENTER = auto() EXIT = auto() INVOKE = auto() + DEFER = auto() + REPLAY = auto() VALIDATOR = auto() BEFORE = auto() ON = auto() diff --git a/statemachine/engines/async_.py b/statemachine/engines/async_.py index 0cd6c4a8..75c803a0 100644 --- a/statemachine/engines/async_.py +++ b/statemachine/engines/async_.py @@ -279,8 +279,34 @@ async def _enter_states( # noqa: C901 if target.final: self._handle_final_state(target, on_entry_result) + await self._replay_deferred_events() return result + async def _defer_event(self, trigger_data: TriggerData): + self._debug("%s Deferring event '%s'", self._log_id, trigger_data.event) + self._deferred_queue.append(trigger_data) + on_error = self._on_error_handler() + for state in self.sm.configuration: + if state.on_defer.key in self.sm._callbacks: + await self.sm._callbacks.async_call( + state.on_defer.key, + *trigger_data.args, + on_error=on_error, + **self._deferral_callback_kwargs(trigger_data, state), + ) + + async def _replay_deferred_events(self): + on_error = self._on_error_handler() + for trigger_data in self._release_deferred_events(): + for state in self.sm.configuration: + if state.on_replay.key in self.sm._callbacks: + await self.sm._callbacks.async_call( + state.on_replay.key, + *trigger_data.args, + on_error=on_error, + **self._deferral_callback_kwargs(trigger_data, state), + ) + async def microstep(self, transitions: "list[Transition]", trigger_data: TriggerData): self._microstep_count += 1 self._debug( @@ -471,15 +497,18 @@ async def processing_loop( # noqa: C901 if first_result is self._sentinel: first_result = result else: - if not self.sm.allow_event_without_transition: + if self._is_event_deferred(self._event_id(external_event)): + await self._defer_event(external_event) + self._resolve_future(event_future, None) + elif not self.sm.allow_event_without_transition: tna = TransitionNotAllowed( external_event.event, self.sm.configuration ) self._reject_future(event_future, tna) self._reject_pending_futures(tna) raise tna - # Event allowed but no transition — resolve with None - self._resolve_future(event_future, None) + else: + self._resolve_future(event_future, None) except Exception as exc: self._reject_future(event_future, exc) self._reject_pending_futures(exc) diff --git a/statemachine/engines/base.py b/statemachine/engines/base.py index 14ae301d..c5535e53 100644 --- a/statemachine/engines/base.py +++ b/statemachine/engines/base.py @@ -86,6 +86,7 @@ def __init__(self, sm: "StateChart"): self.sm: "StateChart" = sm self.external_queue = EventQueue() self.internal_queue = EventQueue() + self._deferred_queue: list[TriggerData] = [] self._sentinel = object() self.running = True self._processing = Lock() @@ -584,10 +585,84 @@ def _add_state_to_configuration(self, target: State): if not self.sm.atomic_configuration_update: self.sm._config.add(target) + def _is_event_deferred(self, event_id: str) -> bool: + atomic_states = [s for s in self.sm.configuration if s.is_atomic] + return bool(atomic_states) and all( + event_id in s.deferred_events_recursive for s in atomic_states + ) + + @staticmethod + def _event_id(trigger_data: TriggerData) -> str: + return str(trigger_data.event) if trigger_data.event else "" + + def _deferral_callback_kwargs(self, trigger_data: TriggerData, state: State) -> dict: + kwargs = trigger_data.kwargs.copy() + kwargs.update( + { + "machine": self.sm, + "model": self.sm.model, + "event": trigger_data.event, + "state": state, + "source": state, + "target": None, + } + ) + return kwargs + + def _defer_event(self, trigger_data: TriggerData): + self._debug("%s Deferring event '%s'", self._log_id, trigger_data.event) + self._deferred_queue.append(trigger_data) + on_error = self._on_error_handler() + for state in self.sm.configuration: + if state.on_defer.key in self.sm._callbacks: + self.sm._callbacks.call( + state.on_defer.key, + *trigger_data.args, + on_error=on_error, + **self._deferral_callback_kwargs(trigger_data, state), + ) + + def _release_deferred_events(self) -> list[TriggerData]: + """Move the deferred events that are no longer deferred back to the external queue. + + Returns the released events in FIFO order. + """ + still_deferred: list[TriggerData] = [] + released: list[TriggerData] = [] + for trigger_data in self._deferred_queue: + if self._is_event_deferred(self._event_id(trigger_data)): + still_deferred.append(trigger_data) + else: + released.append(trigger_data) + self._deferred_queue = still_deferred + for trigger_data in released: + self._debug("%s Replaying deferred event '%s'", self._log_id, trigger_data.event) + self.external_queue.put(trigger_data) + return released + + def _replay_deferred_events(self): + on_error = self._on_error_handler() + for trigger_data in self._release_deferred_events(): + for state in self.sm.configuration: + if state.on_replay.key in self.sm._callbacks: + self.sm._callbacks.call( + state.on_replay.key, + *trigger_data.args, + on_error=on_error, + **self._deferral_callback_kwargs(trigger_data, state), + ) + + def pending_deferred_event_ids(self) -> list[str]: + return [self._event_id(td) for td in self._deferred_queue] + + def clear_deferred_queue(self): + self._deferred_queue.clear() + def stop(self): """Stop this engine externally (e.g. when a parent cancels a child invocation).""" self._debug("%s Stopping engine", self._log_id) self.running = False + self._deferred_queue.clear() try: self._invoke_manager.cancel_all() except Exception: # pragma: no cover @@ -718,6 +793,7 @@ def _enter_states( # noqa: C901 if target.final: self._handle_final_state(target, on_entry_result) + self._replay_deferred_events() return result def compute_entry_set( diff --git a/statemachine/engines/sync.py b/statemachine/engines/sync.py index 627b51ae..6f590a51 100644 --- a/statemachine/engines/sync.py +++ b/statemachine/engines/sync.py @@ -167,7 +167,9 @@ def processing_loop(self, caller_future=None): # noqa: C901 raise else: - if not self.sm.allow_event_without_transition: + if self._is_event_deferred(self._event_id(external_event)): + self._defer_event(external_event) + elif not self.sm.allow_event_without_transition: raise TransitionNotAllowed(external_event.event, self.sm.configuration) finally: diff --git a/statemachine/factory.py b/statemachine/factory.py index 33f66e28..57217c8b 100644 --- a/statemachine/factory.py +++ b/statemachine/factory.py @@ -32,6 +32,10 @@ class StateMachineMetaclass(type): """If ``True`` and final states exist, non-final states without a path to any final state raise ``InvalidDefinition``.""" + validate_deferred_events: bool = False + """If ``True``, deferred event IDs on states are validated against declared + events at class construction time. Unknown event IDs raise ``InvalidDefinition``.""" + def __init__( cls, name: str, @@ -182,6 +186,7 @@ def _check(cls): cls._check_initial_state() cls._check_final_states() + cls._check_deferred_events() cls._check_disconnected_state() cls._check_trap_states() cls._check_reachable_final_states() @@ -211,6 +216,26 @@ def _check_final_states(cls): ) ) + def _check_deferred_events(cls): + if not cls.validate_deferred_events: + return + known_event_ids = {str(e.id) for e in cls._events} + for state in iterate_states(cls.states): + if not state.deferred_events: + continue + unknown = set(state.deferred_events) - known_event_ids + if unknown: + raise InvalidDefinition( + _( + "State '{state}' defers unknown event(s): {events}. " + "Declared events are: {known}" + ).format( + state=state.id, + events=sorted(unknown), + known=sorted(known_event_ids), + ) + ) + def _check_trap_states(cls): if not cls.validate_trap_states: return diff --git a/statemachine/state.py b/statemachine/state.py index 02268083..e7701064 100644 --- a/statemachine/state.py +++ b/statemachine/state.py @@ -157,6 +157,13 @@ class State: See :ref:`actions`. exit: One or more callbacks assigned to be executed when the state is exited. See :ref:`actions`. + defer: Ids of the events this state keeps in a queue, instead of discarding them, when + no transition handles them. They are sent again once the machine reaches a + configuration that no longer defers them. Not allowed on final states. + See :ref:`deferral`. + on_defer: One or more callbacks assigned to be executed when an event is deferred. + on_replay: One or more callbacks assigned to be executed when a deferred event is sent + again. State is a core component on how this library implements an expressive API to declare StateMachines. @@ -237,6 +244,9 @@ def __init__( exit: Any = None, invoke: Any = None, donedata: Any = None, + defer: "list[str] | None" = None, + on_defer: Any = None, + on_replay: Any = None, _callbacks: Any = None, ): self.name = name @@ -266,6 +276,15 @@ def __init__( if not final: raise InvalidDefinition(_("'donedata' can only be specified on final states.")) self.enter.add(donedata, priority=CallbackPriority.INLINE) + if defer and final: + raise InvalidDefinition(_("Cannot defer events on final states.")) + self._deferred_events: frozenset[str] = frozenset(defer or []) + self.on_defer = self._specs.grouper(CallbackGroup.DEFER).add( + on_defer, priority=CallbackPriority.INLINE + ) + self.on_replay = self._specs.grouper(CallbackGroup.REPLAY).add( + on_replay, priority=CallbackPriority.INLINE + ) self.document_order = 0 self._hash = id(self) self._init_states() @@ -300,6 +319,16 @@ def _setup(self): self.invoke.add( f"on_invoke_{self.id}", priority=CallbackPriority.NAMING, is_convention=True ) + self.on_defer.add("on_defer_event", priority=CallbackPriority.GENERIC, is_convention=True) + self.on_defer.add( + f"on_defer_{self.id}", priority=CallbackPriority.NAMING, is_convention=True + ) + self.on_replay.add( + "on_replay_event", priority=CallbackPriority.GENERIC, is_convention=True + ) + self.on_replay.add( + f"on_replay_{self.id}", priority=CallbackPriority.NAMING, is_convention=True + ) def _on_event_defined(self, event: str, transition: Transition, states: list["State"]): """Called by statemachine factory when an event is defined having a transition @@ -352,6 +381,11 @@ def final(self): def parallel(self): return self._parallel + @property + def deferred_events(self) -> frozenset[str]: + """Event ids this state defers while it is active.""" + return self._deferred_events + @property def is_compound(self): return bool(self.states) and not self.parallel @@ -371,6 +405,11 @@ def ancestors(self, parent: "State | None" = None) -> Generator["State", None, N def is_descendant(self, state: "State") -> bool: return state in self.ancestors() + @property + def deferred_events_recursive(self) -> frozenset[str]: + """Event ids deferred by this state and by all of its ancestors.""" + return self._deferred_events.union(*(s._deferred_events for s in self.ancestors())) + class InstanceState(State): """Per-instance proxy for a State, delegating attribute access to the underlying State. diff --git a/statemachine/statemachine.py b/statemachine/statemachine.py index 5277c4fc..0e150555 100644 --- a/statemachine/statemachine.py +++ b/statemachine/statemachine.py @@ -510,6 +510,28 @@ def cancel_event(self, send_id: str): """Cancel all the delayed events with the given ``send_id``.""" self._engine.cancel_event(send_id) + @property + def deferred_events(self) -> "list[TriggerData]": + """A copy of the queue of events deferred by the current configuration, in FIFO order.""" + return list(self._engine._deferred_queue) + + @property + def pending_deferred_event_ids(self) -> "list[str]": + """The ids of the events in the deferred queue, in FIFO order.""" + return list(self._engine.pending_deferred_event_ids()) + + def is_event_deferred(self, event_id: str) -> bool: + """Whether an event with the given id would be deferred right now. + + It is only deferred when every atomic state of the active configuration defers it, + either itself or through one of its ancestors. + """ + return bool(self._engine._is_event_deferred(event_id)) + + def clear_deferred_queue(self) -> None: + """Discard the deferred events without replaying them.""" + self._engine.clear_deferred_queue() + @property def is_terminated(self): """Whether the state machine has reached a final state. diff --git a/tests/test_statechart_deferral.py b/tests/test_statechart_deferral.py new file mode 100644 index 00000000..6bcdace7 --- /dev/null +++ b/tests/test_statechart_deferral.py @@ -0,0 +1,487 @@ +"""Event deferral. + +Tests exercise states that defer events: the events are kept instead of discarded, replayed +in order once the machine reaches a configuration that no longer defers them, and visible +through the introspection API. Deferral is inherited from ancestors and, in parallel regions, +requires every active atomic state to agree. + +Theme: The Council of Elrond and the Fellowship's journey. +""" + +import pickle + +import pytest +from statemachine.event_data import TriggerData +from statemachine.exceptions import InvalidDefinition +from statemachine.exceptions import TransitionNotAllowed + +from statemachine import State +from statemachine import StateChart +from statemachine import StateMachine + + +class Council(StateChart): + gathering = State(initial=True, defer=["speak"]) + deliberating = State() + decided = State(final=True) + + convene = gathering.to(deliberating) + speak = deliberating.to(decided) + + +class Road(StateChart): + shire = State(initial=True, defer=["ambush"]) + bree = State(defer=["ambush"]) + weathertop = State() + rivendell = State(final=True) + + to_bree = shire.to(bree) + to_weathertop = bree.to(weathertop) + ambush = weathertop.to(rivendell) + + +class TestStateParameters: + def test_defer_is_stored_as_frozenset(self): + state = State(defer=["speak", "listen"]) + assert state.deferred_events == frozenset({"speak", "listen"}) + + def test_nothing_is_deferred_by_default(self): + assert State().deferred_events == frozenset() + + def test_final_state_cannot_defer(self): + with pytest.raises(InvalidDefinition, match="final"): + State(final=True, defer=["speak"]) + + def test_recursive_includes_every_ancestor(self): + grandparent = State(defer=["a"]) + parent = State(defer=["b", "a"]) + child = State(defer=["c"]) + parent.parent = grandparent + child.parent = parent + + assert child.deferred_events_recursive == frozenset({"a", "b", "c"}) + assert parent.deferred_events_recursive == frozenset({"a", "b"}) + assert State().deferred_events_recursive == frozenset() + + +@pytest.mark.timeout(5) +class TestDeferralAndReplay: + async def test_deferred_event_is_replayed_after_leaving_the_state(self, sm_runner): + sm = await sm_runner.start(Council) + + await sm_runner.send(sm, "speak") + assert "gathering" in sm.configuration_values + assert sm.pending_deferred_event_ids == ["speak"] + + await sm_runner.send(sm, "convene") + assert "decided" in sm.configuration_values + assert sm.deferred_events == [] + + async def test_event_stays_queued_while_the_next_state_also_defers_it(self, sm_runner): + sm = await sm_runner.start(Road) + await sm_runner.send(sm, "ambush") + + await sm_runner.send(sm, "to_bree") + assert "bree" in sm.configuration_values + assert sm.pending_deferred_event_ids == ["ambush"] + + await sm_runner.send(sm, "to_weathertop") + assert "rivendell" in sm.configuration_values + assert sm.deferred_events == [] + + async def test_replay_keeps_the_order_the_events_were_sent(self, sm_runner): + entered = [] + + class Watch(StateChart): + resting = State(initial=True, defer=["first_watch", "second_watch"]) + first = State() + second = State() + dawn = State(final=True) + + wake = resting.to(first) + first_watch = first.to(second) + second_watch = second.to(dawn) + + def on_enter_state(self, target): + entered.append(target.id) + + sm = await sm_runner.start(Watch) + await sm_runner.send(sm, "first_watch") + await sm_runner.send(sm, "second_watch") + assert sm.pending_deferred_event_ids == ["first_watch", "second_watch"] + + await sm_runner.send(sm, "wake") + assert "dawn" in sm.configuration_values + assert entered == ["resting", "first", "second", "dawn"] + + async def test_replayed_event_keeps_its_arguments(self, sm_runner): + received = [] + + class Messenger(StateChart): + riding = State(initial=True, defer=["deliver"]) + arrived = State() + delivered = State(final=True) + + arrive = riding.to(arrived) + deliver = arrived.to(delivered) + + def on_deliver(self, message=None, urgent=False): + received.append((message, urgent)) + + sm = await sm_runner.start(Messenger) + await sm_runner.send(sm, "deliver", message="the ring", urgent=True) + await sm_runner.send(sm, "arrive") + + assert received == [("the ring", True)] + + async def test_event_with_an_enabled_transition_is_not_deferred(self, sm_runner): + class Gate(StateChart): + closed = State(initial=True, defer=["knock"]) + open = State(final=True) + + knock = closed.to(open) + + sm = await sm_runner.start(Gate) + await sm_runner.send(sm, "knock") + + assert "open" in sm.configuration_values + assert sm.deferred_events == [] + + async def test_events_not_listed_in_defer_are_not_queued(self, sm_runner): + sm = await sm_runner.start(Council) + await sm_runner.send(sm, "unknown") + + assert sm.deferred_events == [] + + async def test_deferral_does_not_require_a_declared_event(self, sm_runner): + class Lookout(StateChart): + watching = State(initial=True, defer=["not_declared"]) + relieved = State(final=True) + + relieve = watching.to(relieved) + + sm = await sm_runner.start(Lookout) + await sm_runner.send(sm, "not_declared") + + assert sm.pending_deferred_event_ids == ["not_declared"] + + async def test_queue_is_cleared_when_the_machine_terminates(self, sm_runner): + class Hobbit(StateChart): + hiding = State(initial=True, defer=["call"]) + gone = State(final=True) + + leave = hiding.to(gone) + + sm = await sm_runner.start(Hobbit) + await sm_runner.send(sm, "call") + assert len(sm.deferred_events) == 1 + + await sm_runner.send(sm, "leave") + assert sm.is_terminated + assert sm.deferred_events == [] + + +@pytest.mark.timeout(5) +class TestDeferralPrecedence: + async def test_deferral_wins_over_allow_event_without_transition(self, sm_runner): + class Tolerant(Council): + allow_event_without_transition = True + + sm = await sm_runner.start(Tolerant) + await sm_runner.send(sm, "speak") + assert sm.pending_deferred_event_ids == ["speak"] + + await sm_runner.send(sm, "convene") + assert "decided" in sm.configuration_values + + async def test_deferral_wins_over_transition_not_allowed(self, sm_runner): + class Strict(StateMachine): + allow_event_without_transition = False + + gathering = State(initial=True, defer=["speak"]) + deliberating = State() + decided = State(final=True) + + convene = gathering.to(deliberating) + speak = deliberating.to(decided) + + sm = await sm_runner.start(Strict) + await sm_runner.send(sm, "speak") + assert sm.pending_deferred_event_ids == ["speak"] + + await sm_runner.send(sm, "convene") + assert "decided" in sm.configuration_values + + async def test_other_events_are_still_rejected(self, sm_runner): + class Strict(StateMachine): + allow_event_without_transition = False + + gathering = State(initial=True, defer=["speak"]) + decided = State(final=True) + + finish = gathering.to(decided) + + sm = await sm_runner.start(Strict) + with pytest.raises(TransitionNotAllowed): + await sm_runner.send(sm, "unknown") + assert sm.deferred_events == [] + + +@pytest.mark.timeout(5) +class TestDeferralInHierarchy: + async def test_children_inherit_the_deferral_of_a_compound_state(self, sm_runner): + class Journey(StateChart): + class mordor(State.Compound, defer=["rest"]): + gates = State(initial=True) + mount_doom = State() + + climb = gates.to(mount_doom) + + shire = State() + done = State(final=True) + + return_home = mordor.to(shire) + rest = shire.to(done) + + sm = await sm_runner.start(Journey) + await sm_runner.send(sm, "rest") + await sm_runner.send(sm, "climb") + assert sm.pending_deferred_event_ids == ["rest"] + + await sm_runner.send(sm, "return_home") + assert "done" in sm.configuration_values + + async def test_parallel_regions_must_all_defer_the_event(self, sm_runner): + class Fellowship(StateChart): + class company(State.Parallel): + class ringbearer(State.Compound): + walking = State(initial=True, defer=["rest"]) + + class guard(State.Compound): + watching = State(initial=True) + + camp = State(final=True) + + make_camp = company.to(camp) + + sm = await sm_runner.start(Fellowship) + assert not sm.is_event_deferred("rest") + + await sm_runner.send(sm, "rest") + assert sm.deferred_events == [] + + async def test_parallel_regions_that_all_defer_the_event_queue_it(self, sm_runner): + class Fellowship(StateChart): + class company(State.Parallel): + class ringbearer(State.Compound): + walking = State(initial=True, defer=["rest"]) + + class guard(State.Compound): + watching = State(initial=True, defer=["rest"]) + + camp = State() + asleep = State(final=True) + + make_camp = company.to(camp) + rest = camp.to(asleep) + + sm = await sm_runner.start(Fellowship) + await sm_runner.send(sm, "rest") + assert sm.pending_deferred_event_ids == ["rest"] + + await sm_runner.send(sm, "make_camp") + assert "asleep" in sm.configuration_values + + +@pytest.mark.timeout(5) +class TestIntrospection: + async def test_is_event_deferred_follows_the_configuration(self, sm_runner): + sm = await sm_runner.start(Council) + assert sm.is_event_deferred("speak") + assert not sm.is_event_deferred("convene") + + await sm_runner.send(sm, "convene") + assert not sm.is_event_deferred("speak") + + async def test_deferred_events_returns_a_copy_of_trigger_data(self, sm_runner): + sm = await sm_runner.start(Council) + await sm_runner.send(sm, "speak") + + events = sm.deferred_events + assert [type(item) for item in events] == [TriggerData] + assert events[0].event == "speak" + + events.clear() + assert len(sm.deferred_events) == 1 + + async def test_pending_ids_are_listed_in_fifo_order(self, sm_runner): + class Watch(StateChart): + resting = State(initial=True, defer=["a", "b"]) + moving = State(final=True) + + go = resting.to(moving) + + sm = await sm_runner.start(Watch) + assert sm.pending_deferred_event_ids == [] + + await sm_runner.send(sm, "b") + await sm_runner.send(sm, "a") + assert sm.pending_deferred_event_ids == ["b", "a"] + + async def test_cleared_events_are_not_replayed(self, sm_runner): + sm = await sm_runner.start(Council) + await sm_runner.send(sm, "speak") + + sm.clear_deferred_queue() + assert sm.deferred_events == [] + + await sm_runner.send(sm, "convene") + assert "deliberating" in sm.configuration_values + + def test_pickle_does_not_carry_the_queue(self): + sm = Council() + sm.send("speak") + assert len(sm.deferred_events) == 1 + + restored = pickle.loads(pickle.dumps(sm)) + assert restored.deferred_events == [] + assert "gathering" in restored.configuration_values + + +@pytest.mark.timeout(5) +class TestCallbacks: + async def test_inline_callbacks(self, sm_runner): + calls = [] + + def deferred(event, state, source, target, machine, model): + calls.append(("defer", str(event), state.id, source.id, target)) + assert machine is sm + assert model is sm.model + + def replayed(event, state): + calls.append(("replay", str(event), state.id)) + + class Messenger(StateChart): + riding = State(initial=True, defer=["deliver"], on_defer=deferred) + arrived = State(on_replay=replayed) + delivered = State(final=True) + + arrive = riding.to(arrived) + deliver = arrived.to(delivered) + + sm = await sm_runner.start(Messenger) + await sm_runner.send(sm, "deliver") + await sm_runner.send(sm, "arrive") + + assert calls == [ + ("defer", "deliver", "riding", "riding", None), + ("replay", "deliver", "arrived"), + ] + + async def test_naming_conventions(self, sm_runner): + calls = [] + + class Messenger(StateChart): + riding = State(initial=True, defer=["deliver"]) + arrived = State() + delivered = State(final=True) + + arrive = riding.to(arrived) + deliver = arrived.to(delivered) + + def on_defer_event(self, event): + calls.append(f"defer_event {event}") + + def on_defer_riding(self, event): + calls.append(f"defer_riding {event}") + + def on_replay_event(self, event): + calls.append(f"replay_event {event}") + + def on_replay_arrived(self, event): + calls.append(f"replay_arrived {event}") + + sm = await sm_runner.start(Messenger) + await sm_runner.send(sm, "deliver") + await sm_runner.send(sm, "arrive") + + assert calls == [ + "defer_event deliver", + "defer_riding deliver", + "replay_event deliver", + "replay_arrived deliver", + ] + + +@pytest.mark.timeout(5) +class TestAsyncCallbacks: + async def test_coroutine_callbacks_are_awaited(self): + calls = [] + + async def deferred(**kwargs): + calls.append("defer") + + async def replayed(**kwargs): + calls.append("replay") + + class Messenger(StateChart): + riding = State(initial=True, defer=["deliver"], on_defer=deferred) + arrived = State(on_replay=replayed) + delivered = State(final=True) + + arrive = riding.to(arrived) + deliver = arrived.to(delivered) + + sm = Messenger() + await sm.activate_initial_state() + await sm.send("deliver") + await sm.send("arrive") + + assert calls == ["defer", "replay"] + assert "delivered" in sm.configuration_values + + +class TestValidation: + def test_unknown_deferred_event_is_ignored_by_default(self): + class Lookout(StateChart): + watching = State(initial=True, defer=["not_declared"]) + relieved = State(final=True) + + relieve = watching.to(relieved) + + assert Lookout().is_event_deferred("not_declared") + + def test_unknown_deferred_event_is_rejected_when_validating(self): + with pytest.raises(InvalidDefinition, match="unknown event"): + + class Lookout(StateChart): + validate_deferred_events = True + + watching = State(initial=True, defer=["not_declared"]) + relieved = State(final=True) + + relieve = watching.to(relieved) + + def test_nested_states_are_validated_too(self): + with pytest.raises(InvalidDefinition, match="unknown event"): + + class Journey(StateChart): + validate_deferred_events = True + + class mordor(State.Compound, defer=["not_declared"]): + gates = State(initial=True) + + done = State(final=True) + + leave = mordor.to(done) + + def test_declared_events_pass_validation(self): + class Lookout(StateChart): + validate_deferred_events = True + + watching = State(initial=True, defer=["relieve"]) + relieved = State(final=True) + + relieve = watching.to(relieved) + + assert Lookout().is_event_deferred("relieve")