pytest integration

A pytest plugin that turns a JSON chart into fixtures: a started interpreter on a simulated clock, stub logic that records what ran, and snapshot assertions — for both engines.

Testing a statechart is always the same setup: build the machine with stub logic, start a SyncInterpreter on a SimulatedClock, send events, advance the clock, assert on the configuration. Our own xsm gt -t pytest template emits exactly that boilerplate. The [testing] extra makes it declarative instead: put one marker on the test, name the fixtures you need, and the plugin builds the machine, starts it, and stops it at teardown — for the sync and the async engine. It also gives you a snapshot assertion that records to a file, fails with a unified diff, and never writes unless you ask it to.

Install

pip install "xstate-statemachine[testing]"

Requires pytest>=8. The extra also installs hypothesis>=6.100, used by the model-based tester; everything else on this page needs only pytest. The plugin is registered through a pytest11 entry point, so pytest finds it on its own — no conftest.py, no pytest_plugins line. xsm_ainterp additionally needs pytest-asyncio. Tested versions are in the compatibility table.

Quick start

A test file needs the marker and the fixture names — nothing else. The block below writes such a file and runs it through pytest, so it is executed in CI exactly as a user would run it:

import json, pathlib, sys, tempfile
import pytest
from xstate_statemachine import __version__ as _  # the package must be installed

TEST = '''
import pytest

CHECKOUT = {
    "id": "checkout",
    "initial": "cart",
    "context": {
        "items": 0,
    },
    "states": {
        "cart": {
            "on": {
                "PAY": {
                    "target": "paying",
                    "guard": "hasItems",
                    "actions": "charge",
                },
            },
        },
        "paying": {
            "after": {
                "3000": "confirmed",
            },
        },
        "confirmed": {
            "type": "final",
        },
    },
}


@pytest.mark.xstate_machine(CHECKOUT)  # a dict, a JSON path, or a MachineNode
def test_pay_then_confirm(xsm_interp, xsm_clock, xsm_ran, xsm_send_all):
    xsm_send_all(
        xsm_interp, "PAY", "+3000"
    )  # the `xsm simulate --events` grammar
    assert xsm_interp.matches("checkout.confirmed")
    assert xsm_ran == ["charge"]  # stub actions record their names


@pytest.mark.xstate_machine(CHECKOUT)
@pytest.mark.xstate_guards_false("hasItems")  # stub guard forced False...
def test_guard_blocks_payment(xsm_interp, xsm_guards):
    assert xsm_interp.send("PAY", wait=True).denied
    xsm_guards["hasItems"] = True  # ...and flipped live
    xsm_interp.send("PAY")
    assert xsm_interp.matches("checkout.paying")


@pytest.mark.xstate_machine(CHECKOUT)
def test_snapshot(xsm_interp, xsm_snapshot):
    xsm_interp.send("PAY")
    xsm_snapshot(
        xsm_interp, "snapshots/paying.json"
    )  # --xsm-update-snapshots records it
'''

with tempfile.TemporaryDirectory() as tmp:
    pathlib.Path(tmp, "test_checkout.py").write_text(TEST, encoding="utf-8")
    plugin = "xstate_statemachine.contrib.testing.pytest_plugin"
    common = ["-q", "-p", "no:cacheprovider", f"--rootdir={tmp}", tmp]
    # first run records the snapshot; the second must pass against it unchanged
    assert pytest.main([*common, "--xsm-update-snapshots"]) == 0, "recording run"
    assert pytest.main(common) == 0, "verification run"
    print(pathlib.Path(tmp, "snapshots", "paying.json").read_text(encoding="utf-8"))

The recorded snapshot holds only what describes behaviour — state_ids, value, context, status — with sorted keys and a trailing newline, so it is byte-identical across runs and machines (sets render as sorted lists). Deliberately not recorded: history, actors, pending or scheduled events and after deadlines — a regression in history restoration or timer scheduling is invisible to xsm_snapshot; cover those with xsm_path or an explicit assertion:

