diff --git a/CLAUDE.md b/CLAUDE.md index f07a231..704ede6 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -96,6 +96,10 @@ The critical execution order is: `main_event_loop` → `microstep` → `main_eve query the snapshot with `state.has_tag("loading")` / `state.hasTag(...)` or read the aggregated `state.tags` frozenset. Tags union across the whole active configuration (compound ancestors + parallel regions) and are recomputed from the machine definition, so snapshots stay tag-free +- **`stateIn` guard** (0.7.0, `from xstate import stateIn`) — first-class guard over the current + configuration: `stateIn("#id")`, `stateIn("parent.child")`, or `stateIn({parent: child})`. + Composes with `and_`/`or_`/`not_` and can be registered as a named guard; it reuses the same + matcher as the internal transition `in` guard (`algorithm._matches_in_state`) Handler-signature note: guards/assigners are invoked arity-aware by `algorithm._invoke`, which supports four calling conventions: `()`, `(context)`, diff --git a/src/xstate/__init__.py b/src/xstate/__init__.py index 171bdac..49fd572 100644 --- a/src/xstate/__init__.py +++ b/src/xstate/__init__.py @@ -21,7 +21,7 @@ UnregisteredImplementationError, XStateError, ) -from xstate.guards import and_, not_, or_ # noqa +from xstate.guards import and_, not_, or_, stateIn # noqa from xstate.handlers import HandlerArgs # noqa from xstate.interpreter import Interpreter, interpret # noqa from xstate.machine import Machine # noqa @@ -78,6 +78,8 @@ "and_", "or_", "not_", + # stateIn guard (0.7.0) + "stateIn", # Snapshot serialization (0.6.0) "serialize_snapshot", "deserialize_snapshot", diff --git a/src/xstate/algorithm.py b/src/xstate/algorithm.py index cca2ef2..f8be415 100644 --- a/src/xstate/algorithm.py +++ b/src/xstate/algorithm.py @@ -551,7 +551,7 @@ def condition_match( inner = cond.fn if isinstance(cond, HandlerAdapter) else cond if isinstance(inner, _ComposableGuard): guards = getattr(transition.source.machine, "guards", {}) or {} - if not inner._call(context, event, guards): + if not inner._call(context, event, guards, configuration): return False elif not bool(invoke_handler(cond, context, event, params=params)): return False diff --git a/src/xstate/guards.py b/src/xstate/guards.py index a9913ac..c6e8732 100644 --- a/src/xstate/guards.py +++ b/src/xstate/guards.py @@ -1,26 +1,30 @@ -"""Composable guard combinators (0.6.0). +"""Composable guard combinators (0.6.0) + ``stateIn`` guard (0.7.0). ``and_``, ``or_``, and ``not_`` compose guards from smaller pieces without -writing wrapper lambdas. Sub-guards can be callables or strings that reference -other named guards in the machine's ``guards`` registry. +writing wrapper lambdas. Sub-guards can be callables, strings that reference +other named guards in the machine's ``guards`` registry, or other composable +guards (including :func:`stateIn`). + +``stateIn`` is a guard over the *current configuration* — it passes when the +machine is in the given state — and composes with the combinators above. Usage:: - from xstate import Machine, and_, not_, or_ + from xstate import Machine, and_, not_, or_, stateIn machine = Machine(config, guards={ "isLoggedIn": lambda ctx, evt: ctx.get("logged_in"), "hasPermission": lambda ctx, evt: ctx.get("permission"), - "canGo": and_("isLoggedIn", "hasPermission"), + "canGo": and_("isLoggedIn", "hasPermission", stateIn("#ready")), }) Or via the ``setup()`` builder (recommended):: - from xstate import setup, and_ + from xstate import setup, and_, stateIn machine = setup(guards={ "isLoggedIn": lambda ctx, evt: ctx.get("logged_in"), - "canGo": and_("isLoggedIn", lambda ctx, evt: ctx.get("value") > 0), + "canGo": and_("isLoggedIn", stateIn("ready")), }).create_machine(config) String sub-guard names are resolved lazily from the machine's ``guards`` @@ -33,13 +37,21 @@ class _ComposableGuard: - """Base class for ``and_``, ``or_``, and ``not_`` guard combinators. + """Base class for ``and_``, ``or_``, ``not_``, and ``stateIn`` guards. Instances are callable (``guard(context, event)``) and can be registered - directly in the machine's ``guards`` dict. + directly in the machine's ``guards`` dict. Evaluation threads the active + ``configuration`` through so nested ``stateIn`` sub-guards can see it. """ - def _eval(self, guard: Any, context: Any, event: Any, registry: dict) -> bool: + def _eval( + self, + guard: Any, + context: Any, + event: Any, + registry: dict, + configuration: Any = None, + ) -> bool: if isinstance(guard, str): fn = registry.get(guard) if fn is None: @@ -49,17 +61,23 @@ def _eval(self, guard: Any, context: Any, event: Any, registry: dict) -> bool: ) guard = fn if isinstance(guard, _ComposableGuard): - return guard._call(context, event, registry) + return guard._call(context, event, registry, configuration) from xstate.handlers import invoke_handler return bool(invoke_handler(guard, context, event)) - def _call(self, context: Any, event: Any, registry: dict) -> bool: + def _call( + self, + context: Any, + event: Any, + registry: dict, + configuration: Any = None, + ) -> bool: raise NotImplementedError def __call__(self, context: Any = None, event: Any = None) -> bool: - """Direct call (no registry — string sub-guards must not be used).""" - return self._call(context, event, {}) + """Direct call (no registry/configuration — for standalone use).""" + return self._call(context, event, {}, None) class _AndGuard(_ComposableGuard): @@ -68,8 +86,13 @@ class _AndGuard(_ComposableGuard): def __init__(self, guards: tuple[Any, ...]): self._guards = guards - def _call(self, context: Any, event: Any, registry: dict) -> bool: - return all(self._eval(g, context, event, registry) for g in self._guards) + def _call( + self, context: Any, event: Any, registry: dict, configuration: Any = None + ) -> bool: + return all( + self._eval(g, context, event, registry, configuration) + for g in self._guards + ) class _OrGuard(_ComposableGuard): @@ -78,8 +101,13 @@ class _OrGuard(_ComposableGuard): def __init__(self, guards: tuple[Any, ...]): self._guards = guards - def _call(self, context: Any, event: Any, registry: dict) -> bool: - return any(self._eval(g, context, event, registry) for g in self._guards) + def _call( + self, context: Any, event: Any, registry: dict, configuration: Any = None + ) -> bool: + return any( + self._eval(g, context, event, registry, configuration) + for g in self._guards + ) class _NotGuard(_ComposableGuard): @@ -88,8 +116,32 @@ class _NotGuard(_ComposableGuard): def __init__(self, guard: Any): self._guard = guard - def _call(self, context: Any, event: Any, registry: dict) -> bool: - return not self._eval(self._guard, context, event, registry) + def _call( + self, context: Any, event: Any, registry: dict, configuration: Any = None + ) -> bool: + return not self._eval(self._guard, context, event, registry, configuration) + + +class _StateInGuard(_ComposableGuard): + """Guard that passes when the machine is in the given state (v5 ``stateIn``). + + The *spec* uses the same syntax as a transition ``in`` guard: ``"#id"`` for + an explicit id, a dotted ``"parent.child"`` path, or a ``{parent: child}`` + dict. Evaluation needs the active configuration; when called without one + (e.g. directly, outside a transition) it returns ``False``. + """ + + def __init__(self, spec: Any): + self._spec = spec + + def _call( + self, context: Any, event: Any, registry: dict, configuration: Any = None + ) -> bool: + if configuration is None: + return False + from xstate.algorithm import _matches_in_state + + return _matches_in_state(self._spec, configuration) def and_(*guards: Any) -> _AndGuard: @@ -105,3 +157,12 @@ def or_(*guards: Any) -> _OrGuard: def not_(guard: Any) -> _NotGuard: """Return a guard that passes when *guard* does NOT pass.""" return _NotGuard(guard) + + +def stateIn(spec: Any) -> _StateInGuard: + """Return a guard that passes when the machine is currently in *spec*. + + *spec* is ``"#id"``, a dotted ``"parent.child"`` path, or a ``{parent: child}`` + dict — the same shape accepted by a transition ``in`` guard. + """ + return _StateInGuard(spec) diff --git a/src/xstate/machine.py b/src/xstate/machine.py index 11a7fc6..dcbf2b6 100644 --- a/src/xstate/machine.py +++ b/src/xstate/machine.py @@ -142,7 +142,7 @@ def _get_actions( result.append( self._bind_action( action, - self.actions[action.type], + self.actions[action.type], # type: ignore[index] context, event, ) diff --git a/tests/test_state_in_guard.py b/tests/test_state_in_guard.py new file mode 100644 index 0000000..131b210 --- /dev/null +++ b/tests/test_state_in_guard.py @@ -0,0 +1,209 @@ +"""Tests for the user-facing stateIn() guard (0.7.0). + +`stateIn(spec)` is a composable guard that passes when the machine is currently +in the given state. It reuses the same matching as a transition `in` guard and +composes with and_/or_/not_. +""" + + +from xstate import Machine, and_, create_actor, not_, or_, stateIn + +# --------------------------------------------------------------------------- +# Standalone semantics (no configuration → False) +# --------------------------------------------------------------------------- + + +def test_stateIn_called_directly_without_config_is_false(): + # With no active configuration there is nothing to match against. + assert stateIn("#ready")(None, None) is False + + +# --------------------------------------------------------------------------- +# stateIn as a transition guard — by id +# --------------------------------------------------------------------------- + + +def _two_region_machine(guard): + return Machine( + { + "id": "m", + "type": "parallel", + "states": { + "switch": { + "initial": "off", + "states": { + "off": {"on": {"FLIP": "on"}}, + "on": {"id": "switchOn", "on": {"FLIP": "off"}}, + }, + }, + "worker": { + "initial": "waiting", + "states": { + "waiting": {"on": {"GO": {"target": "active", "guard": guard}}}, + "active": {}, + }, + }, + }, + } + ) + + +def test_stateIn_by_id_blocks_when_not_in_state(): + machine = _two_region_machine(stateIn("#switchOn")) + actor = create_actor(machine).start() + # switch is off → guard fails → worker stays waiting + actor.send("GO") + assert actor.get_snapshot().value["worker"] == "waiting" + + +def test_stateIn_by_id_passes_when_in_state(): + machine = _two_region_machine(stateIn("#switchOn")) + actor = create_actor(machine).start() + actor.send("FLIP") # switch → on (#switchOn) + assert actor.get_snapshot().value["switch"] == "on" + actor.send("GO") + assert actor.get_snapshot().value["worker"] == "active" + + +# --------------------------------------------------------------------------- +# stateIn by dotted path +# --------------------------------------------------------------------------- + + +def test_stateIn_by_dotted_path(): + machine = Machine( + { + "id": "m", + "type": "parallel", + "states": { + "switch": { + "initial": "off", + "states": { + "off": {"on": {"FLIP": "on"}}, + "on": {"on": {"FLIP": "off"}}, + }, + }, + "worker": { + "initial": "waiting", + "states": { + "waiting": { + "on": { + "GO": { + "target": "active", + "guard": stateIn("switch.on"), + } + } + }, + "active": {}, + }, + }, + }, + } + ) + actor = create_actor(machine).start() + actor.send("GO") + assert actor.get_snapshot().value["worker"] == "waiting" + actor.send("FLIP") + actor.send("GO") + assert actor.get_snapshot().value["worker"] == "active" + + +# --------------------------------------------------------------------------- +# Composition with and_/or_/not_ +# --------------------------------------------------------------------------- + + +def test_stateIn_composed_with_and(): + machine = Machine( + { + "id": "m", + "type": "parallel", + "states": { + "switch": { + "initial": "off", + "states": { + "off": {"on": {"FLIP": "on"}}, + "on": {"id": "switchOn", "on": {"FLIP": "off"}}, + }, + }, + "worker": { + "initial": "waiting", + "states": { + "waiting": { + "on": { + "GO": { + "target": "active", + "guard": and_("isAllowed", stateIn("#switchOn")), + } + } + }, + "active": {}, + }, + }, + }, + }, + guards={"isAllowed": lambda c, e: True}, + ) + actor = create_actor(machine).start() + actor.send("GO") # switchOn not active → blocked + assert actor.get_snapshot().value["worker"] == "waiting" + actor.send("FLIP") + actor.send("GO") # both true → passes + assert actor.get_snapshot().value["worker"] == "active" + + +def test_stateIn_composed_with_not(): + machine = _two_region_machine(not_(stateIn("#switchOn"))) + actor = create_actor(machine).start() + # switch is off → not(stateIn) is True → passes immediately + actor.send("GO") + assert actor.get_snapshot().value["worker"] == "active" + + +def test_stateIn_composed_with_or(): + machine = _two_region_machine(or_(stateIn("#switchOn"), lambda c, e: False)) + actor = create_actor(machine).start() + actor.send("GO") + assert actor.get_snapshot().value["worker"] == "waiting" + actor.send("FLIP") + actor.send("GO") + assert actor.get_snapshot().value["worker"] == "active" + + +# --------------------------------------------------------------------------- +# Named registry resolution + setup() +# --------------------------------------------------------------------------- + + +def test_stateIn_registered_as_named_guard(): + machine = Machine( + { + "id": "m", + "type": "parallel", + "states": { + "switch": { + "initial": "off", + "states": { + "off": {"on": {"FLIP": "on"}}, + "on": {"id": "switchOn", "on": {"FLIP": "off"}}, + }, + }, + "worker": { + "initial": "waiting", + "states": { + "waiting": { + "on": {"GO": {"target": "active", "guard": "switchIsOn"}} + }, + "active": {}, + }, + }, + }, + }, + guards={"switchIsOn": stateIn("#switchOn")}, + ) + actor = create_actor(machine).start() + actor.send("GO") + assert actor.get_snapshot().value["worker"] == "waiting" + actor.send("FLIP") + actor.send("GO") + assert actor.get_snapshot().value["worker"] == "active"