dodil/erp-platform
tests/test_signoff.py
"""erp-compose signoff: the customer signs the process, not the drawing; a build is checked against it by stable id.
The fixture is brick-hr 0.4.0's show_playbook_bpmn for the gb pack's employee-onboarding playbook (v1)."""
import json
import pathlib
import re
import pytest
from erp_platform import compose, signoff
SIGNED = (pathlib.Path(__file__).parent / "data" / "employee-onboarding.v1.bpmn").read_text()
DEVS = {"playbook_id": "employee-onboarding", "version": 1, "deviations": [
{"key": "send_contract", "change": "changed", "field": "responsible_group", "before": "hr", "after": "director",
"class": "variation", "reason": "a director signs"}]}
SCEN = {"playbook_id": "employee-onboarding", "version": 1, "paths": [
{"scenario_id": "SC-HR-29", "title": "a contract that needs an approval", "path": ["approve_rate", "send_contract"],
"walks": True, "problem": None}]}
def _moved(xml: str) -> str:
"""Every shape and waypoint moved, as a modeler would after dragging the diagram."""
return re.sub(r'(x|y)="(\d+(?:\.\d+)?)"', lambda m: f'{m.group(1)}="{float(m.group(2)) + 37.5}"', xml)
def _in_lane(xml: str, element: str, lane: str) -> str:
"""Move a task to another lane (the responsible group) — a process change."""
xml = xml.replace(f"<bpmn:flowNodeRef>{element}</bpmn:flowNodeRef>", "", 1)
return re.sub(rf'(<bpmn:lane id="{lane}"[^>]*>)', rf"\1<bpmn:flowNodeRef>{element}</bpmn:flowNodeRef>", xml, 1)
def _files(tmp_path, devs=DEVS, scen=SCEN):
(tmp_path / "p.bpmn").write_text(SIGNED)
(tmp_path / "d.json").write_text(json.dumps(devs))
(tmp_path / "s.json").write_text(json.dumps(scen))
return [str(tmp_path / f) for f in ("p.bpmn", "s.json", "d.json")]
def test_same_diagram_same_hash_even_after_moving_every_shape():
moved = _moved(SIGNED)
assert moved != SIGNED
assert signoff.content_hash(moved) == signoff.content_hash(SIGNED)
def test_reordering_prefixes_whitespace_and_the_exporter_stamp_change_nothing():
other = (SIGNED.replace("bpmn:", "bpmn2:").replace("xmlns:bpmn=", "xmlns:bpmn2=")
.replace('exporterVersion="0.4.0"', 'exporterVersion="9.9.9"').replace("\n ", "\n "))
assert signoff.content_hash(other) == signoff.content_hash(SIGNED)
def test_a_step_change_is_a_mismatch_and_the_diff_names_the_step(tmp_path):
p, s, d = _files(tmp_path)
signoff.record("hr", p, s, d, "Jane Doe, MD", directory=tmp_path)
drift = _in_lane(_moved(SIGNED), "Task_send_contract", "Lane_it")
res = signoff.verify("hr", drift, directory=tmp_path)
assert not res["ok"] and res["signed"] != res["current"]
row = next(r for r in res["diff"] if r["id"] == "Task_send_contract")
assert row["op"] == "changed" and row["step"] == "send_contract" and row["kind"] == "task"
assert row["changes"] == {"lane": ["Lane_hr", "Lane_it"]}
assert len(res["diff"]) == 1, res["diff"] # moving the shapes is not in it
assert "~ task Task_send_contract [step send_contract]" in signoff.report("hr", res)
def test_added_removed_flows_gateways_and_timers_are_named_by_kind():
cur = re.sub(r'(<bpmn:timeDate[^>]*>)[^<]*(</bpmn:timeDate>)', r"\g<1>5 working days before start_date\g<2>",
SIGNED, count=1)
cur = re.sub(r'<bpmn:sequenceFlow id="Flow_Gateway_start__Timer_right_to_work_check"[^>]*/>', "", cur)
rows = {r["id"]: r for r in signoff.diff(SIGNED, cur)}
assert rows["Flow_Gateway_start__Timer_right_to_work_check"]["op"] == "removed"
assert rows["Flow_Gateway_start__Timer_right_to_work_check"]["kind"] == "flow"
t = next(r for r in rows.values() if r["kind"] == "timer")
assert t["op"] == "changed" and "timer" in t["changes"]
gw = SIGNED.replace("</bpmn:process>", '<bpmn:exclusiveGateway id="Gateway_new__if" /></bpmn:process>')
assert [(r["op"], r["kind"]) for r in signoff.diff(SIGNED, gw)] == [("added", "gateway")]
def test_record_writes_the_record_the_signed_copy_and_the_design_file(tmp_path):
p, s, d = _files(tmp_path)
(tmp_path / "ERP_HR.md").write_text("# ERP_HR\n\n## Processes\nThe signed BPMN.\n\n## Scenarios\n\n"
"## Deviations\n\n## Decisions\n- kept\n")
rec = signoff.record("hr", p, s, d, "Jane Doe, MD", directory=tmp_path, signed_at="2026-10-18T12:00:00Z")
on_disk = json.loads((tmp_path / "signoff" / "hr" / "employee-onboarding.json").read_text())
assert on_disk == rec
assert {k: rec[k] for k in ("module", "playbook_id", "version", "scenario_ids", "approver", "signed_at")} == {
"module": "hr", "playbook_id": "employee-onboarding", "version": 1, "scenario_ids": ["SC-HR-29"],
"approver": "Jane Doe, MD", "signed_at": "2026-10-18T12:00:00Z"}
assert rec["bpmn_sha256"] == signoff.content_hash(SIGNED)
assert rec["deviations"] == [{"key": "send_contract", "change": "changed", "field": "responsible_group",
"class": "variation", "reason": "a director signs", "accepted": True}]
assert (tmp_path / "signoff" / "hr" / "employee-onboarding.bpmn").read_text() == SIGNED
md = (tmp_path / "ERP_HR.md").read_text()
assert rec["bpmn_sha256"] in md and "SC-HR-29" in md and "a director signs" in md and "Jane Doe, MD, 2026-10-18" in md
assert md.index("SC-HR-29") < md.index("## Deviations") < md.index("a director signs") < md.index("## Decisions")
signoff.record("hr", p, s, d, "Jane Doe, MD", directory=tmp_path, signed_at="2026-10-19T12:00:00Z")
again = (tmp_path / "ERP_HR.md").read_text()
assert again.count(rec["bpmn_sha256"]) == md.count(rec["bpmn_sha256"]) # updated in place, not appended
assert "2026-10-19" in again and "- kept" in again
def test_a_foundation_deviation_is_refused_and_nothing_is_written(tmp_path):
bad = {**DEVS, "deviations": DEVS["deviations"] + [
{"key": "right_to_work_check", "change": "removed", "field": None, "class": "foundation",
"reason": "the UK right-to-work record is a legal gate"}]}
p, s, d = _files(tmp_path, devs=bad)
with pytest.raises(signoff.SignoffError, match="foundation.*right_to_work_check"):
signoff.record("hr", p, s, d, "Jane Doe", directory=tmp_path)
assert not (tmp_path / "signoff").exists()
assert compose.main(["signoff", "record", "--module", "hr", "--bpmn", p, "--scenarios", s, "--deviations", d,
"--approver", "Jane Doe", "--dir", str(tmp_path)]) == 1
def test_a_scenario_that_does_not_walk_cannot_be_signed(tmp_path):
p, s, d = _files(tmp_path, scen={"paths": [{"scenario_id": "SC-HR-29", "walks": False, "problem": "no step x"}]})
with pytest.raises(signoff.SignoffError, match="SC-HR-29 does not walk"):
signoff.record("hr", p, s, d, "Jane Doe", directory=tmp_path)
def test_the_command_exits_zero_on_conformance_and_non_zero_with_a_diff(tmp_path, capsys):
p, s, d = _files(tmp_path)
assert compose.main(["signoff", "record", "--module", "hr", "--bpmn", p, "--scenarios", s, "--deviations", d,
"--approver", "Jane Doe", "--dir", str(tmp_path)]) == 0
(tmp_path / "moved.bpmn").write_text(_moved(SIGNED))
assert compose.main(["signoff", "verify", "--module", "hr", "--bpmn", str(tmp_path / "moved.bpmn"),
"--dir", str(tmp_path)]) == 0
assert "conforms" in capsys.readouterr().out
(tmp_path / "drift.bpmn").write_text(_in_lane(SIGNED, "Task_send_contract", "Lane_it"))
assert compose.main(["signoff", "verify", "--module", "hr", "--bpmn", str(tmp_path / "drift.bpmn"),
"--dir", str(tmp_path)]) == 1
out = capsys.readouterr().out
assert "MISMATCH" in out and "Task_send_contract" in out and "lane: Lane_hr → Lane_it" in out
# ── 0.8.1: one module, many processes ──────────────────────────────────────────────────────
FREELANCER = SIGNED.replace("employee-onboarding", "freelancer-onboarding")
def _sign(tmp_path, xml, name, **kw):
(tmp_path / f"{name}.bpmn").write_text(xml)
return signoff.record("hr", str(tmp_path / f"{name}.bpmn"), None, None, "Jane Doe, MD", directory=tmp_path, **kw)
def test_two_processes_of_one_module_are_signed_side_by_side(tmp_path, capsys):
emp = _sign(tmp_path, SIGNED, "emp")
fre = _sign(tmp_path, FREELANCER, "fre")
assert (emp["process_id"], fre["process_id"]) == ("employee-onboarding", "freelancer-onboarding")
base = tmp_path / "signoff" / "hr"
assert json.loads((base / "employee-onboarding.json").read_text()) == emp # the first is untouched
assert set(signoff.records(tmp_path, "hr")) == {"employee-onboarding", "freelancer-onboarding"}
# superseding ONE process makes its v2 and leaves the other at v1
changed = _in_lane(FREELANCER, "Task_send_contract", "Lane_it")
with pytest.raises(signoff.SignoffError, match="hr/freelancer-onboarding is signed as v1: a new version needs"):
_sign(tmp_path, changed, "fre2")
v2 = _sign(tmp_path, changed, "fre2", reason="freelancers: IT sends the contract")
assert v2["signed_version"] == 2 and v2["supersedes"]["signed_version"] == 1
assert json.loads((base / "employee-onboarding.json").read_text()) == emp
assert {f.name for f in (base / "history").iterdir()} >= {
"employee-onboarding.v1.json", "freelancer-onboarding.v1.json", "freelancer-onboarding.v2.json"}
md = (tmp_path / "ERP_HR.md").read_text()
assert "| employee-onboarding | v1 |" in md and "| freelancer-onboarding | v2 |" in md
# verify picks the process the BPMN draws; each conforms to its own signature
assert signoff.verify("hr", SIGNED, directory=tmp_path)["ok"]
assert signoff.verify("hr", changed, directory=tmp_path)["ok"]
res = signoff.verify("hr", FREELANCER, directory=tmp_path)
assert not res["ok"] and res["process"] == "freelancer-onboarding"
assert "signoff verify hr/freelancer-onboarding: MISMATCH" in signoff.report("hr", res)
# an explicit --process wins
assert signoff.main(["record", "--module", "hr", "--process", "contractor", "--bpmn", str(tmp_path / "emp.bpmn"),
"--approver", "Jane", "--dir", str(tmp_path)]) == 0
assert "also signed, untouched: employee-onboarding, freelancer-onboarding" in capsys.readouterr().out
def test_the_gate_and_verify_all_cover_every_signed_process(tmp_path, capsys):
from erp_platform import gate
_sign(tmp_path, SIGNED, "emp")
_sign(tmp_path, FREELANCER, "fre")
for k in ("employee-onboarding", "freelancer-onboarding"): # a playbook regenerates through the bricks
p = tmp_path / "signoff" / "hr" / f"{k}.json"
rec = json.loads(p.read_text())
rec["source"] = {"kind": "bpmn", "from": f"{k}.src.bpmn"}
p.write_text(json.dumps(rec))
import erp_platform.gate as g
orig = g.regenerate
g.regenerate = lambda root, rec, cp=None: (tmp_path / "signoff" / "hr" / f"{rec['process_id']}.bpmn").read_text()
try:
rows = gate.check(tmp_path)
assert [(gate.label(r), r["state"]) for r in rows] == [
("hr/employee-onboarding", "signed"), ("hr/freelancer-onboarding", "signed")]
assert signoff.main(["verify", "--all", "--dir", str(tmp_path)]) == 0
(tmp_path / "signoff" / "hr" / "freelancer-onboarding.bpmn").write_text(
_in_lane(FREELANCER, "Task_send_contract", "Lane_it"))
assert signoff.main(["verify", "--all", "--dir", str(tmp_path)]) == 1
out = capsys.readouterr().out
assert "signoff hr/employee-onboarding: SIGNED" in out and "signoff hr/freelancer-onboarding: DRIFTED" in out
assert gate.status(tmp_path, "hr")["state"] == "drifted"
finally:
g.regenerate = orig
def test_the_0_8_0_single_file_layout_is_migrated_on_read(tmp_path):
p, s, d = _files(tmp_path)
old = signoff.record("hr", p, s, d, "Jane Doe, MD", directory=tmp_path)
base, legacy = tmp_path / "signoff" / "hr", tmp_path / "signoff"
# rebuild what 0.8.0 wrote: signoff/hr.json + hr.bpmn, signoff/history/hr.v1.*, a pending proposal
(legacy / "history").mkdir()
rec = {k: v for k, v in old.items() if k not in ("process_id", "scenarios")}
rec["bpmn_file"] = "signoff/hr.bpmn"
(legacy / "hr.json").write_text(json.dumps(rec))
(legacy / "hr.bpmn").write_text(SIGNED)
(legacy / "history" / "hr.v1.json").write_text(json.dumps(rec))
(legacy / "history" / "hr.v1.bpmn").write_text(SIGNED)
(legacy / "hr.pending.json").write_text(json.dumps({"proposed_sha256": "x", "against_sha256": old["bpmn_sha256"]}))
import shutil
shutil.rmtree(base)
assert signoff.verify("hr", _moved(SIGNED), directory=tmp_path)["ok"] # read → migrated
assert not (legacy / "hr.json").exists() and not (legacy / "history" / "hr.v1.json").exists()
moved = json.loads((base / "employee-onboarding.json").read_text())
assert moved["process_id"] == "employee-onboarding"
assert moved["bpmn_file"] == "signoff/hr/employee-onboarding.bpmn"
assert (base / "history" / "employee-onboarding.v1.bpmn").read_text() == SIGNED
assert json.loads((base / "employee-onboarding.pending.json").read_text())["process_id"] == "employee-onboarding"
fre = _sign(tmp_path, FREELANCER, "fre") # and a second process lands beside it
assert fre["signed_version"] == 1 and set(signoff.records(tmp_path, "hr")) == {"employee-onboarding",
"freelancer-onboarding"}
def test_a_change_is_proposed_against_the_one_process_it_changes(tmp_path):
from erp_platform import change, gate
_sign(tmp_path, SIGNED, "emp")
_sign(tmp_path, FREELANCER, "fre")
p = change.propose("hr", _in_lane(FREELANCER, "Task_send_contract", "Lane_it").encode(), directory=tmp_path)
assert p["process_id"] == "freelancer-onboarding"
assert (tmp_path / "signoff" / "hr" / "freelancer-onboarding.pending.json").exists()
assert not (tmp_path / "signoff" / "hr" / "employee-onboarding.pending.json").exists()
assert "--process freelancer-onboarding" in change.report(p)
rows = {r["process"]: r["state"] for r in gate.check(tmp_path, can_boot=False)}
assert rows["freelancer-onboarding"] == "pending"
# ── 0.9.1: canonical v2 — the generator's provenance line is not the process ─────────────────────
def _drawn_from(name: str) -> str:
spec = pathlib.Path(__file__).parent / "fixtures" / "scaffold" / "quote.module.yaml"
from erp_platform import bpmn
return bpmn.generate(spec).replace("generated from quote.module.yaml", f"generated from {name}")
def test_v2_ignores_which_file_the_diagram_was_drawn_from_and_nothing_else(tmp_path):
a, b = _drawn_from("module.yaml"), _drawn_from("process.yaml")
assert a != b and signoff.content_hash(a) == signoff.content_hash(b)
assert signoff.content_hash(a, "legacy") != signoff.content_hash(b, "legacy")
(tmp_path / "m.bpmn").write_text(a)
rec = signoff.record("example", tmp_path / "m.bpmn", None, None, "Jane Doe, MD", directory=tmp_path)
assert rec["canonical"] == "v2"
res = signoff.verify("example", b, directory=tmp_path) # the brick's served view (process.yaml)
assert res["ok"] and res["canonical"] == "v2" and "canonical v2" in signoff.report("example", res)
other = b.replace("States:", "Stages:", 1) # any other documentation word still counts
assert not signoff.verify("example", other, directory=tmp_path)["ok"]
def test_a_signature_recorded_under_the_legacy_form_keeps_verifying(tmp_path):
a = _drawn_from("module.yaml")
(tmp_path / "m.bpmn").write_text(a)
signoff.record("example", tmp_path / "m.bpmn", None, None, "Jane Doe, MD", directory=tmp_path)
for f in (tmp_path / "signoff" / "example").rglob("*.json"): # as 0.9.0 wrote it
rec = json.loads(f.read_text())
rec["bpmn_sha256"] = signoff.content_hash(a, "legacy")
rec.pop("canonical", None)
rec["source"] = {"kind": "bpmn", "from": "module.yaml", "workflow": None}
f.write_text(json.dumps(rec))
res = signoff.verify("example", a, directory=tmp_path)
assert res["ok"] and res["canonical"] == "legacy"
# the brick's served view says process.yaml: the record's own source puts the signed provenance back
served = signoff.verify("example", _drawn_from("process.yaml"), directory=tmp_path)
assert served["ok"] and served["canonical"] == "legacy"
assert not signoff.verify("example", _drawn_from("process.yaml").replace("States:", "Stages:", 1),
directory=tmp_path)["ok"]
# re-signing the same process is the same version, not a change that needs a reason
again = signoff.record("example", tmp_path / "m.bpmn", None, None, "Jane Doe, MD", directory=tmp_path)
assert again["signed_version"] == 1 and again["canonical"] == "v2"
# ── 0.9.11: a composition of many modules — processes/<module>/module.yaml beside signoff/<module>/ ─────────
def test_receipts_signed_in_a_module_folder_verify_unedited_in_a_composition(tmp_path, capsys):
import shutil
from erp_platform import gate
fx = pathlib.Path(__file__).parent
specs = {"projects": fx / "fixtures" / "scaffold" / "projects.module.yaml", "quote": fx / "data" / "quote.module.yaml"}
method, comp = tmp_path / "method", tmp_path / "comp"
for m, spec in specs.items(): # the method's layout: modules/<m>/module.yaml + signoff/<m>/
(method / m).mkdir(parents=True)
shutil.copy(spec, method / m / "module.yaml")
for wf in [w.entity for w in __import__("erp_platform.bpmn", fromlist=["load"]).load(spec).workflows]:
signoff.record(m, None, None, None, "Method Author", directory=method / m, source="module.yaml", workflow=wf)
comp.mkdir()
(comp / "compose.yaml").write_text("name: c\nmodules: [projects, quote]\nbricks: []\n")
for m in specs: # copied byte for byte: processes/<m>/ + signoff/<m>/
(comp / "processes" / m).mkdir(parents=True)
shutil.copy(method / m / "module.yaml", comp / "processes" / m / "module.yaml")
shutil.copytree(method / m / "signoff" / m, comp / "signoff" / m)
before = {p: p.read_bytes() for p in (comp / "signoff").rglob("*.json")}
rows = gate.check(comp)
assert len(rows) == 3 and {r["state"] for r in rows} == {"signed"}, rows
assert signoff.main(["verify", "--all", "--dir", str(comp)]) == 0
assert {p: p.read_bytes() for p in (comp / "signoff").rglob("*.json")} == before # nothing edited
# the customer's one approval: every process of every module, one receipt each, content unchanged
capsys.readouterr()
assert signoff.main(["record", "--all", "--approver", "Jo Buyer, MD", "--reason", "approved the composition",
"--dir", str(comp)]) == 0
recs = {k: r for m in specs for k, r in signoff.records(comp, m).items()}
assert len(recs) == 3 and {r["approver"] for r in recs.values()} == {"Jo Buyer, MD"}
assert {r["source"]["from"] for r in recs.values()} == {"module.yaml"}
assert {r["signed_version"] for r in recs.values()} == {1} # the same process: no new version
assert "signed 3 process(es) of 2 module(s)" in capsys.readouterr().out
assert {r["state"] for r in gate.check(comp)} == {"signed"}
for m in specs: # the method's receipts as a reference baseline (init)
shutil.copytree(method / m / "signoff" / m, comp / "processes" / m / "baseline")
assert all("identical to the standard v1" in r["detail"] for r in gate.check(comp))
# a spec edited without a new sign-off is drift, per module folder
q = comp / "processes" / "quote" / "module.yaml"
q.write_text(q.read_text().replace("expired", "lapsed"))
assert "drifted" in {r["state"] for r in gate.check(comp) if r["module"] == "quote"}
def test_a_receipt_naming_its_source_from_the_root_still_verifies(tmp_path):
from erp_platform import gate
fx = pathlib.Path(__file__).parent / "data" / "quote.module.yaml"
(tmp_path / "processes" / "quote").mkdir(parents=True)
(tmp_path / "processes" / "quote" / "module.yaml").write_text(fx.read_text())
rec = signoff.record("quote", None, None, None, "A", directory=tmp_path, source="processes/quote/module.yaml")
assert rec["source"]["from"] == "module.yaml" # recorded relative to its folder
r = json.loads((tmp_path / "signoff" / "quote" / f"{rec['process_id']}.json").read_text())
r["source"]["from"] = "processes/quote/module.yaml" # what the 0.9.10 deploy hand-edited
(tmp_path / "signoff" / "quote" / f"{rec['process_id']}.json").write_text(json.dumps(r))
assert {x["state"] for x in gate.check(tmp_path, mods=["quote"])} == {"signed"}
def test_record_all_names_a_module_without_a_spec_folder(tmp_path, capsys):
(tmp_path / "compose.yaml").write_text("name: c\nmodules: [quote]\nbricks: []\n")
assert signoff.main(["record", "--all", "--approver", "A", "--dir", str(tmp_path)]) == 1
assert "processes/<module>/module.yaml" in capsys.readouterr().err