"""pj's scenario catalogue: the spec's scenarios (module.yaml tests.scenarios, SC-PJ-01..07) and the ones the
2026-10-18 audit asked for (SC-PJ-08..11), each mapped to its stories and written ONCE over the composition's doors —
routes, MCP tools, the public link — never a table. Each makes its own sale (s.key), so none depends on another.
A team member does the work and the project's lead records the client's answer: whoever submitted work never
records its acceptance.
erp-scenarios list tests/scenarios/catalogue.py
erp-scenarios run tests/scenarios/catalogue.py
"""
import pathlib
import sys
from urllib.parse import quote
HERE = pathlib.Path(__file__).resolve().parent
sys.path.insert(0, str(HERE)) # fake_orders, the sales-order stand-in
from erp_platform.scenarios import Catalogue # noqa: E402
from erp_platform.scenarios.personas import EMAIL_DOMAIN # noqa: E402
ROLES = HERE.parents[1] / "src" / "brick_projects" / "roles.yaml" # the brick's default roles, as shipped
CAT = Catalogue("pj", compose=HERE / "compose.yaml", roles=ROLES, clock="pj", zone="UTC",
storage={"so": lambda engine: None})
WORKER = f"team_member.1@{EMAIL_DOMAIN}" # the in-process identity of the "team_member" persona
def q(key: str) -> str:
return quote(key, safe="")
def sale(s, lines, kind="contract", **kw):
return {"source_kind": kind, "source_id": s.key("C" if kind == "contract" else "SO"), "bp_id": "7311",
"lines": lines, "starts_on": "2026-11-01", "ends_on": "2026-12-31", **kw}
PHASES = {"line_id": "L1", "title": "Implementation phase", "quantity": 2, "billing_kind": "one_off",
"billing_plan": "single"}
HOSTING = {"line_id": "L2", "title": "Hosting month", "quantity": 12, "billing_kind": "recurring",
"billing_plan": "periodic"}
MAINTENANCE = {"line_id": "L3", "title": "Maintenance month", "quantity": 5, "billing_kind": "recurring",
"billing_plan": "periodic"}
def item_of(body, line="L1", n=1):
key = "pj:{}:{}|{}|{}".format(body["source_kind"], body["source_id"], line, n)
return f"/pj/items/{q(key)}"
def worked(s, lead, worker, item, evidence="https://staging.client.example/phase-1"):
"""The lead gives the item to the contributor, who does it and submits it with its evidence."""
lead.post(f"{item}/assign", {"owner": WORKER}).ok()
worker.post(f"{item}/start", {}).ok()
worker.post(f"{item}/submit", {"evidence": evidence} if evidence else {}).ok()
@CAT.scenario("SC-PJ-01", "a mixed sale plans only what is delivered", stories=["PJ-01", "PJ-20"],
tags=["doors", "idempotency"], personas=["project_lead"])
def mixed_sale(s):
lead = s.as_("project_lead")
body = sale(s, [PHASES, HOSTING, MAINTENANCE])
s.given("an active contract: 2 x implementation phase (one-off, single), 12 x hosting month (recurring), "
"5 x maintenance month (recurring)")
s.when("plan_from_sale is called twice")
first = lead.post("/pj/plans", body).ok()
again = lead.tool("plan_from_contract", **body).ok()
s.then("one project and 2 items; the second call creates nothing")
first.field("created", 2).matches("project.items", lambda v: [i["line_id"] for i in v] == ["L1", "L1"],
"two items, both on the implementation line")
again.field("created", 0)
@CAT.scenario("SC-PJ-02", "one implementation paid in instalments is one item", stories=["PJ-01"],
tags=["money"], personas=["project_lead", "team_member"])
def instalments(s):
lead, worker = s.as_("project_lead"), s.as_("team_member")
s.given("1 x implementation sold on 6 instalments (one-off, periodic), with the default rule 'one item'")
body = sale(s, [{"line_id": "L4", "title": "Implementation", "quantity": 1, "billing_kind": "one_off",
"billing_plan": "periodic"}])
lead.post("/pj/plans", body).ok().field("created", 1)
s.when("it is worked and accepted")
item = item_of(body, "L4")
worked(s, lead, worker, item, "https://client.example/live")
s.then("one item; acceptance bills nothing here, because the instalments are invoicing's schedule")
lead.post(f"{item}/accept", {}).ok().field("state", "done").field("billing_status", "not_billed")
@CAT.scenario("SC-PJ-03", "done without proof is refused", stories=["PJ-04"], personas=["project_lead", "team_member"])
def no_proof(s):
lead, worker = s.as_("project_lead"), s.as_("team_member")
s.given("an item in review with no evidence")
body = sale(s, [PHASES])
lead.post("/pj/plans", body).ok()
worked(s, lead, worker, item_of(body), evidence=None)
s.when("the lead records acceptance")
r = lead.post(f"{item_of(body)}/accept", {})
s.then('"Attach the evidence first: a link, a file or a sign-off"')
r.status(409).field("detail", "Attach the evidence first: a link, a file or a sign-off")
@CAT.scenario("SC-PJ-04", "no answer, warned, deemed, billed once", stories=["PJ-03", "PJ-11", "PJ-22"],
tags=["clock"], personas=["project_lead", "team_member"])
def deemed(s):
lead, worker = s.as_("project_lead"), s.as_("team_member")
s.given("an item from a one-off, single-payment line, submitted Monday 2 November 2026, a 10-day window "
"and a 5-day notice")
s.clock.set("2026-11-02T09:00 UTC")
body = sale(s, [PHASES])
lead.post("/pj/plans", body).ok()
item = item_of(body)
worked(s, lead, worker, item)
s.when("the window passes with no answer")
s.clock.set("2026-11-17T09:00 UTC")
lead.get(item).ok().field("review_clock.warning_due", True)
s.when("the lead sends the warning, and its notice passes")
lead.post(f"{item}/send-warning", {}).ok()
lead.post(f"{item}/accept-deemed", {}).status(409)
s.clock.set("2026-11-24T09:00 UTC")
s.then("deemed acceptance is recorded as deemed; recording it again changes nothing")
lead.post(f"{item}/accept-deemed", {}).ok().field("approved_how", "deemed").field("state", "done")
lead.post(f"{item}/accept-deemed", {}).ok().field("approved_how", "deemed")
s.then("with no invoicing in this composition the draft is left pending, for one later request (acc:<item>)")
lead.get(item).ok().field("billing_status", "pending")
@CAT.scenario("SC-PJ-05", "waiting on the client is not our slippage", stories=["PJ-02"], tags=["clock"],
personas=["project_lead", "team_member"])
def blocked(s):
lead, worker = s.as_("project_lead"), s.as_("team_member")
s.given("an item due in 10 days")
s.clock.set("2026-11-02T09:00 UTC")
body = sale(s, [PHASES])
lead.post("/pj/plans", body).ok()
item = item_of(body)
lead.post(f"{item}/assign", {"due_on": "2026-11-12", "owner": WORKER}).ok()
worker.post(f"{item}/start", {}).ok()
s.when("it is blocked for 4 days waiting for the brief, then unblocked")
worker.post(f"{item}/block", {"reason": "waiting for brief"}).ok().field("blocked_since", "2026-11-02")
s.clock.set("2026-11-06T09:00 UTC")
worker.post(f"{item}/unblock", {}).ok().field("blocked_days", 4)
s.then("on 16 November it is 4 days past due, none of them ours")
s.clock.set("2026-11-16T09:00 UTC")
lead.get(item).ok().field("slippage_days", 0)
@CAT.scenario("SC-PJ-06", "a short window raises one makegood", stories=["PJ-06", "PJ-07", "PJ-12"],
tags=["clock", "idempotency"], personas=["project_lead", "team_member", "scheduler"])
def short_window(s):
lead, worker = s.as_("project_lead"), s.as_("team_member")
s.given("6 sold, 5 accepted when the window closes")
s.clock.set("2026-11-02T09:00 UTC")
body = sale(s, [dict(PHASES, quantity=6, title="Article")])
lead.post("/pj/plans", body).ok()
pid = f"pj:contract:{body['source_id']}"
for n in range(1, 7):
item = item_of(body, n=n)
lead.post(f"{item}/assign", {"owner": WORKER}).ok()
worker.post(f"{item}/start", {}).ok()
if n <= 5:
worker.post(f"{item}/submit", {"evidence": f"https://client.example/article-{n}"}).ok()
lead.post(f"{item}/accept", {}).ok()
s.when("the project is reconciled twice, then the lead raises a makegood twice")
s.clock.set("2027-01-01T06:00 UTC")
scheduler = s.as_("scheduler")
scheduler.post(f"/pj/projects/{q(pid)}/reconcile", {}).ok().field("state", "closed")
closed = scheduler.post(f"/pj/projects/{q(pid)}/reconcile", {}).ok()
s.then("one reconciliation (short 1); one makegood with no sold line; the second is refused")
closed.field("reconciliation.lines.0.short", 1).field("reconciliation.lines.0.delivered", 5)
short = item_of(body, n=6)
lead.post(f"{short}/makegood", {}).ok().field("source", "makegood").field("line_id", "")
lead.post(f"{short}/makegood", {}).status(409)
@CAT.scenario("SC-PJ-07", "the client link sees one project", stories=["PJ-05"], tags=["public"],
personas=["project_lead"])
def client_link(s):
lead = s.as_("project_lead")
s.given("links for projects A and B")
a, b = sale(s, [PHASES]), sale(s, [dict(PHASES, title="Campaign flight")])
lead.post("/pj/plans", a).ok()
lead.post("/pj/plans", b).ok()
link = lead.post(f"/pj/projects/{q('pj:contract:' + a['source_id'])}/client-link",
{"email": s.email("client")}).ok()
s.when("A's link is opened")
page = s.link(link.body["url"]).open()
s.then("it shows A's work, read-only, and nothing of B's")
page.ok()
s.check("Implementation phase" in page.text and "Campaign flight" not in page.text,
"only project A's items are on the page", page.text[:300])
# ── what the audit asked for ────────────────────────────────────────────────────────────────
@CAT.scenario("SC-PJ-08", "a lead with one item in another lead's project has no say over that project",
stories=["PJ-04", "PJ-12"], tags=["permissions"],
personas=["project_lead", "project_lead#2", "team_member"])
def no_say_over_anothers_project(s):
lena, leo, worker = s.as_("project_lead"), s.as_("project_lead", 2), s.as_("team_member")
s.given("Lena's project, one of whose items is given to Leo, who leads projects too; a team member does the work")
body = sale(s, [PHASES])
lena.post("/pj/plans", body).ok()
pid, item = q(f"pj:contract:{body['source_id']}"), item_of(body)
lena.post(f"{item}/assign", {"owner": f"project_lead.2@{EMAIL_DOMAIN}"}).ok()
lena.post(f"{item}/assign", {"owner": WORKER}).ok()
worker.post(f"{item}/start", {}).ok()
worker.post(f"{item}/submit", {"evidence": "https://client.example/1"}).ok()
lena.post(f"{item}/assign", {"owner": f"project_lead.2@{EMAIL_DOMAIN}"}).ok()
s.when("Leo, holding a lead's permissions, tries to cancel or share Lena's project and record the client's answer")
s.then("each is refused: being given an item is not leading its project")
leo.post(f"/pj/projects/{pid}/cancel", {"reason": "mine"}).status(403)
leo.post(f"/pj/projects/{pid}/client-link", {"email": s.email("client")}).status(403)
leo.post(f"{item}/accept", {}).status(403)
s.then("Lena records the client's acceptance")
lena.post(f"{item}/accept", {}).ok().field("approved_by", f"project_lead.1@{EMAIL_DOMAIN}")
@CAT.scenario("SC-PJ-09", "a cancelled project's work does not move or bill, and a short item is made good, not "
"finished", stories=["PJ-07", "PJ-22"], tags=["money"],
personas=["project_lead", "team_member", "scheduler"])
def frozen_projects(s):
lead, worker = s.as_("project_lead"), s.as_("team_member")
s.given("work in review on a project the client withdrew from")
gone = sale(s, [PHASES])
lead.post("/pj/plans", gone).ok()
worked(s, lead, worker, item_of(gone))
lead.post(f"/pj/projects/{q('pj:contract:' + gone['source_id'])}/cancel", {"reason": "client withdrew"}).ok()
s.then("its acceptance is refused and nothing is billed")
lead.post(f"{item_of(gone)}/accept", {}).status(409)
lead.get(item_of(gone)).ok().field("billing_status", "not_billed")
s.given("a closed project whose item was left short and has been made good")
s.clock.set("2026-11-02T09:00 UTC")
short = sale(s, [PHASES])
lead.post("/pj/plans", short).ok()
worked(s, lead, worker, item_of(short, n=2))
s.clock.set("2027-01-01T06:00 UTC")
s.as_("scheduler").post(f"/pj/projects/{q('pj:contract:' + short['source_id'])}/reconcile", {}).ok()
lead.post(f"{item_of(short, n=2)}/makegood", {}).ok()
s.then("the short item cannot be accepted after all: it would be delivered and billed twice")
lead.post(f"{item_of(short, n=2)}/accept", {}).status(409)
lead.get(item_of(short, n=2)).ok().field("billing_status", "not_billed")
ORDER_LINE = {"line_id": "SO-L1", "title": "Installation", "quantity": 2, "billing_kind": "one_off",
"billing_plan": "single"}
@CAT.scenario("SC-PJ-10", "the order hears about accepted work only, once, with its line",
stories=["PJ-21"], tags=["seam", "idempotency"], personas=["project_lead", "team_member"])
def order_told_once(s):
lead, worker = s.as_("project_lead"), s.as_("team_member")
s.given("a confirmed order of 2 installations, planned as 2 items")
body = sale(s, [ORDER_LINE], kind="sales_order")
lead.post("/pj/plans", body).ok()
first, second = item_of(body, "SO-L1", 1), item_of(body, "SO-L1", 2)
s.when("a retry is asked for work not yet accepted")
lead.post(f"{second}/retry-handover", {}).ok().field("delivery_status", "not_reported")
s.when("the first installation is accepted, and the handover retried")
worked(s, lead, worker, first)
lead.post(f"{first}/accept", {}).ok().field("delivery_status", "reported")
lead.post(f"{first}/retry-handover", {}).ok().field("delivery_status", "reported")
s.then("the order heard of exactly one delivery, on its own line")
lead.get(f"/so/orders/{body['source_id']}/deliveries").ok().matches(
"deliveries", lambda v: [(d["line_no"], d["quantity"]) for d in v] == [("SO-L1", 1)], "one, on SO-L1")
@CAT.scenario("SC-PJ-11", "the order is down when work is accepted: the news waits, then arrives once",
stories=["PJ-21"], tags=["seam", "failure"], personas=["project_lead", "team_member"])
def order_down(s):
lead, worker = s.as_("project_lead"), s.as_("team_member")
body = sale(s, [ORDER_LINE], kind="sales_order")
lead.post("/pj/plans", body).ok()
first = item_of(body, "SO-L1", 1)
worked(s, lead, worker, first)
with s.fault.scope() as f:
s.given("the sales-order brick is unreachable")
f.seam_down("so")
s.when("the installation is accepted")
s.then("it is accepted, and the delivery is pending — not lost, not an error")
lead.post(f"{first}/accept", {}).ok().field("state", "done").field("delivery_status", "pending")
s.when("the order is back and the handover is retried twice")
lead.post(f"{first}/retry-handover", {}).ok().field("delivery_status", "reported")
lead.post(f"{first}/retry-handover", {}).ok().field("delivery_status", "reported")
s.then("the order heard of it once")
lead.get(f"/so/orders/{body['source_id']}/deliveries").ok().matches(
"deliveries", lambda v: len(v) == 1, "one delivery")