"""Probes in the pattern erp-compose scaffold 0.9.1 generates: WHO may run each command, in WHICH state, with WHICH side
effect — from module.yaml workflows (from/to/by/terminal), actors' permissions and seams.
A plain test is what the scaffolded code already enforces and must stay green. An xfail(strict) is a rule the
brick has to write — a role the workflow does not name, a terminal parent, a separation of duties, a seam
that must fire: CI lists it until the rule exists, then the mark must go. A case the business really allows
is taken OUT of the tables below with its reason, never left xfailing. The tables are the spec, not the code:
edit them when the spec changes (or re-scaffold), never to make a probe pass.
"""
import json
import pathlib
import pytest
import yaml
from conftest import api, person
from erp_platform import seam
from brick_projects.application import use_cases
from brick_projects.domain.item import Item
from brick_projects.domain.project import Project
CLASSES = {"project": Project, "item": Item}
PLURAL = {'project': 'projects', 'item': 'items'}
STATE_FIELD = {'project': 'state', 'item': 'state'}
KEYS = {'project': {'project_id': 'PROJECT-ID-1'}, 'item': {'item_id': 'ITEM-ID-1'}}
STATES = {'project': ['planned', 'active', 'closed', 'cancelled'], 'item': ['planned', 'in_progress', 'blocked', 'in_review', 'done', 'cancelled']}
TERMINAL = {'project': ['closed', 'cancelled'], 'item': ['done', 'cancelled']}
OWNED = ['item', 'project'] # roots with an owner: row visibility by owner
PARENTS = {'item': ('project', 'project_id')} # root -> (parent root, its key on the child)
ROLES = yaml.safe_load((pathlib.Path(__file__).resolve().parents[1] / "src" / "brick_projects" / "roles.yaml")
.read_text())["roles"] # the brick's default roles, as shipped
# Which of the spec's actors (the lanes the workflow names in `by`) each default role fills.
ACTORS_OF = {"project_manager": ["manager", "project_lead"], "project_lead": ["project_lead"],
"team_member": ["contributor"], "billing": ["finance"], "viewer": [], "scheduler": ["system"]}
SEE_ALL = [r for r, perms in ROLES.items() if "pj:project:read_all" in perms] # they see every owner's rows
# Outbound seams (module.yaml seams) -> the transitions that may fire them. TODO(rule): derived from each
# seam story's trigger words — correct it to the spec; the probes then hold the code to it.
SEAMS = {'so': ['accept', 'accept_deemed'], 'ar': ['accept', 'accept_deemed'], 'crm': []}
# Separation of duties: (the request, the approval) — the person who did the first may not do the second.
SOD = [('submit', 'accept'), ('submit', 'accept_deemed')]
COMMANDS = [
{'name': 'reconcile', 'root': 'project', 'route': 'POST /projects/{project_id}/reconcile', 'permission': 'pj:project:reconcile', 'frm': ['active'], 'to': ['closed'], 'by': ['system'], 'body': {}, 'loop': False, 'idempotent': True},
{'name': 'cancel_project', 'root': 'project', 'route': 'POST /projects/{project_id}/cancel', 'permission': 'pj:project:write', 'frm': ['planned', 'active'], 'to': ['cancelled'], 'by': ['project_lead'], 'body': {'reason': 'because'}, 'loop': False, 'idempotent': True},
{'name': 'start', 'root': 'item', 'route': 'POST /items/{item_id}/start', 'permission': 'pj:item:work', 'frm': ['planned'], 'to': ['in_progress'], 'by': ['contributor'], 'body': {}, 'loop': False, 'idempotent': True},
{'name': 'block', 'root': 'item', 'route': 'POST /items/{item_id}/block', 'permission': 'pj:item:work', 'frm': ['in_progress'], 'to': ['blocked'], 'by': ['contributor'], 'body': {'reason': 'because'}, 'loop': False, 'idempotent': True},
{'name': 'unblock', 'root': 'item', 'route': 'POST /items/{item_id}/unblock', 'permission': 'pj:item:work', 'frm': ['blocked'], 'to': ['in_progress'], 'by': ['contributor'], 'body': {}, 'loop': False, 'idempotent': True},
{'name': 'submit', 'root': 'item', 'route': 'POST /items/{item_id}/submit', 'permission': 'pj:item:work', 'frm': ['in_progress'], 'to': ['in_review'], 'by': ['contributor'], 'body': {}, 'loop': False, 'idempotent': True},
{'name': 'send_warning', 'root': 'item', 'route': 'POST /items/{item_id}/send-warning', 'permission': 'pj:item:write', 'frm': ['in_review'], 'to': ['in_review'], 'by': ['project_lead'], 'body': {}, 'loop': True, 'idempotent': False},
{'name': 'request_changes', 'root': 'item', 'route': 'POST /items/{item_id}/request-changes', 'permission': 'pj:item:write', 'frm': ['in_review'], 'to': ['in_progress'], 'by': ['project_lead'], 'body': {'reason': 'because'}, 'loop': False, 'idempotent': True},
{'name': 'accept', 'root': 'item', 'route': 'POST /items/{item_id}/accept', 'permission': 'pj:item:accept', 'frm': ['in_review'], 'to': ['done'], 'by': ['project_lead', 'manager'], 'body': {}, 'loop': False, 'idempotent': True},
{'name': 'accept_deemed', 'root': 'item', 'route': 'POST /items/{item_id}/accept-deemed', 'permission': 'pj:item:accept', 'frm': ['in_review'], 'to': ['done'], 'by': ['project_lead', 'manager'], 'body': {}, 'loop': False, 'idempotent': True},
{'name': 'descope', 'root': 'item', 'route': 'POST /items/{item_id}/descope', 'permission': 'pj:item:write', 'frm': ['planned', 'in_progress', 'blocked', 'in_review'], 'to': ['cancelled'], 'by': ['project_lead'], 'body': {'reason': 'because'}, 'loop': False, 'idempotent': True},
]
SUSPECTS = ['owner', 'created_at', 'created_by', 'approved_by', 'approved_at', 'roles', 'permissions', 'is_admin', 'tenant']
BY = {c["name"]: c for c in COMMANDS}
REFUSED = (403, 404, 409)
def who(role, email=None):
return person(email or f"{role}@shop.example", ROLES.get(role, []), roles=[role])
ALL_PERMS = sorted({p for perms in ROLES.values() for p in perms} | {c["permission"] for c in COMMANDS})
BOSS = person("boss@shop.example", ALL_PERMS, roles=["project_manager"]) # holds every permission
def seed(root, state, owner="boss@shop.example", parent_state="active"):
"""A row of `root` in `state` (and its parent in `parent_state`), written through the repository, with what
its gates need: evidence and a submission long past its review window, a warning long past its notice."""
from datetime import datetime
if root in PARENTS and parent_state:
seed(PARENTS[root][0], parent_state, owner)
agg = CLASSES[root](**KEYS[root])
setattr(agg, STATE_FIELD[root], state)
if root == "item":
agg.evidence, agg.evidence_kind = "https://client.example/proof", "link"
agg.submitted_at, agg.warned_at = datetime(2026, 9, 1, 9), (
datetime(2026, 9, 20, 9) if state == "in_review" else None)
agg.billing_kind, agg.billing_plan, agg.line_id = "one_off", "single", "L1"
agg.source_kind, agg.source_id = "sales_order", "SO-1"
if root in OWNED:
agg.owner = owner
if root in PARENTS:
parent, fk = PARENTS[root]
setattr(agg, fk, next(iter(KEYS[parent].values())))
with use_cases.repository() as repo:
getattr(repo, PLURAL[root]).save(agg, 0)
def state_of(root):
with use_cases.repository() as repo:
found = getattr(repo, PLURAL[root]).get(*KEYS[root].values())
return getattr(found[0], STATE_FIELD[root])
def run(c, headers, extra=None):
path = c["route"].split(" ", 1)[1]
for k, v in KEYS[c["root"]].items():
path = path.replace("{" + k + "}", str(v))
return api("POST", path, headers, {**c["body"], **(extra or {})})
@pytest.fixture
def seams(monkeypatch):
"""Every seam call any code makes, recorded (and answered ok) — wherever the call is written."""
calls = []
def record(self, path, *a, **kw):
calls.append((self.name, path))
return seam.Reply(True, 200, data={})
monkeypatch.setattr(seam.Seam, "post", record)
monkeypatch.setattr(seam.Seam, "get", record)
monkeypatch.setattr(seam.Seam, "installed", property(lambda self: True)) # the neighbours are composed
return calls
def todo(reason):
return pytest.mark.xfail(strict=True, raises=AssertionError, reason=reason)
def cases(rows):
return [pytest.param(*r[:-1], marks=r[-1], id="-".join(map(str, r[:-1]))) if r[-1] else
pytest.param(*r[:-1], id="-".join(map(str, r[:-1]))) for r in rows]
# ── the permission matrix: every role × every command ─────────────────────────────────────
def _matrix():
rows = []
for c in COMMANDS:
for role, perms in ROLES.items():
holds, named = c["permission"] in perms, any(a in c["by"] for a in ACTORS_OF[role])
if not holds:
rows.append((c["name"], role, "refused", None))
elif not named:
rows.append((c["name"], role, "refused", None)) # the role check in the use case
else:
rows.append((c["name"], role, "served", None))
for role in [r for r in ROLES if any(a in c["by"] for a in ACTORS_OF[r])]:
if c["permission"] not in ROLES[role]:
rows.append((c["name"], role, "served", todo(
f"TODO(permission): the workflow gives {c['name']} to {role}, whose permissions lack "
f"{c['permission']} — fix the spec's actors")))
return rows
@pytest.mark.parametrize("name,role,expect", cases(_matrix()))
def test_the_permission_matrix(name, role, expect, seams):
c = BY[name]
me = f"{role}@shop.example"
seed(c["root"], c["frm"][0], owner=me)
status, body = run(c, who(role, me))
if expect == "refused":
assert status in REFUSED, (status, body)
assert state_of(c["root"]) == c["frm"][0] and seams == []
else:
assert status == 200, (status, body)
@pytest.mark.parametrize("name", [c["name"] for c in COMMANDS if c["root"] in OWNED])
def test_another_person_of_the_same_role_gets_404(name):
c = BY[name]
role = next((r for r in ROLES if any(a in c["by"] for a in ACTORS_OF[r]) and r not in SEE_ALL
and c["permission"] in ROLES[r]), None)
if role is None:
pytest.skip("no person role is given this command with its permission")
seed(c["root"], c["frm"][0], owner=f"{role}@shop.example")
status, _ = run(c, who(role, f"another.{role}@shop.example"))
assert status == 404 # never 403: a 403 would confirm the record exists
assert state_of(c["root"]) == c["frm"][0]
@pytest.mark.parametrize("request_step,approval", [tuple(x) for x in SOD] or
[pytest.param(None, None, marks=pytest.mark.skip(reason="no approval in the workflows"))])
def test_the_requester_may_not_approve(request_step, approval):
first, second = BY[request_step], BY[approval]
seed(first["root"], first["frm"][0])
with use_cases.repository() as repo: # submitted now: the warning is not sent yet
it, v = repo.items.get("ITEM-ID-1")
it.warned_at = None
repo.items.save(it, v)
assert run(first, BOSS)[0] == 200
status, body = run(second, BOSS)
assert status in REFUSED, (status, body)
# ── state guards: every command in every state its workflow does not allow ─────────────────
def _wrong_states():
rows = []
for c in COMMANDS:
ok = set(c["frm"]) | (set() if c["loop"] else set(c["to"]))
rows += [(c["name"], s, None) for s in STATES[c["root"]] if s not in ok]
return rows
@pytest.mark.parametrize("name,state", cases(_wrong_states()))
def test_a_command_in_a_state_its_workflow_does_not_allow_is_refused(name, state, seams):
c = BY[name]
seed(c["root"], state)
status, body = run(c, BOSS)
assert status == 409, (status, body)
assert state_of(c["root"]) == state and seams == []
def _under_terminal_parent():
rows = []
for c in COMMANDS:
if c["root"] in PARENTS:
parent = PARENTS[c["root"]][0]
rows += [(c["name"], s, None) for s in TERMINAL[parent]] # Item.may_move_in
return rows
@pytest.mark.parametrize("name,parent_state", cases(_under_terminal_parent()) or
[pytest.param(None, None, marks=pytest.mark.skip(reason="no parent with a lifecycle"))])
def test_a_command_under_a_terminal_parent_is_refused(name, parent_state, seams):
c = BY[name]
seed(c["root"], c["frm"][0], parent_state=parent_state)
status, body = run(c, BOSS)
assert status in REFUSED, (status, body)
assert state_of(c["root"]) == c["frm"][0] and seams == []
# ── side effects: a seam fires only on the transitions that name it ────────────────────────
@pytest.mark.parametrize("name", [c["name"] for c in COMMANDS])
def test_a_seam_fires_only_on_the_transitions_that_name_it(name, seams):
c = BY[name]
seed(c["root"], c["frm"][0])
assert run(c, BOSS)[0] == 200
wrong = [(brick, path) for brick, path in seams if name not in SEAMS.get(brick, [])]
assert wrong == [], f"{name} called a seam the workflow does not give it"
@pytest.mark.parametrize("brick,name", [(b, n) for b, steps in SEAMS.items() for n in steps] or
[pytest.param(None, None, marks=pytest.mark.skip(reason="no outbound seam"))])
def test_the_transition_that_names_a_seam_calls_it(brick, name, seams):
c = BY[name]
seed(c["root"], c["frm"][0])
assert run(c, BOSS)[0] == 200
assert brick in [b for b, _ in seams] # accept hands over to both: the order, then invoicing
# ── idempotency and mass assignment ────────────────────────────────────────────────────────
@pytest.mark.parametrize("name", [c["name"] for c in COMMANDS if c["idempotent"]])
def test_a_replay_answers_the_same_and_moves_nothing(name, seams):
c = BY[name]
seed(c["root"], c["frm"][0])
first = run(c, BOSS)
assert first[0] == 200, first
calls = list(seams)
assert run(c, BOSS) == first
assert seams == calls # no second side effect
@pytest.mark.parametrize("name", [c["name"] for c in COMMANDS])
def test_fields_it_did_not_ask_for_are_never_applied(name):
c = BY[name]
seed(c["root"], c["frm"][0])
sentinel = "mass-assign-sentinel"
extra = {f: sentinel for f in SUSPECTS + [STATE_FIELD[c["root"]]] if f not in c["body"]}
status, body = run(c, BOSS, extra)
assert status in (200, 422), (status, body)
assert sentinel not in json.dumps(body)
assert state_of(c["root"]) != sentinel