dodil/erp-brick-consolidation
DODIL ERP brick: consolidation
tests/test_probes.py
"""Generated probes (erp-compose scaffold 0.9.2): 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 pytest
from conftest import api, person
from erp_platform import auth, seam
from brick_consolidation.application import use_cases
from brick_consolidation.domain.cons_journal import ConsJournal
from brick_consolidation.domain.cons_migration_run import ConsMigrationRun
from brick_consolidation.domain.structure_change import StructureChange
CLASSES = {"structure_change": StructureChange, "cons_journal": ConsJournal, "cons_migration_run": ConsMigrationRun}
PLURAL = {'structure_change': 'structure_changes', 'cons_journal': 'cons_journals', 'cons_migration_run': 'cons_migration_runs'}
STATE_FIELD = {'structure_change': 'status', 'cons_journal': 'status', 'cons_migration_run': 'status'}
KEYS = {'structure_change': {'change_id': 'CHANGE-ID-1'}, 'cons_journal': {'cons_journal_id': 'CONS-JOURNAL-ID-1'}, 'cons_migration_run': {'cons_migration_run_id': 'CONS-MIGRATION-RUN-ID-1'}}
STATES = {'structure_change': ['proposed', 'effective', 'rejected'], 'cons_journal': ['proposed', 'posted', 'rejected', 'reversed'], 'cons_migration_run': ['loaded', 'reconciled', 'signed', 'discarded']}
TERMINAL = {'structure_change': ['effective', 'rejected'], 'cons_journal': ['rejected', 'reversed'], 'cons_migration_run': ['signed', 'discarded']}
OWNED = [] # roots with an owner: row visibility by owner
PARENTS = {} # root -> (parent root, its key on the child)
ROLES = {'group_accountant': ['cons:group:read', 'cons:structure:write', 'cons:data:collect', 'cons:data:upload', 'cons:translate:run', 'cons:eliminate:run', 'cons:inventory:write', 'cons:journal:write', 'cons:journal:reverse', 'cons:period:open', 'cons:report:read', 'cons:settings:write', 'cons:migration:read'], 'group_controller': ['cons:group:read', 'cons:structure:approve', 'cons:data:accept', 'cons:journal:approve', 'cons:period:signoff', 'cons:period:reopen', 'cons:report:read', 'cons:migration:read'], 'auditor': ['cons:group:read', 'cons:report:read', 'cons:migration:read'], 'owner': ['cons:group:read', 'cons:report:read', 'cons:settings:controls', 'cons:migration:read', 'cons:migration:sign'], 'system': ['cons:group:read', 'cons:data:load', 'cons:journal:post_auto', 'cons:carry_forward:derive', 'cons:statement:materialise', 'cons:number_series:claim', 'cons:migration:load', 'cons:migration:reconcile']}
READ_ALL = {} # root -> the permission that sees every owner's rows (never a role)
NS = 'cons'
# 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 = {'gl': []}
# Separation of duties: (the request, the approval) — the person who did the first may not do the second.
SOD = []
COMMANDS = [
{'name': 'reject_cons_entry', 'root': 'cons_journal', 'route': 'POST /cons-journals/{cons_journal_id}/reject-cons-entry', 'permission': 'cons:journal:approve', 'frm': ['proposed'], 'to': ['rejected'], 'by': ['group_controller'], 'body': {'reason': 'because'}, 'loop': False, 'idempotent': True},
# approve_cons_entry and reverse_cons_entry are out of this table (slice B1): both need the entry's group and period (the period not signed,
# the number claimed in the group's book and year), which a bare seeded row has not; their state, second-person and number rules are
# tested in tests/test_slice_b1.py.
]
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):
# the role's PERMISSIONS and no role name: a brick decides by permission, so the probes carry none
return person(email or f"{role}@shop.example", ROLES.get(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)
def seed(root, state, owner="boss@shop.example", parent_state=None):
"""A row of `root` in `state` (and its parent in `parent_state`), written through the repository."""
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 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)
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, role in c["by"]
if not holds:
rows.append((c["name"], role, "refused", None))
elif not named:
rows.append((c["name"], role, "refused", todo(
f"TODO(permission): {role} holds {c['permission']} but the workflow gives {c['name']} to "
f"{', '.join(c['by'])} — a finer permission, or the role check in the use case")))
else:
rows.append((c["name"], role, "served", None))
for role in c["by"]:
if role not in ROLES:
continue
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)
# ── wildcard grants (erp_platform.auth 0.9.2): the gate itself, held to every command ────────────
MACHINE = auth.machine_permissions() # exact grant only: no wildcard reaches them
MALFORMED = ["*", "*:*", "*:*:*:*", f"{NS}:*", f"{NS.upper()}:*:*", f"{NS}:i*:*", " *:*:*", "", 7]
def gate(c):
return {"detail": f"missing permission: {c['permission']}"}
@pytest.mark.parametrize("name", [c["name"] for c in COMMANDS])
def test_every_permission_by_wildcard_passes_every_person_command_and_no_machine_one(name, seams):
c = BY[name]
seed(c["root"], c["frm"][0], owner="wild@shop.example")
status, body = run(c, person("wild@shop.example", ["*:*:*"]))
if c["permission"] in MACHINE:
assert (status, body) == (403, gate(c))
else:
assert status != 401 and body != gate(c), (status, body)
@pytest.mark.parametrize("name", [c["name"] for c in COMMANDS])
def test_a_module_wildcard_is_confined_to_its_module(name, seams):
c = BY[name]
seed(c["root"], c["frm"][0], owner="wild@shop.example")
assert run(c, person("wild@shop.example", ["zz_other:*:*"])) == (403, gate(c))
assert state_of(c["root"]) == c["frm"][0] and seams == []
status, body = run(c, person("wild@shop.example", [f"{NS}:*:*"]))
assert c["permission"] in MACHINE or body != gate(c), (status, body)
@pytest.mark.parametrize("name", [c["name"] for c in COMMANDS])
def test_malformed_grants_pass_nothing(name, seams):
c = BY[name]
seed(c["root"], c["frm"][0], owner="wild@shop.example")
assert run(c, person("wild@shop.example", MALFORMED)) == (403, gate(c))
assert state_of(c["root"]) == c["frm"][0] and seams == []
@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 c["by"] if r in ROLES and READ_ALL.get(c["root"]) not in ROLES[r] 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"))])
@todo("TODO(rule): separation of duties — who did the request may not approve it")
def test_the_requester_may_not_approve(request_step, approval):
first, second = BY[request_step], BY[approval]
seed(first["root"], first["frm"][0])
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, todo(f"TODO(rule): {c['name']} under a {s} {parent}")) for s in TERMINAL[parent]]
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"))])
@todo("TODO(rule): the seam call this transition makes (module.yaml seams)")
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 [b for b, _ in seams] == [brick]
# ── 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