"""The scenario catalogue of the proof composition (tests/scenarios/compose.yaml): SC-SHOP-nn.
Written once; run in process by tests/test_scenarios.py, and live by scripts/live_scenario_check.py (the same
composition served locally against the erp-platform-test bucket). Each scenario makes its own data with
unique keys (s.key / s.email), so none depends on another, or on the order they run in.
erp-scenarios list tests/scenarios/catalogue.py
erp-scenarios run tests/scenarios/catalogue.py
"""
import pathlib
import sys
from decimal import Decimal
HERE = pathlib.Path(__file__).resolve().parent
sys.path.insert(0, str(HERE.parent / "scenario_bricks"))
from erp_platform.scenarios import Catalogue, Rows # noqa: E402
CAT = Catalogue("scenario-shop", compose=HERE / "compose.yaml", roles=HERE / "roles.yaml",
ticks=["POST /orders/tick"], tick_as="ticker", clock="orders", zone="Europe/London")
def order(s, sku=None, quantity=3, unit_price="33.33", customer="Acme"):
"""The body of a one-line order, with a fresh id (and SKU, so stock counts are this scenario's own)."""
return {"order_id": s.key("O"), "customer": customer,
"lines": [{"sku": sku or s.key("SKU"), "quantity": quantity, "unit_price": unit_price}]}
def stock_of(s, body, who="clerk"):
return s.as_(who).get(f"/stock/skus/{body['lines'][0]['sku']}/reservations")
@CAT.scenario("SC-SHOP-01", "an order's total is exact, and an agent reads the same order through MCP",
stories=["SHOP-01", "SHOP-02"], tags=["money", "doors"], personas=["clerk"])
def exact_money_both_doors(s):
s.given("a clerk")
clerk = s.as_("clerk")
s.when("they place 3 × 33.33")
body = order(s)
placed = clerk.post("/orders/orders", body)
s.then("the line and the total are exactly 99.99 — a string, never a float")
placed.ok().field("total", Decimal("99.99")).field("lines.0.amount", Decimal("99.99")).field("currency", "EUR")
s.then("their agent reads exactly the same order through MCP")
via_api = clerk.get(f"/orders/orders/{body['order_id']}").ok()
clerk.tool("get_order", order_id=body["order_id"]).ok().same_as(via_api)
@CAT.scenario("SC-SHOP-02", "due three working days later at 09:00 London time, across the end of summer time",
stories=["SHOP-03"], tags=["clock"], personas=["clerk"])
def due_across_dst(s):
s.given("Friday 16 October 2026, 10:00 in London (summer time, UTC+1)")
s.clock.set("2026-10-16T10:00 Europe/London")
clerk = s.as_("clerk")
s.when("a clerk places an order")
summer = clerk.post("/orders/orders", order(s)).ok()
s.then("it is due Wednesday 21st 09:00 BST — 08:00Z")
summer.field("placed_at", "2026-10-16T09:00:00Z").field("due_at", "2026-10-21T08:00:00Z")
s.given("a week later, Friday 23rd 10:00 — the clocks go back on Sunday 25th")
s.clock.advance("+5 working days")
s.when("another order is placed")
winter = clerk.post("/orders/orders", order(s)).ok()
s.then("it is due Wednesday 28th 09:00 GMT — 09:00Z, still 09:00 on the office wall")
winter.field("placed_at", "2026-10-23T09:00:00Z").field("due_at", "2026-10-28T09:00:00Z")
@CAT.scenario("SC-SHOP-03", "the tick makes an order overdue only once its due time has passed, and once",
stories=["SHOP-04"], tags=["clock", "tick"], personas=["clerk", "ticker"])
def overdue_by_tick(s):
s.given("an order placed Friday 23 October 2026, 10:00 London — due Wednesday 28th 09:00")
s.clock.set("2026-10-23T10:00 Europe/London")
clerk = s.as_("clerk")
body = order(s)
clerk.post("/orders/orders", body).ok().field("due_at", "2026-10-28T09:00:00Z")
get = lambda: clerk.get(f"/orders/orders/{body['order_id']}")
s.when("the scheduler ticks at 08:59 on the Wednesday")
s.clock.set("2026-10-28T08:59 Europe/London")
s.tick()
s.then("it is still open")
get().field("status", "open")
s.when("two minutes pass and the scheduler keeps ticking")
s.clock.advance("+2m")
s.then("it becomes overdue")
s.eventually(get, until=lambda r: r["status"] == "overdue", tick=True, timeout=5, what="the order is overdue")
s.then("a further tick does not touch it again")
again = s.tick()[0]
s.check(body["order_id"] not in again["overdue"], "a second tick changes nothing", again.body)
@CAT.scenario("SC-SHOP-04", "stock down: the order is still placed and says why; the retry reserves it once",
stories=["SHOP-05"], tags=["failure", "seam"], personas=["clerk"])
def stock_down(s):
clerk = s.as_("clerk")
body = order(s)
with s.fault.scope() as f:
s.given("the stock brick is unreachable")
f.seam_down("stock")
s.when("a clerk places an order")
r = clerk.post("/orders/orders", body)
s.then("the order exists, not reserved, saying why — not a 500")
r.ok().field("reserved", False).matches("reserve_note", lambda v: "unreachable" in (v or ""), "a reason")
s.when("stock is back and the clerk places the same order again")
clerk.post("/orders/orders", body).ok().field("reserved", True).field("reserve_note", None)
s.then("exactly one reservation was made")
stock_of(s, body).ok().field("count", 1).field("quantity", 3)
@CAT.scenario("SC-SHOP-05", "stock too slow: the order does not hang, and a reservation that landed late is not doubled",
stories=["SHOP-05"], tags=["failure", "seam"], personas=["clerk"])
def stock_slow(s):
clerk = s.as_("clerk")
body = order(s)
with s.fault.scope() as f:
s.given("stock answers after 1.5s and the seam waits 0.3s")
f.seam_slow("stock", 1.5, timeout=0.3)
s.when("a clerk places an order")
r = clerk.post("/orders/orders", body)
s.then("it answers at once: placed, not reserved")
r.ok().field("reserved", False)
s.check(r.exchange.duration_ms < 1400, "the order did not wait for the slow neighbour", r.exchange.duration_ms)
s.then("the late reservation lands anyway — a timeout is not a 'no'")
s.eventually(lambda: stock_of(s, body), until=lambda x: x["count"] == 1, timeout=5, what="stock shows it")
s.when("stock is quick again and the clerk retries")
clerk.post("/orders/orders", body).ok().field("reserved", True)
s.then("still one reservation")
stock_of(s, body).ok().field("count", 1)
@CAT.scenario("SC-SHOP-06", "a crash between storing the order and reserving its stock, then a retry",
stories=["SHOP-05", "SHOP-06"], tags=["failure", "retry"], personas=["clerk"])
def crash_between_steps(s):
clerk = s.as_("clerk")
body = order(s)
s.given("the app dies right after it stored the order, before it asked stock")
s.fault.crash_after("scn_orders.application.use_cases:_store_new")
s.when("a clerk places an order")
clerk.post("/orders/orders", body).status(500)
s.then("the order is there, unreserved, and stock has nothing")
clerk.get(f"/orders/orders/{body['order_id']}").ok().field("reserved", False)
stock_of(s, body).ok().field("count", 0)
s.when("the clerk's retry arrives")
clerk.post("/orders/orders", body).ok().field("reserved", True)
s.then("one order, one reservation")
stock_of(s, body).ok().field("count", 1)
ids = clerk.get("/orders/orders").ok()["orders"]
s.check(ids.count(body["order_id"]) == 1, "the order is listed once", ids)
@CAT.scenario("SC-SHOP-07", "no mail provider: the confirmation becomes a task carrying exactly what would be sent",
stories=["SHOP-07"], tags=["failure", "capability"], personas=["clerk"])
def no_provider(s):
clerk = s.as_("clerk")
body = order(s, customer="Bergstrom")
clerk.post("/orders/orders", body).ok()
s.given("the composition has no provider for mail.send")
s.fault.capability("mail.send", "absent")
s.when("the clerk sends the confirmation")
r = clerk.post(f"/orders/orders/{body['order_id']}/confirmation")
s.then("a manual task with the exact input and the plan in words")
r.ok().field("delivery", "manual").field("task.capability", "mail.send") \
.field("task.input.to", ["bergstrom@customer.example"]).field("task.input.subject", f"Order {body['order_id']}") \
.field("task.input.body_text", f"Your order {body['order_id']}: 99.99 EUR.") \
.matches("task.plan", lambda p: "bergstrom@customer.example" in p, "a plan naming the recipient")
s.then("sending again is the same task, not a second one")
clerk.post(f"/orders/orders/{body['order_id']}/confirmation").same_as(r)
tasks = [t for t in clerk.get("/orders/tasks").ok()["tasks"] if t["order_id"] == body["order_id"]]
s.check(len(tasks) == 1, "one task for the order", tasks)
@CAT.scenario("SC-SHOP-08", "a failing mail provider is retryable; one needing a connection hands the step to a person",
stories=["SHOP-07"], tags=["failure", "capability"], personas=["clerk", "manager"])
def failing_provider(s):
clerk, manager = s.as_("clerk"), s.as_("manager")
first, second = order(s, customer="Failco"), order(s, customer="Nocon")
for b in (first, second):
clerk.post("/orders/orders", b).ok()
with s.fault.scope() as f:
s.given("the mail provider's vendor is not answering")
f.capability("mail.send", "failing")
s.when("the clerk sends a confirmation")
s.then("it failed, retryably — the order is fine, the step is not done")
clerk.post(f"/orders/orders/{first['order_id']}/confirmation").ok() \
.field("delivery", "failed").field("retryable", True)
s.when("the vendor is back and the clerk retries")
clerk.post(f"/orders/orders/{first['order_id']}/confirmation").ok().field("delivery", "sent")
manager.get("/mailer/outbox", to="failco@customer.example").ok().length("messages", 1)
with s.fault.scope() as f:
s.given("the provider has no connection to the mail system")
f.capability("mail.send", "connection_required")
s.then("the step is a task for a person, at once, naming what is missing")
clerk.post(f"/orders/orders/{second['order_id']}/confirmation").ok().field("delivery", "manual") \
.matches("task.reason", lambda v: "connection" in v, "a reason naming the connection")
s.when("the connection exists and the confirmation is asked for again")
s.then("the person's task stands — nothing is sent a second time behind their back")
clerk.post(f"/orders/orders/{second['order_id']}/confirmation").ok().field("delivery", "manual")
manager.get("/mailer/outbox", to="nocon@customer.example").ok().length("messages", 0)
@CAT.scenario("SC-SHOP-09", "the database says 'not now': retried where safe, surfaced where the commit may have landed",
stories=["SHOP-08"], tags=["failure", "database"], personas=["clerk", "manager"])
def transient_database(s):
clerk, manager = s.as_("clerk"), s.as_("manager")
body = order(s)
s.given("two serialization failures (40001) are waiting for the next order")
hit = s.fault.db_error("40001", use_case="place_order", times=2)
s.then("the clerk never sees them: one order")
clerk.post("/orders/orders", body).ok()
s.check(hit["injected"] == 2, "both failures were injected and absorbed", hit)
v = clerk.get(f"/orders/orders/{body['order_id']}").ok()["version"]
s.given("the connection drops after the note's commit reached the database (08006)")
s.fault.db_error("08006", use_case="update_note", after=True)
s.when("the clerk changes the note")
s.then("a non-idempotent command is NOT re-run: the clerk sees the failure…")
clerk.post(f"/orders/orders/{body['order_id']}/note", {"note": "call first", "expected_version": v}).status(500)
s.then("…and on reload the change is there, once")
clerk.get(f"/orders/orders/{body['order_id']}").ok().field("note", "call first").field("version", v + 1)
s.given("the same drop on approving, which is idempotent")
s.fault.db_error("08006", use_case="approve_order", after=True)
s.then("it is retried, and the manager sees it approved")
manager.post(f"/orders/orders/{body['order_id']}/approve").ok().field("status", "approved")
@CAT.scenario("SC-SHOP-10", "five people save one order at once: one wins, the rest are told it changed",
stories=["SHOP-09"], tags=["concurrency"], personas=["clerk", "manager"])
def concurrent_save(s):
clerk = s.as_("clerk")
body = order(s)
v = clerk.post("/orders/orders", body).ok()["version"]
s.given(f"everyone opened version {v}")
s.when("five saves of a different note land at the same moment")
rs = s.concurrently(5, lambda i: (clerk if i % 2 else s.as_("manager")).post(
f"/orders/orders/{body['order_id']}/note", {"note": f"note {i}", "expected_version": v}))
s.then("one is saved; four are refused 409 'changed since you opened it'")
rs.expect({200: 1, 409: 4})
for r in rs.where(409):
r.refused(409, "changed since you opened it")
winner = rs.where(200)[0]["note"]
after = clerk.get(f"/orders/orders/{body['order_id']}").ok().field("note", winner).field("version", v + 1)
s.then("a refused person reloads and saves: a normal change")
clerk.post(f"/orders/orders/{body['order_id']}/note", {"note": "mine", "expected_version": after["version"]}) \
.ok().field("version", v + 2)
@CAT.scenario("SC-SHOP-11", "the customer's approval link acts once; a replaced link is dead",
stories=["SHOP-10"], tags=["public", "concurrency"], personas=["clerk"])
def public_link_once(s):
clerk = s.as_("clerk")
body = order(s)
clerk.post("/orders/orders", body).ok()
s.given("the clerk makes an approval link for the customer")
buyer = s.email("buyer")
issued = clerk.post(f"/orders/orders/{body['order_id']}/approval-link", {"customer_email": buyer}).ok()
link = s.link(issued["url"], "the customer")
s.when("the customer opens it")
page = link.open().status(200)
s.check("99.99" in page.text, "the page shows the total", page.text[:200])
s.when("they press Approve three times at once")
rs = s.concurrently(3, lambda i: link.submit())
s.then("it acts once; the other presses meet the dead-link page")
rs.expect({200: 1, 404: 2})
clerk.get(f"/orders/orders/{body['order_id']}").ok().field("status", "customer_approved")
s.when("the clerk makes a new link for the same customer")
fresh = clerk.post(f"/orders/orders/{body['order_id']}/approval-link", {"customer_email": buyer}).ok()
s.then("the old link is dead — the same page as any unknown link — and the new one opens")
dead = link.open().status(404)
s.check(dead.text == s.link(issued["url"][:-43] + "x" * 43).open().status(404).text,
"a replaced link and an unknown one are byte for byte the same page")
s.link(fresh["url"]).open().status(200)
@CAT.scenario("SC-SHOP-12", "another clerk's order does not exist for them; a viewer may not place one",
stories=["SHOP-11"], tags=["permissions"], personas=["clerk", "clerk#2", "viewer", "manager", "anonymous"])
def visibility(s):
owner, other = s.as_("clerk"), s.as_("clerk", 2)
body = order(s)
owner.post("/orders/orders", body).ok()
oid = body["order_id"]
s.then("another clerk gets 404 — not 403 — through both doors, and does not see it listed")
other.get(f"/orders/orders/{oid}").not_found()
other.tool("get_order", order_id=oid).not_found()
s.check(oid not in other.get("/orders/orders").ok()["orders"], "not in the other clerk's list")
s.then("a viewer is refused placing an order, naming the permission, through both doors")
viewer = s.as_("viewer")
viewer.post("/orders/orders", order(s)).forbidden("orders:order:write")
viewer.tool("place_order", order(s)).forbidden("orders:order:write")
s.then("nobody signed in gets 401; the manager sees the order, owned by the first clerk")
s.anonymous.get(f"/orders/orders/{oid}").unauthorized()
mine = owner.get(f"/orders/orders/{oid}").ok()["owner"]
s.as_("manager").get(f"/orders/orders/{oid}").ok().field("owner", mine)
@CAT.scenario("SC-SHOP-13", "placing the same order six times at once, through both doors, makes one order",
stories=["SHOP-06"], tags=["concurrency", "idempotency"], personas=["clerk"])
def double_submit(s):
clerk = s.as_("clerk")
body = order(s)
s.when("the same order arrives six times at once — three by the API, three by an agent")
rs = s.concurrently(6, lambda i: clerk.post("/orders/orders", body) if i % 2 else clerk.tool("place_order", body))
s.then("every one answers the order, reserved; there is one order and one reservation")
rs.expect({200: 6})
for r in rs:
r.field("order_id", body["order_id"]).field("reserved", True)
stock_of(s, body).ok().field("count", 1)
ids = clerk.get("/orders/orders").ok()["orders"]
s.check(ids.count(body["order_id"]) == 1, "the order is listed once", ids)
# ── generated probes ──────────────────────────────────────────────────────────────────────
def _placed(s, who="clerk"):
body = order(s)
return s.as_(who).post("/orders/orders", body).ok()
RECIPES = {
"place_order": lambda s: ("clerk", order(s),
lambda s, r: s.as_("clerk").get(f"/orders/orders/{r['order_id']}")),
"approve_order": lambda s: ("manager", {"order_id": _placed(s)["order_id"]}),
"send_confirmation": lambda s: ("clerk", {"order_id": _placed(s)["order_id"]}),
"reserve": lambda s: ("warehouse", {"reservation_key": s.key("R"), "sku": s.key("SKU"), "quantity": 2}),
"update_note": lambda s: (lambda o: ("clerk", {"order_id": o["order_id"], "note": "n",
"expected_version": o["version"]},
lambda s, r: s.as_("clerk").get(f"/orders/orders/{o['order_id']}")))(_placed(s)),
"request_customer_approval": lambda s: ("clerk", {"order_id": _placed(s)["order_id"],
"customer_email": s.email("buyer")}),
}
CAT.permission_matrix(
rows=Rows(make=lambda s: {"order_id": _placed(s)["order_id"]}, owner="clerk", other="clerk#2"),
exceptions={})
CAT.idempotency(recipes=RECIPES, exceptions={
"tick": "answers what it changed THIS time, so a replay's answer is [] by design; its effect is SC-SHOP-03's"})
CAT.mass_assignment(recipes=RECIPES)