Python

Install

pip install pysnappea

The package is imported as snappea. Its platform wheels bring the snappea checker, so no Rust or Java is needed. Python 3.9 or later. Set SNAPPEA_BIN to use another binary; pip install ./python in a checkout builds one with cargo.

Production code depends on the same package and imports only emit, which has no dependencies and does nothing unless a test is recording or a production sink is set.

Quick start

The examples on every language page check one spec: an inventory where reservations never exceed the stock and releases match reservations.

// The inventory contract, checked against the P model in this folder and against every example application.
event eStocked: (units: int);
event eReserved: (order: string, units: int);
event eReleased: (order: string, units: int);

// Applications give each product its own monitor through emit's key; the P model has one product.
spec InventoryCorrect observes eStocked, eReserved, eReleased {
  var available: int;
  var reservations: map[string, int];

  start state Unstocked {
    on eStocked do (p: (units: int)) {
      assert p.units >= 0, "negative initial stock";
      available = p.units;
      goto Ready;
    }
    on eReserved do { assert false, "reservation before stock setup"; }
    on eReleased do { assert false, "release before stock setup"; }
  }

  state Ready {
    on eStocked do { assert false, "product stocked twice"; }
    on eReserved do (p: (order: string, units: int)) {
      assert p.units > 0, "reservation must be positive";
      assert !(p.order in reservations), "duplicate reservation";
      available = available - p.units;
      assert available >= 0, "reservation exceeds available stock";
      reservations[p.order] = p.units;
    }
    on eReleased do (p: (order: string, units: int)) {
      assert p.order in reservations, "release without a reservation";
      assert p.units == reservations[p.order], "released quantity differs from reservation";
      available = available + p.units;
      reservations -= p.order;
    }
  }
}

The service emits an event after each change:

from snappea import emit


class Inventory:
    def __init__(self):
        self.stock = {}
        self.reservations = {}

    def add_product(self, sku, units):
        self.stock[sku] = units
        emit("eStocked", key=sku, units=units)

    def reserve(self, sku, order, units):
        if units > self.stock[sku]:
            raise ValueError("insufficient stock")
        self.stock[sku] -= units
        self.reservations[sku, order] = units
        emit("eReserved", key=sku, order=order, units=units)  # after the change

Point the pytest plugin at the specs. From then on every test also checks every spec:

[tool.snappea]
spec_files = ["specs/*.p"]

import contextlib

from inventory import Inventory


def test_two_orders_for_the_last_units():
    inventory = Inventory()
    inventory.add_product("coffee", 3)
    inventory.reserve("coffee", "order-1", 2)
    with contextlib.suppress(ValueError):  # rejecting order-2 is fine too
        inventory.reserve("coffee", "order-2", 2)

pytest passes. Now break the service: delete the stock check from reserve. The test's own code still passes, because it accepts either outcome for the second order. The spec does not:

FAILED test_inventory.py::test_two_orders_for_the_last_units

E   snappea.SpecViolation: P spec InventoryCorrect was broken: reservation exceeds available stock

      Events (key "coffee"):
          1  eStocked   units=3                inventory/__init__.py:11
          2  eReserved  order=order-1 units=2  inventory/__init__.py:16
          3  eReserved  order=order-2 units=2  inventory/__init__.py:16  <- broke the spec here

      Spec: specs/inventory.p:27
      snappea output: /tmp/pytest-of-me/pytest-1/snappea/snappea-check-zxk7z0st

Test setup

Every test: the pytest plugin

Installing the package adds the plugin. It does nothing until a [tool.snappea] table (or SNAPPEA_SPEC_FILES) exists:

[tool.snappea]
spec_files = ["specs/*.p"]       # globs, relative to this file; folders mean every .p in them
exclude = ["DraftSpec"]          # optional: specs not checked anywhere yet
mode = "auto"                    # optional: "auto" (every test) | "marked" | "off"

• Events emitted while fixtures are set up count. The check runs when the test body ends, so a broken spec is a test failure, not a teardown error.
• A test that already failed is not checked. A test that emits nothing passes without running the checker.
• pytest --snappea off (or marked, auto) overrides the mode for one run.

Per-test options are marks:

@pytest.mark.snappea(skip=True)              # don't check this test
@pytest.mark.snappea(skip=["DraftSpec"])     # leave out some specs here
@pytest.mark.snappea(check_hot=True)         # also fail on a spec left waiting

The snappea_events fixture is the test's recorder, for ordinary assertions: snappea_events.count("eEnter"), snappea_events.events_named("eEnter").

Explicit checks

Three functions make the check visible in the test. They work without the plugin.

from snappea import check_spec, check_events, expect_violation

def test_release_makes_stock_available_again():
    with check_spec("InventoryCorrect") as events:     # no names: every spec
        inventory = Inventory()
        inventory.add_product("coffee", 3)
        inventory.reserve("coffee", "order-1", 3)
        inventory.release("coffee", "order-1")
    assert events.count("eReleased") == 1

• check_spec(*specs, ...) checks the events emitted inside the block and raises SpecViolation at the end of it. Options left as None come from [tool.snappea], then the defaults.
• check_events(events, *specs) checks a hand-written list of events, for testing a spec on its own.
• Recorder(specs) is the manual form: start(), stop(), check().
• Checks nest, and every active recorder gets every event, including events from threads the test starts.

