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")