{
  "context": {
    "items": 0
  },
  "state_ids": ["checkout.paying"],
  "status": "running",
  "value": "paying"
}

Reference

Markers

Both markers are registered, so --strict-markers projects need nothing extra.

Fixtures

All fixtures are prefixed xsm_ so they cannot shadow your own machine, clock or store. Requesting one on a test without the marker fails at setup with a message naming the marker.

Options

The --xsm-*path* options are listed under Path generation, the --xsm-*coverage* options under State & transition coverage.

Helpers (importable from xstate_statemachine.contrib.testing)

Generated tests on the fixtures

xsm gt -t pytest --fixtures machine.json emits the recorded-trajectory test module on the plugin’s marker and fixtures instead of building the interpreter by hand: the same STEPS, the same assertions, no stub builder or fixture boilerplate in the file. Without --fixtures the output is unchanged.

Path generation

What a traversal does to your process. It explores a private copy of the machine (your machine.logic is never swapped, so a live interpreter beside the traversal is unaffected), but it mutes the xstate_statemachine logger to CRITICAL for its duration — process-wide. A step the engine refuses is dropped and a contained failure (an always loop at maxIterations) is kept; either raises one RuntimeWarning per traversal naming the error. Path.replay() on your real machine forces each step’s assumptions onto machine.logic for that step only and restores them; do not replay the same machine from two threads at once.

Request the xsm_path fixture under an xstate_machine marker and pytest parametrises the test over graph.shortest_paths(machine) — one case per reachable configuration, found by running the real engine on stub logic and a simulated clock (parallel regions, history and after timers included):

import pytest

@pytest.mark.xstate_machine("machines/AdvancePayment.json")
def test_every_state_is_reachable(xsm_path, xsm_interp, xsm_clock):
    xsm_path.replay(xsm_interp, xsm_clock)          # events + clock advances
    assert xsm_interp.current_state_ids == set(xsm_path.final_states)

test_every_state_is_reachable[path[editing]] PASSED
test_every_state_is_reachable[path[editing->challenge]] PASSED
test_every_state_is_reachable[path[editing->challenge->success]] PASSED

Ids name the leaf configurations the path walks through (parallel leaves joined with +). xsm_path is a graph.Path: steps, final_states, event_string() (the xsm simulate --events grammar) and replay(interp, clock), which forces each step’s recorded guard/service assumptions for that step only.

Bounds must be >= 0 (a negative value is a usage error). Tests on the same chart and options share one exploration per session. A chart that cannot start does not abort collection: each test requesting xsm_path on it gets a single path[error] case that errors with the engine’s message. To run one case, select it by node id (pytest "test_x.py::test_reach[path[editing->challenge]]"); -k cannot match the ids because its expression grammar rejects >.

📝 Named after delays (declared in MachineLogic.delays) have no static duration: their steps advance the clock by a large sentinel and carry a delay:<name>=unknown assumption. With real logic= the generation still uses stubs; replay then runs your real guards, so a path whose guard assumption your logic does not satisfy will (correctly) not reach its configuration.

State & transition coverage

Line coverage cannot tell you whether any test ever reached timeout or took paying --PAY_FAILED--> failed. The chart knows every state and transition; --xsm-coverage records which ones ran:

