"""The scenario harness (erp_platform.scenarios), proved on a throwaway three-brick composition.
1. The proof catalogue (tests/scenarios/catalogue.py: 13 hand-written SC-SHOP scenarios + the generated
probes) passes IN PROCESS — one test per scenario, so a failure names the promise that broke.
2. The same catalogue passes in LIVE mode against the composition served on local sockets behind the
stand-in gateway, with real bearer tokens; the fault scenarios are SKIPPED with their reason, and with
DODIL_ENV=live the clock scenarios are skipped too — never passed.
3. The harness catches defects: each known defect put back into the orders brick makes the scenario written
for it FAIL. These are the tests that fail when the harness's own assertions are weakened (see the
mutation tests at the end, and AGENTS.md: verify by breaking).
4. The pieces: the demo clock route (refused in prod), work.set_clock, assertions, personas and secrets,
reports (redaction, JUnit), the CLI.
"""
import json
import os
import pathlib
import sys
import xml.etree.ElementTree as ET
from datetime import datetime, timezone
from decimal import Decimal
import pytest
HERE = pathlib.Path(__file__).resolve().parent
sys.path.insert(0, str(HERE / "scenario_bricks"))
from erp_platform import work # noqa: E402
from erp_platform.scenarios import core, report # noqa: E402
from erp_platform.scenarios.inprocess import InProcessRunner # noqa: E402
from erp_platform.scenarios.personas import LiveCredentials, RoleCatalogue, SecretsRefused # noqa: E402
sys.path.insert(0, str(HERE / "scenarios"))
import catalogue as shop # noqa: E402
CAT = shop.CAT
@pytest.fixture(scope="module")
def runner():
with InProcessRunner(CAT) as r:
yield r
@pytest.fixture(autouse=True)
def no_defects():
from scn_orders.application import use_cases
use_cases.DEFECTS.clear()
yield use_cases.DEFECTS
use_cases.DEFECTS.clear()
def by_id(runner, sid):
return next(sc for sc in runner.scenarios() if sc.id == sid)
def _ids():
from erp_platform import compose
return [sc.id for sc in CAT.expand(compose.load(CAT.compose))]
# ── 1. the catalogue, in process ───────────────────────────────────────────────────────────
@pytest.mark.parametrize("sid", _ids())
def test_the_catalogue_holds_in_process(runner, sid):
res = runner.run_one(by_id(runner, sid))
assert res.outcome == "passed", f"{sid}: {res.outcome} — {res.message}\n" + "\n".join(res.steps[-12:])
def test_the_catalogue_covers_what_it_must(runner):
"""~12 scenarios over every feature, plus all three generated probe kinds."""
scs = runner.scenarios()
hand = [sc for sc in scs if not sc.generated]
tags = {t for sc in hand for t in sc.tags}
assert len(hand) >= 12
assert {"clock", "failure", "concurrency", "public", "permissions", "money"} <= tags
kinds = {sc.id.split("-")[1] for sc in scs if sc.generated}
assert kinds == {"PERM", "ROWS", "IDEM", "MASS", "WILD"}
wild = {sc.id for sc in scs if sc.id.startswith("PROBE-WILD-")}
assert wild == {"PROBE-WILD-STAR", "PROBE-WILD-MALFORMED", "PROBE-WILD-ORDERS", "PROBE-WILD-STOCK",
"PROBE-WILD-MAILER"}
def test_scenarios_do_not_depend_on_their_order():
with InProcessRunner(CAT) as r:
results = r.run_all(lambda sc: not sc.generated, shuffle=20261017)
assert [x.outcome for x in results] == ["passed"] * len(results), [(x.id, x.message) for x in results
if x.outcome != "passed"]
# ── 2. live mode, against the composition on local sockets ─────────────────────────────────
@pytest.fixture
def local_deployment(tmp_path_factory):
# Function-scoped: a staged composition sets process-wide state (the seam registry, public.configure, the
# capability providers, PUBLIC_LINK_BASE_URL), so it must not outlive the test beside the in-process runner.
from erp_platform.scenarios import local
from erp_platform.scenarios.gateway import LocalIssuer
from erp_platform.scenarios.inprocess import _sqlite
d = tmp_path_factory.mktemp("live")
key = LocalIssuer.new_key()
issuer = LocalIssuer(key, issuer="https://appid.local/test/shop", audience="pool:shop")
local.write_secrets(d / "secrets", key, issuer=issuer.issuer, audience=issuer.audience)
with local.serve(CAT, lambda name: _sqlite(d / f"{name}.db"), issuer) as (base, public_base):
yield base, public_base, d / "secrets", issuer
def live_run(local_deployment, select=None):
from erp_platform.scenarios.live import LiveRunner
base, public_base, secrets, _ = local_deployment
with LiveRunner(CAT, base_url=base, public_base_url=public_base, allow=True, secrets=str(secrets)) as r:
return r.run_all(select)
def test_the_same_catalogue_runs_live_and_says_what_it_skipped(local_deployment):
results = live_run(local_deployment)
by = {r.id: r for r in results}
bad = [(r.id, r.message) for r in results if r.outcome in ("failed", "error")]
assert not bad, bad
skipped = {r.id for r in results if r.outcome == "skipped"}
wild = {i for i in skipped if i.startswith("PROBE-WILD-")} # synthetic personas: no live token
assert skipped - wild == {"SC-SHOP-04", "SC-SHOP-05", "SC-SHOP-06", "SC-SHOP-07", "SC-SHOP-08", "SC-SHOP-09"}
assert len(wild) == 5 and all("declared inprocess only" in by[i].message for i in wild)
assert all("fault injection" in by[i].message for i in skipped - wild)
# the clock moved through the deployment's demo clock route, and the permission matrix said what it did not call
assert by["SC-SHOP-03"].outcome == "passed"
assert "were not called live" in by["PROBE-PERM-CLERK"].message
def test_live_clock_scenarios_are_skipped_on_a_live_environment_never_passed(local_deployment, monkeypatch):
monkeypatch.setenv("DODIL_ENV", "live")
results = live_run(local_deployment, lambda sc: sc.id in ("SC-SHOP-02", "SC-SHOP-03", "SC-SHOP-01"))
by = {r.id: r for r in results}
assert by["SC-SHOP-01"].outcome == "passed"
for sid in ("SC-SHOP-02", "SC-SHOP-03"):
assert by[sid].outcome == "skipped"
assert "no demo clock (GET /orders/demo-clock → 404)" in by[sid].message
def test_the_stand_in_gateway_strips_forged_identity(local_deployment):
import httpx
from erp_platform.scenarios.gateway import LocalIssuer
base, _, _, issuer = local_deployment
forged = {"X-Dodil-User": json.dumps({"sub": "boss", "email": "boss@x", "app_roles": ["manager"]})}
assert httpx.get(f"{base}/orders/orders", headers=forged).status_code == 401
wrong = LocalIssuer(LocalIssuer.new_key(), issuer=issuer.issuer, audience=issuer.audience)
tok = wrong.mint({"sub": "x", "email": "x@x", "app_roles": ["manager"]})
assert httpx.get(f"{base}/orders/orders", headers={"Authorization": f"Bearer {tok}"}).status_code == 401
other_aud = LocalIssuer(issuer._key, issuer=issuer.issuer, audience="pool:other")
tok = other_aud.mint({"sub": "x", "email": "x@x", "app_roles": ["manager"]})
assert httpx.get(f"{base}/orders/orders", headers={"Authorization": f"Bearer {tok}"}).status_code == 401
def test_a_second_staged_composition_leaves_the_first_one_whole(runner, local_deployment):
"""Standing the composition up again (the local deployment) and down must give the in-process runner back
its seams: its next order still reserves stock."""
from erp_platform import seam
assert set(seam._LOCAL) >= {"orders", "stock", "mailer"}
def test_after_the_local_deployment_the_runner_still_reaches_its_neighbours(runner):
run = core.Run(runner, core.Scenario("SC-T-08", "t", lambda s: None), CAT)
run.as_("clerk").post("/orders/orders", shop.order(run)).ok().field("reserved", True)
def test_live_refuses_a_bucket_that_is_not_stamped_demo(monkeypatch, tmp_path):
from erp_platform.scenarios import live
r = live.LiveRunner(CAT, base_url="http://127.0.0.1:9")
with pytest.raises(live.LiveRefused, match="name the bucket"):
r.start()
monkeypatch.setattr(live, "bucket_stamp", lambda bucket, org="": "prod")
r = live.LiveRunner(CAT, base_url="http://127.0.0.1:9", bucket="customer-books")
with pytest.raises(live.LiveRefused, match="stamped 'prod', not 'demo'"):
r.start()
# ── 3. the harness catches defects (each fails when its assertion is weakened) ─────────────
def test_a_403_for_another_persons_row_is_caught(runner, no_defects):
no_defects.add("others_get_403")
for sid in ("SC-SHOP-12", "PROBE-ROWS-CLERK"):
res = runner.run_one(by_id(runner, sid))
assert res.outcome == "failed" and "403" in res.message, (sid, res.message)
def test_a_replay_that_changes_the_record_is_caught(runner, no_defects):
no_defects.add("replay_bumps")
res = runner.run_one(by_id(runner, "PROBE-IDEM-PLACE_ORDER"))
assert res.outcome == "failed" and "replay answered differently" in res.message, res.message
def test_a_save_that_retries_its_conflict_is_caught(runner, no_defects):
"""Retrying a Conflict is the defect BRICKS.md forbids: the last writer silently wins (hr SC-HR-08)."""
no_defects.add("retry_conflict")
res = runner.run_one(by_id(runner, "SC-SHOP-10"))
assert res.outcome == "failed" and "expected {200: 1, 409: 4}" in res.message, res.message
def test_an_owner_taken_from_the_body_is_caught(runner, no_defects):
no_defects.add("owner_from_body")
res = runner.run_one(by_id(runner, "PROBE-MASS-PLACE_ORDER"))
assert res.outcome == "failed" and "applied owner" in res.message, res.message
def test_a_skipped_permission_check_is_caught(runner, monkeypatch):
from erp_platform import bricks
monkeypatch.setattr(bricks, "_authorize", lambda uc, caller: None)
res = runner.run_one(by_id(runner, "PROBE-PERM-VIEWER"))
assert res.outcome == "failed" and "not 403 'missing permission: orders:order:write'" in res.message
def test_a_refusal_that_does_not_name_the_permission_is_caught(runner, monkeypatch):
"""The matrix wants exactly "403: missing permission: <perm>" — a 403 for another reason, or a 404, is a
different promise (and the CRM's live security check holds the deployment to the exact words)."""
from erp_platform import bricks
def vague(uc, caller):
if uc.permission and not bricks.auth.has_permission(caller, uc.permission):
raise bricks.Refused(403, "no")
monkeypatch.setattr(bricks, "_authorize", vague)
res = runner.run_one(by_id(runner, "PROBE-PERM-VIEWER"))
assert res.outcome == "failed" and "answered 403 'no'" in res.message, res.message
def test_the_matrix_can_expect_404_where_a_403_would_confirm_the_record(monkeypatch):
"""hr SC-HR-20: a manager gets 404, not 403, on show_sensitive. `expect=` overrides a cell — and holds it."""
from erp_platform import compose
from erp_platform.scenarios import probes
comp = compose.load(CAT.compose)
gen = probes.permission_matrix(personas=["viewer"], expect={("viewer", "approve_order"): 404})
sc = next(x for x in gen(CAT, comp) if x.id == "PROBE-PERM-VIEWER")
with InProcessRunner(CAT) as r:
res = r.run_one(sc) # the brick answers 403 "missing permission" — the override says 404
assert res.outcome == "failed" and "approve_order via http: answered 403" in res.message
assert "not 404 (the matrix's expect=)" in res.message
def test_a_public_link_can_submit_a_file(runner):
run = core.Run(runner, core.Scenario("SC-T-07", "t", lambda s: None), CAT)
clerk = run.as_("clerk")
body = shop.order(run)
clerk.post("/orders/orders", body).ok()
url = clerk.post(f"/orders/orders/{body['order_id']}/approval-link", {"customer_email": run.email()}).ok()["url"]
sent = run.link(url).submit({"note": "ok"}, files={"proof": ("po.pdf", b"%PDF-1.4", "application/pdf")})
assert sent.status_code in (200, 415) # 415 only where python-multipart is not installed
assert sent.exchange.request == {"form": {"note": "ok"}}
def test_a_missing_recipe_fails_the_probe_instead_of_passing_it(runner, monkeypatch):
monkeypatch.delitem(shop.RECIPES, "approve_order")
res = runner.run_one(by_id(runner, "PROBE-IDEM-APPROVE_ORDER"))
assert res.outcome == "failed" and "has no recipe" in res.message
# ── 4a. the clock ──────────────────────────────────────────────────────────────────────────
def test_the_clock_advances_in_working_days_at_the_same_wall_clock_time(runner):
run = core.Run(runner, core.Scenario("SC-T-01", "t", lambda s: None), CAT)
try:
run.clock.set("2026-10-23T10:00 Europe/London") # Friday, BST
assert run.clock.advance("+3 working days") == datetime(2026, 10, 28, 10, 0, tzinfo=timezone.utc) # GMT
finally:
work.reset_clock()
# ── 4b. assertions ─────────────────────────────────────────────────────────────────────────
def resp(status=200, body=None):
return core.Response(core.Exchange("http", "clerk", "GET", "/x", None, status, body if body is not None else {}, 1))
def test_not_found_means_404_and_nothing_else():
resp(404).not_found()
for status in (403, 401, 200, 500):
with pytest.raises(core.ScenarioFailed):
resp(status).not_found()
def test_money_is_compared_exactly():
resp(body={"total": "100.00"}).field("total", Decimal("100.00"))
for wrong in ("100.0", "100", "99.99", 100.0, True):
with pytest.raises(core.ScenarioFailed):
resp(body={"total": wrong}).field("total", Decimal("100.00"))
def test_types_are_strict():
with pytest.raises(core.ScenarioFailed):
resp(body={"n": "1"}).field("n", 1)
with pytest.raises(core.ScenarioFailed):
resp(body={"n": 1}).field("n", True)
resp(body={"a": {"b": [1, {"c": "x"}]}}).field("a.b.1.c", "x")
def test_same_as_compares_the_body_not_just_the_status():
resp(body={"v": 1, "t": "a"}).same_as(resp(body={"v": 1, "t": "b"}), ignore=("t",))
with pytest.raises(core.ScenarioFailed, match="in v"):
resp(body={"v": 2}).same_as(resp(body={"v": 1}))
with pytest.raises(core.ScenarioFailed):
resp(409, {"v": 1}).same_as(resp(200, {"v": 1}))
def test_refused_checks_the_reason():
r = resp(409, {"detail": "changed since you opened it — reload"})
r.refused(409, "changed since you opened it")
with pytest.raises(core.ScenarioFailed):
r.refused(409, "not your order")
def test_eventually_gives_up_with_the_last_answer(runner):
run = core.Run(runner, core.Scenario("SC-T-04", "t", lambda s: None), CAT)
with pytest.raises(core.ScenarioFailed, match="did not hold within 0.2s"):
run.eventually(lambda: resp(body={"s": "open"}), until=lambda r: r["s"] == "done", timeout=0.2, every=0.05)
def test_concurrent_answers_are_counted_exactly():
rs = core.Responses([resp(200), resp(409), resp(409)])
rs.expect({200: 1, 409: 2})
with pytest.raises(core.ScenarioFailed):
rs.expect({200: 3})
def test_a_scenario_id_follows_the_catalogue_convention():
cat = core.Catalogue("t", compose="x", roles={"roles": {"a": []}})
with pytest.raises(ValueError, match="SC-<MODULE>-nn"):
cat.scenario("orders-1", "t")(lambda s: None)
cat.scenario("SC-HR-08", "t")(lambda s: None)
with pytest.raises(ValueError, match="already"):
cat.scenario("SC-HR-08", "t")(lambda s: None)
def test_a_scenario_acts_only_as_the_personas_it_declares(runner):
sc = core.Scenario("SC-T-05", "t", lambda s: s.as_("manager"), personas=("clerk",))
res = runner.run_one(sc)
assert res.outcome == "error" and "declares personas" in res.message
def test_a_fault_in_live_mode_is_a_skip_not_a_pass():
r = core.Runner(CAT)
r.mode = "live"
res = r.run_one(core.Scenario("SC-T-06", "t", lambda s: s.fault.seam_down("stock")))
assert res.outcome == "skipped" and "in-process only" in res.message
# ── 4c. personas and secrets ───────────────────────────────────────────────────────────────
def test_roles_resolve_to_the_gateways_identity():
roles = RoleCatalogue.load(HERE / "scenarios" / "roles.yaml")
assert roles.claims("clerk")["permissions"] == ["orders:order:write", "stock:reservation:read"]
assert roles.claims("clerk#2")["email"] != roles.claims("clerk")["email"]
assert roles.claims("anonymous") is None
assert roles.holds("manager", "orders:order:approve") and not roles.holds("clerk", "orders:order:approve")
def test_secrets_readable_by_others_are_refused(tmp_path):
roles = RoleCatalogue.load(HERE / "scenarios" / "roles.yaml")
d = tmp_path / "s"
d.mkdir()
(d / "personas.json").write_text(json.dumps({"pool": "dodil/x", "personas": {}}))
os.chmod(d / "personas.json", 0o644)
with pytest.raises(SecretsRefused, match="chmod 600"):
LiveCredentials.from_env(roles, str(d))
os.chmod(d / "personas.json", 0o600)
creds = LiveCredentials.from_env(roles, str(d))
assert creds.missing(["clerk", "anonymous"]) == ["clerk"]
assert creds.token_url == "https://appid.dodil.io/dodil/x/token"
def test_live_tokens_come_from_the_password_grant_and_client_credentials(tmp_path, monkeypatch):
import httpx
roles = RoleCatalogue.load(HERE / "scenarios" / "roles.yaml")
d = tmp_path / "s"
d.mkdir()
p = d / "personas.json"
p.write_text(json.dumps({"token_url": "https://appid.example/o/p/token", "personas": {
"clerk": {"email": "c@x", "password": "pw-c"}, "ticker": {"client_id": "tick", "client_secret": "sec"}}}))
os.chmod(p, 0o600)
sent = []
def fake_post(url, data, timeout):
sent.append((url, dict(data)))
return httpx.Response(200, json={"access_token": f"tok-{data['grant_type']}", "expires_in": 600})
monkeypatch.setattr(httpx, "post", fake_post)
creds = LiveCredentials.from_env(roles, str(d))
assert creds.token("clerk") == "tok-password" and creds.token("clerk") == "tok-password" # cached
assert creds.token("ticker") == "tok-client_credentials"
assert sent == [("https://appid.example/o/p/token", {"grant_type": "password", "email": "c@x", "password": "pw-c"}),
("https://appid.example/o/p/token", {"grant_type": "client_credentials", "client_id": "tick",
"client_secret": "sec"})]
# ── 4d. reports ────────────────────────────────────────────────────────────────────────────
def failed_result():
x = core.Exchange("public", "the customer", "POST", "/p/" + "A" * 43,
{"headers": {"Authorization": "Bearer eyJhbGciOi.eyJzdWIiOiJ4In0.sig", "X-Dodil-User-JWT": "h.p.s"},
"body": {"password": "hunter2", "url": "https://pub.example/p/" + "B" * 43}},
500, {"detail": "boom"}, 3)
return core.Result("SC-SHOP-99", "a failure", "live", "failed", 0.5, message="expected 200, got 500",
stories=("SHOP-01",), steps=["WHEN it fails"], exchanges=[x], failing=x)
def test_reports_redact_tokens_secrets_and_link_tokens():
res = [failed_result(), core.Result("SC-SHOP-98", "skipped one", "live", "skipped", 0.0, message="in-process only: x")]
for text in (report.markdown(res), report.junit(res), report.console(res)):
for secret in ("hunter2", "A" * 43, "B" * 43, "eyJzdWIiOiJ4In0", "h.p.s"):
assert secret not in text
md = report.markdown(res)
assert "Request → response" in md and "→ 500" in md and "/p/[token]" in md
assert "## Skipped — and why" in md and "in-process only: x" in md
def test_junit_is_well_formed_and_counts():
res = [failed_result(), core.Result("SC-A-01", "ok", "inprocess", "passed", 0.1),
core.Result("SC-A-02", "skip", "live", "skipped", 0.0, message="why")]
suite = ET.fromstring(report.junit(res, "shop"))
assert (suite.get("tests"), suite.get("failures"), suite.get("skipped")) == ("3", "1", "1")
assert suite.find("testcase/failure") is not None and suite.find("testcase/skipped").get("message") == "why"
# ── 4e. the CLI ────────────────────────────────────────────────────────────────────────────
def test_the_cli_lists_and_runs(tmp_path, capsys):
from erp_platform.scenarios import cli
target = str(HERE / "scenarios" / "catalogue.py")
assert cli.main(["list", target]) == 0
assert "SC-SHOP-04" in capsys.readouterr().out
xml, md = tmp_path / "out.xml", tmp_path / "out.md"
assert cli.main(["run", target, "-k", "SC-SHOP-01", "-k", "clock", "--junit", str(xml), "--markdown", str(md)]) == 0
out = capsys.readouterr().out
assert "3 scenarios: 3 passed" in out
assert ET.parse(xml).getroot().get("tests") == "3" and "SC-SHOP-02" in md.read_text()
def test_the_cli_exits_non_zero_on_a_failure(capsys, no_defects):
from erp_platform.scenarios import cli
no_defects.add("others_get_403")
assert cli.main(["run", str(HERE / "scenarios" / "catalogue.py"), "-k", "SC-SHOP-12"]) == 1
# ── 5. mutation: weaken the harness, and the defect tests above are what notice ───────────
def test_mutation_a_404_check_that_accepts_403_lets_the_defect_through(runner, monkeypatch, no_defects):
"""Proves test_a_403_for_another_persons_row_is_caught has teeth: with not_found() weakened to accept 403
(the mutation), the defective brick passes SC-SHOP-12 — so that test would fail."""
def weak(self):
if self.status_code not in (403, 404):
self.fail("expected 404")
return self
monkeypatch.setattr(core.Response, "not_found", weak)
no_defects.add("others_get_403")
assert runner.run_one(by_id(runner, "SC-SHOP-12")).outcome == "passed"
def test_mutation_an_idempotency_check_on_status_alone_lets_the_defect_through(runner, monkeypatch, no_defects):
def weak(self, before, ignore=()):
if self.status_code != before.status_code:
self.fail("status differs")
return self
monkeypatch.setattr(core.Response, "same_as", weak)
no_defects.add("replay_bumps")
assert runner.run_one(by_id(runner, "PROBE-IDEM-PLACE_ORDER")).outcome == "passed"