Negative controls

Prove the spec catches the bug it is meant to catch:

def test_spec_rejects_overselling():
    with expect_violation("InventoryCorrect", match="exceeds available stock"):
        check_events([
            {"event": "eStocked", "key": "coffee", "units": 1},
            {"event": "eReserved", "key": "coffee", "order": "o1", "units": 2},
        ], "InventoryCorrect")

It fails unless that spec, and only it, is broken with a matching message. Enclosing checks ignore that spec for the rest of the test, so the deliberate violation is not reported twice.

Property tests

With Hypothesis, give each generated case its own recording: with check_spec(...) inside the @given body, or a Recorder per RuleBasedStateMachine.

Emitting events

emit(event, payload=..., /, *, key=None, at=None, **fields)

emit("eReserved", key=sku, order=order, units=units)
emit("eCommitted", key=tx, at=row.commit_time)   # when it happened, not when logged

• Payload: keyword arguments, matched by name against the P event's payload type. A single positional value works for payloads that are not named tuples, or whose fields are called key or at. Enums are sent by name, sets and tuples as lists.
• key splits events into independent streams, such as one per product. Each stream gets its own instance of every spec.
• at sets the event time: a datetime, or an integer of epoch nanoseconds (a logical clock works). Default: now.
• Emit when a change becomes visible (after the commit), not when it is attempted, or specs see states that never existed.
• Each event records the file and line it was emitted from, for failure reports.

In production

The same calls can feed snappea observe on production logs. Set a sink once at start-up:

import snappea

sink = snappea.set_sink("/var/log/app/events.jsonl")   # a file, appended to
snappea.set_sink("-")                                  # or stdout, or any text stream
snappea.set_sink(logger.info)                          # or a callable taking one JSON line

snappea observe specs/*.p /var/log/app/events.jsonl --spec InventoryCorrect

A background thread writes the lines. emit never raises or blocks: when the queue (10,000 lines) is full, a write fails or a payload cannot be encoded, the event is dropped and counted in sink.dropped. Queued lines are written at exit; sink.flush() waits for them earlier and snappea.set_sink(None) turns the sink off. In a test, the recording takes precedence over the sink.

Other processes

with check_spec("InventoryCorrect") as rec:
    subprocess.run(["./worker"], env=rec.env(), check=True)

Stop or join child processes before the check. A service started once per session (a session-scoped fixture) starts before any test records, so its events reach no test. Give it a folder of its own and check that folder at the end of the session:

@pytest.fixture(scope="session")
def server(tmp_path_factory):
    events = tmp_path_factory.mktemp("server-events")
    proc = start_server(env={**os.environ, "SNAPPEA_EVENTS_DIR": str(events)})
    yield proc
    proc.terminate()
    rec = Recorder()
    rec.add_events_dir(str(events))
    rec.check()

Failures

A failure shows the spec's assert message, the events of the key that broke it (the last 20) with where each was emitted, the assert in the P file, and the folder that keeps the events and the checker's output. An event whose time was set with at= says so, which explains an order that differs from the emit order.

snappea.SpecViolation is an AssertionError, so pytest and Hypothesis treat it as an ordinary failure. Its fields: .spec, .message, .key, .events (each with .event, .payload, .src), .location, .liveness (a hot state at the end) and .others (further violations). Anything else is a snappea.SetupError with a .kind:

Reference

Settings

Set in [tool.snappea] of the nearest pyproject.toml (paths relative to it), or pass as arguments to check_spec, check_events, expect_violation and Recorder. Precedence: argument, environment variable, [tool.snappea], default. Unknown keys are an error.

Strict checks

Strict means every selected spec checked at least one relevant event, which catches instrumentation that was removed or never ran. An explicit check with named specs (check_spec("Name"), Recorder(specs=[...])) defaults to strict: a named spec that saw no relevant event fails the check with a SetupError of kind "validation". Pass strict=False where a spec may rightly see no events. (snappea.StrictDefaultWarning, which warned about this change for one release, is no longer emitted.) The plugin checks every test, and most tests emit no events for a given spec, so its checks are not strict unless asked, even with [tool.snappea] specs: set strict = true, run pytest --snappea-strict, or mark one test @pytest.mark.snappea(strict=True). Mark uninstrumented tests skip=True.

Environment variables

Deprecated for one release, with a warning: SNAPPEA_BINARY (use SNAPPEA_BIN), liveness= (use check_hot=) and snappea.SnappeaError (use SetupError).

Suite coverage

At the end of a run the plugin prints the spec coverage of all its checks: the states, handlers, statements and branches they reached, then what none reached, such as Spec.State: state never reached. Nothing is written, so each run reports only itself; pytest-xdist workers report to the controller. pytest --snappea-coverage=coverage/monitor.json (or SNAPPEA_COVERAGE) also writes it as JSON for CI. Runs merge into an existing file under a lock, so workers and separate runs can share one path; delete the file to start over. --snappea-coverage=off (or SNAPPEA_COVERAGE=off) turns the report off. Without a path, a reporting error is printed but never fails the run.

How it works

Each check writes the test's events to a file and runs snappea observe on them with --json-report, then reads the report. A check costs about 5 ms; tests that emit nothing cost nothing.

API

• snappea (pdoc)

Runnable projects: Python examples.