What 100 % means. The transition denominator is every transition the chart declares. On real charts some can never be taken: an always alternative shadowed by an earlier guard, an invoke onDone / onError the stub actors cannot drive, a named after delay with no implementation, a state no event sequence reaches, and self-target on handlers that generated paths skip (a hand-written test still hits them). On 30 Stately-corpus charts 105 of 731 declared transitions were unreachable this way (#270 battle). Gate on state coverage, or on a transition threshold below 100, and read the unhit list as a review item rather than a defect count. A chart with no transitions counts as 100 %. Spawned and invoked children are different charts: with --xsm-coverage (a global collector) each gets its own row under its own key; a collector you .use() on the parent does not see them at all. The session collector costs roughly 1.4-1.7× on send() in a test run, most of it the engine’s plugin dispatch – it is a test-time tool, not a production plugin.

$ pytest --xsm-coverage --xsm-fail-under-transition-coverage=90
...
---- xstate coverage ----
checkout          states 12/15 (80%)  transitions 18/22 (81.8%)
  unvisited: timeout, errorRecovery, refund.partial
  unhit:     paying --PAY_FAILED--> failed, after 30000 ...
FAIL xstate coverage: checkout: transition coverage 81.8% < 90%

The session registers one CoverageCollector with core’s plugins.register_global at start-up and unregisters it at the end, so every interpreter built during the run is counted — the xsm_* fixtures, a test’s own SyncInterpreter(...) / Interpreter(...), from_snapshot restores (their configuration counts at start()), spawned children, on any thread. Machines are grouped by machine.id + structure_hash, so two builds of one chart merge and an edited chart is a new row. A parallel configuration marks every active leaf and its ancestors; a history restore marks the states actually re-entered. Denominators: every state except the root and history pseudo-states, and transition_coverage_targets(machine) for transitions (on, always, after, onDone, invoke onDone/onError).

The report options and thresholds need --xsm-coverage: on their own they are a usage error, not ignored. Repeating a report (term twice, the same json:PATH twice) produces it once. Files are written even when tests fail. --collect-only writes and gates nothing. If a threshold is set and no machine was observed (-k matched nothing, or no test built an interpreter), the gate fails rather than passing on nothing. When the run stopped early (-x, --maxfail), the summary says (session interrupted -- coverage partial). A report path that cannot be written (a directory, a parent that is a file) fails the session with one FAIL xstate coverage: cannot write ... line.

With pytest-xdist (-n N). Every worker records its own share and ships it to the controller at the end of the session. The controller merges them by machine (a state or transition counts as covered if any worker covered it), then prints the report, writes the files and applies the thresholds once, over the whole suite. Workers print no coverage section of their own.

The same collector works outside pytest:

from xstate_statemachine import SyncInterpreter, create_machine
from xstate_statemachine.coverage import CoverageCollector

m = create_machine(
    {
        "id": "t",
        "initial": "a",
        "states": {
            "a": {
                "on": {
                    "GO": "b",
                },
            },
            "b": {
                "on": {
                    "BACK": "a",
                },
            },
        },
    }
)
cov = CoverageCollector()
SyncInterpreter(m).use(cov).start().send("GO")
report = cov.report(m)
assert (report.states_visited, report.transitions_hit) == (2, 1)
print(report.to_text())  # also .to_json(), .to_html()

JSON schema (version: 1, stable). A reader must reject any other version.

{
  "version": 1,
  "machines": [
    {
      "machine": "checkout",
      "key": "checkout@<structure_hash>",
      "states": {
        "visited": 12,
        "total": 15,
        "percent": 80.0,
        "unvisited": ["checkout.timeout", ...]
      },
      "transitions": {
        "hit": 18,
        "total": 22,
        "percent": 81.82,
        "unhit": [
          {
            "from": "checkout.paying",
            "label": "on 'PAY_FAILED'",
            "to": "checkout.failed"
          },
          ...
        ]
      }
    }
  ]
}

xsm coverage xsm-coverage.json [--fail-under N] [--plain] [--json] renders it in CI logs and exits 1 when any machine is under N % (state or transition). A CI job:

- run: pytest --xsm-coverage --xsm-coverage-report=term --xsm-coverage-report=json:xsm-coverage.json
- run: xsm coverage xsm-coverage.json --fail-under 90 --plain
- uses: actions/upload-artifact@v4
  with: {name: xstate-coverage, path: xsm-coverage.json}

Model-based testing

model_test() turns the chart into a Hypothesis RuleBasedStateMachine. Hypothesis then generates random event sequences, keeps to the legal ones, checks your invariants after every step and, when one fails, shrinks the run to the shortest sequence that still breaks it. It also writes that sequence to a file you can replay in xsm simulate.

import json, pathlib, tempfile
from hypothesis import settings
from hypothesis.stateful import run_state_machine_as_test
from xstate_statemachine.contrib.testing import model_test

chart = {
    "id": "refund",
    "initial": "empty",
    "context": {
        "total": 0,
    },
    "states": {
        "empty": {
            "on": {
                "ADD": {
                    "target": "cart",
                    "actions": {
                        "type": "xstate.assign",
                        "params": {
                            "assignment": {
                                "total": 10,
                            },
                        },
                    },
                },
            },
        },
        "cart": {
            "on": {
                "PAY": "paid",
            },
        },
        "paid": {
            "on": {
                "REFUND": {
                    "target": "refunded",
                    "actions": {
                        "type": "xstate.assign",
                        "params": {
                            "assignment": {
                                "total": 0,
                            },
                        },
                    },
                },
            },
        },
        # 🐛 the seeded bug: a second REFUND drives the total negative
        "refunded": {
            "on": {
                "REFUND": {
                    "actions": {
                        "type": "xstate.assign",
                        "params": {
                            "assignment": {
                                "total": -10,
                            },
                        },
                    },
                },
            },
        },
    },
}
out = pathlib.Path(tempfile.mkdtemp()) / "failing.json"
TestRefund = model_test(  # in a test module, pytest collects this
    chart,
    invariants={
        "total never negative": lambda i: i.context["total"] >= 0,
    },
    settings=settings(max_examples=200, database=None, derandomize=True),
    failing_path=out,
)
try:
    run_state_machine_as_test(
        TestRefund, settings=TestRefund.TestCase.settings
    )
except AssertionError as exc:
    assert "invariant 'total never negative' violated" in str(exc)
assert [c["send"] for c in json.loads(out.read_text())] == [
    "ADD",
    "PAY",
    "REFUND",
    "REFUND",
]  # shrunk to the minimal 4 events

In a test module you only write TestRefund = model_test(...). The plugin collects it through its generated TestCase (plain unittest discovery can use TestRefund.TestCase). A failing run prints Hypothesis’s usual step list, followed by:

AssertionError: invariant 'total never negative' violated
minimal failing sequence (4 step(s)) written to tests/failing.json
replay: xsm simulate tests/refund.json --script tests/failing.json

What gets generated

model_test() parameters (all keyword-only after the machine)

Two helpers build Hypothesis strategies without a state machine: events_strategy(machine_or_path, *, length=10) draws event-type sequences in which every event is enabled (can()) in the configuration the earlier ones reached and stops early when none is, and payload_strategy(schema) infers a strategy of payload dicts from a pydantic EventModel (honouring field constraints and validators; a required field of an unsupported type raises ValueError, so pass payloads= for it). Both are importable from xstate_statemachine.contrib.testing and need hypothesis.

invariants (name → check(interp)) run after start and after every step. state_assertions (state → check(interp)) run whenever interp.matches(state), which includes every region of a parallel configuration. A check fails if it raises any exception or returns a falsy value other than None – so an assert-style check that returns nothing passes, False / 0 / "" fail, and a KeyError on a missing context key is a failure with the script written (#271 battle). A state_assertions key that names no state in the chart is a ValueError at model_test(). There are no built-in invariants: a refused event is not a failure, and once the machine reaches a top-level final state (status == "done") no further event rule is enabled, so the run settles. Every model test runs with Hypothesis deadline=None (pinned, whatever settings= you pass), so a slow step on a large chart never causes a flaky DeadlineExceeded. Hypothesis’s own health checks still apply. Payload strategies are chosen in this order: your payloads={"ADD": strategy}; otherwise strategies inferred from event_schemas / pydantic EventModel fields (bool, int, float, str, None, Decimal, datetime, date, Literal, Optional/Union, List, nested models as dicts; field constraints such as Field(ge=1) and validators are honoured by filtering draws through the model); otherwise an empty payload. If a required field has a type outside that mapping, you get a ValueError that tells you to pass payloads=. A payloads= strategy must produce dicts without a type key (TypeError otherwise). snapshot_roundtrip fails on purpose for a Decimal in context – a snapshot would bring it back as a str; it restores with restart_services=True, so a long-lived from_callback / from_iterator invoke runs its setup again on the copy. A logic= factory is called once per example so stateful real logic starts fresh; a factory that returns the same cached instance every call still shares state, and the event list, payload inference and delays are read from the first build (a factory that varies event_schemas or delays per call is not supported). assume(...) inside a check discards the example as usual. events_strategy(chart, length=10) is a plain strategy of legal event-name lists, for your own @given tests. The failing script is written as failing.json next to the calling module by default. --xsm-failing-dir=DIR (relative to the directory pytest is run from, created if missing, may be outside the rootdir) or model_test(failing_path=…) moves it. Two model tests that fail in the same run and share a directory write the same failing.json, so the last one wins; give each its own failing_path= when that matters. The file holds the generated event names and payloads (Hypothesis-generated data, nothing from your context), so treat it like any other test artefact and keep it out of version control (failing.json in .gitignore).

📝 In CI, pass settings=settings(derandomize=True, database=None) (or the --hypothesis-seed=N option) so a run is reproducible and does not depend on a local .hypothesis/ example database. --hypothesis-show-statistics and -k work as for any test. A logic factory (logic=make_logic) is called once per generated example and must return fresh state each time.

⚠️ With real logic=, can() runs your real guards. That is how the generated sequences stay legal, and it is safe as long as your guards are pure (no side effects), which they should be. A guard with side effects will run more often than your sends do.

Compared to @xstate/test. @xstate/test builds a test model from the chart, walks the shortest or simple paths it generates, and runs per-state assertions at each state. Its coverage is systematic, but it only follows the paths it enumerates, and when something fails it reports the path it was on. In this library, that systematic walk is the xsm_path fixture. model_test works the other way round. It samples random legal sequences, which reach combinations of loops, payloads, timer races and snapshot restores that path enumeration never builds, and it shrinks each failure to a minimal sequence you can replay. The trade-off: it is probabilistic. A rare path can go unseen at a low max_examples, and a green run is evidence, not proof. Use both. xsm_path shows that every state is reachable, model_test looks for sequences that break your invariants, and --xsm-coverage shows what either one actually exercised.

Fake broker

FakeBrokerAdapter (async) and SyncFakeBrokerAdapter (blocking) are in-memory brokers that satisfy the same BrokerAdapter contract suite as the real Kafka / RabbitMQ / NATS / SQS / Redis adapters: FIFO per subject, explicit ack / nack, a requeued nack goes back to the head, and a failed publish raises. Both are thread-safe.

from xstate_statemachine.contrib.testing import SyncFakeBrokerAdapter
from xstate_statemachine.eda import Envelope

broker = SyncFakeBrokerAdapter()
broker.deliver("orders", Envelope.new(type="xsm.o.PAY", subject="o-1"))
d = next(broker.subscribe("orders", timeout=0))
broker.nack(d, requeue=True)                   # back to the head
d = next(broker.subscribe("orders", timeout=0))
broker.ack(d)
assert (len(broker.acked), len(broker.nacked), broker.in_flight) == (1, 1, 0)

broker.fail_next_publish()                     # next publish raises
try:
    broker.publish("out", Envelope.new(type="xsm.o.DONE", subject="o-1"))
except ConnectionError:                        # BrokerPublishError
    pass
assert broker.published_on("out") == []

Both fakes are thread-safe (one lock around the queues); drain() runs handlers on the calling thread.

The outbox relay is at-least-once on the fake too. OutboxRelay.relay_once() publishes pending rows in order. When a publish raises, the exception reaches the caller: the rows already published are marked sent, and the failing row and every row after it stay pending for the next call. fail_next_publish() tests exactly that path.

Replaying a transition log

Given / When / Then

given(machine) returns a Scenario: a thin, fluent wrapper over a SyncInterpreter on a SimulatedClock. Every method returns the scenario, so one behaviour fits on one line:

from xstate_statemachine import create_machine
from xstate_statemachine.contrib.testing import given

machine = create_machine(
    {
        "id": "order",
        "initial": "pending",
        "states": {
            "pending": {
                "on": {
                    "PAY": "paid",
                },
            },
            "paid": {
                "after": {
                    "1000": "expired",
                },
            },
            "expired": {
                "type": "final",
            },
        },
    }
)
given(machine).in_state("pending").when("PAY").then_state("paid")
given(machine).in_state("paid").after(1000).then_done()

Every then_* failure is an AssertionError that starts with the step trail and then gives the expected and actual values, e.g. in_state('m.a') -> when('GO'): expected state(s) ['done'] to be active; active: ['m.b.a']. Misuse is a RuntimeError: then_changed / then_denied / then_error / then_no_error before any when(), in_state / with_context after it, or when / after once the interpreter is stopped or done.

⚠️ Sync engine only, one thread. Scenario is not thread-safe, and there is no async variant; for the async engine use the xsm_ainterp fixture. Two scenarios on one MachineNode run as independent interpreters, but they share the machine’s logic table, so a stub that keeps state is shared too.

⚠️ A failing then_context prints the values it compared, unredacted, as snapshot diffs do. They are your own test data, but keep real secrets (card tokens, passwords) out of test contexts.

BDD with pytest-bdd

pytest-bdd is not part of the [testing] extra; pip install pytest-bdd if you want Gherkin. Each step is a one-line call on a Scenario fixture. The repository’s own recipe is in tests/contrib/testing/bdd_order_specs/; CI runs it whenever pytest-bdd is installed and skips it otherwise:

Scenario: A paid order is packed and shipped
  Given the order is in state "paid"
  And the context has orderId "o"
  When I send "PACKED"
  Then the state is "shipped"
  And the context has trackingId "TRK-o"

@pytest.fixture
def spec(order_machine):
    with given(order_machine) as s:
        yield s

@bdd_given(parsers.parse('the order is in state "{state}"'))
def _in_state(spec, state):
    spec.in_state(state)

@when(parsers.parse('I send "{event}"'))
def _send(spec, event):
    spec.when(event)

@then(parsers.parse('the state is "{state}"'))
def _state_is(spec, state):
    spec.then_state(state)

Step arguments are plain strings from parsers.parse; nothing in a .feature file is evaluated. Scenario outlines (<start> / Examples:) work as usual. A conftest cannot call importorskip, so wrap its pytest_bdd import in try/except ImportError, and call pytest.importorskip("pytest_bdd") in the test module that calls scenarios(...).

Guarantees

What this does: builds the machine once per test from the marker, starts the interpreter on a simulated clock so no test waits on wall time, stops it at teardown, and fails loudly — with the test id — on any marker or logic mistake. Snapshot files are deterministic (sorted keys, fixed indent, behavioural fields only) and are written only under --xsm-update-snapshots. Without the marker the plugin does nothing: it registers fixtures and two options and requires no configuration. Both engines share the same semantics.

What this does not do: run your services, actions or guards unless you pass logic= — stubs complete synchronously and record, they do not do work; provide an event loop (xsm_ainterp needs pytest-asyncio); assert on timing (use xsm_clock); persist anything outside the snapshot files you name.

See the programme-wide Guarantees and Security pages (#303).

Threat model

Who can call this: whoever runs your test suite. The plugin is active in every pytest session of a project that installed the library; it reads nothing and writes nothing until a test carries the marker.

What it exposes: logic= imports and calls a dotted callable from your test — trusted code, the same trust as a conftest.py or LogicLoader; do not point it at untrusted modules. A marker is code: logic="os:system" would be imported and called like any other dotted name, so review markers as you review the test body. Snapshot diffs print the context values that differ, unredacted — it is your own context in your own test run. --xsm-update-snapshots writes files at the paths your tests name, resolved under the test tree. A snapshot file contains the machine’s context in clear text — do not put secrets in a context you snapshot, or redact before asserting.

You must configure: nothing for the sync engine. For the async engine, install pytest-asyncio and mark tests @pytest.mark.asyncio. Keep snapshot files under version control and review their diffs like code — a changed snapshot is a changed behaviour.

Compatibility

Troubleshooting