"""The customer's process against the standard (0.4.0): stable step keys and the BPMN ids derived from them, the
deviations from the pack's reference playbook (variation / warning / foundation), the scenarios as paths, and each
change a customer is likely to ask for in a demo — one use-case call, then the BPMN and the deviations say so."""
import copy
import hashlib
import json
import os
import pathlib
import re
import xml.etree.ElementTree as ET
import pytest
import yaml
from erp_platform import bpmn as platform_bpmn
from erp_platform import work
from brick_hr.application import engine as E
from brick_hr.application import process, use_cases
from brick_hr.application import support as S
from brick_hr.application.organisation import _playbook_from
from brick_hr.domain import bpmn, playbook as pbk
from brick_hr.domain import run as run_domain
from brick_hr.domain.packs import gb
from bpmn_check import check, q
from conftest import caller, headers
from seed import hire_body, seed, sign_in
HR = headers("hr_admin")
MGR = headers("manager", sub="manager.7")
DODIL = f"{{{bpmn.DODIL_NS}}}"
ROOT = pathlib.Path(__file__).resolve().parents[1]
def _gb(playbook_id: str) -> dict:
return copy.deepcopy(next(b for b in gb.starter_playbooks() if b["playbook_id"] == playbook_id))
def _xml(body: dict, version: int = 1, deviations=()) -> str:
from brick_hr.application.organisation import bpmn_setting
from test_rules import CFG
pb = _playbook_from(body, version, "hr_admin.1", None)
return bpmn.to_bpmn(pb, bpmn_setting(CFG, pb), deviations)
def _ids(xml: str) -> dict:
"""element id → (tag, source, target): what an id names, so 'the same id' means 'the same element'."""
root = ET.fromstring(xml)
proc = root.find(q("bpmn:process"))
return {e.get("id"): (e.tag, e.get("sourceRef"), e.get("targetRef")) for e in proc.iter() if e.get("id")}
# ── stable keys ────────────────────────────────────────────────────────────────────────────
def test_every_gb_step_has_a_designed_key_unique_in_its_playbook():
for body in gb.starter_playbooks():
keys = [s["key"] for s in body["steps"]]
assert len(set(keys)) == len(keys) and all(pbk.KEY.match(k) for k in keys), body["playbook_id"]
assert [s["key"] for s in _gb("employee-onboarding")["steps"]][:5] == [
"starter_details", "approve_rate", "send_contract", "right_to_work_check", "create_account"]
def test_inserting_a_step_changes_no_other_elements_id():
body = _gb("employee-onboarding")
before = _ids(_xml(body))
pb = _playbook_from(body, 1, "x", None)
steps = pbk.edited(pb, "add_step", after=["create_account"], before=["add_to_groups"],
step={"kind": "approval", "title": "Director approves the account", "responsible_group": "hr"})
after = _ids(_xml({**body, "steps": steps}, version=2))
gone = {i for i in before if i not in after and not i.startswith("Process_")} # the process id names its version
# only the one flow the new step now sits on is replaced; every node and every other flow keeps its id
assert gone == {"Flow_Gateway_create_account__fork__Timer_add_to_groups"}, gone
assert all(after[i] == before[i] for i in before if i in after and not i.startswith(("Process_", "LaneSet_")))
assert after["Task_director_approves_the_account"][0] == q("bpmn:userTask")
def test_renumbering_the_checklist_keeps_every_key_and_so_every_id():
body = _gb("employee-onboarding")
v1 = _playbook_from(body, 1, "x", None)
shifted = copy.deepcopy(body)
for s in shifted["steps"]: # a customer inserts at the top and renumbers everything after
s.pop("key")
s["step_no"] += 1
s["depends_on"] = [d + 1 for d in s.get("depends_on") or []]
shifted["steps"].insert(0, {"step_no": 1, "kind": "task", "title": "Desk booked", "responsible_group": "it"})
v2 = _playbook_from(shifted, 2, "x", None, v1)
assert {s.title: s.key for s in v2.steps if s.title != "Desk booked"} == {s.title: s.key for s in v1.steps}
assert v2.step(1).key == "desk_booked"
def test_a_new_step_gets_a_unique_slug_and_never_a_key_the_last_version_used_for_another_step():
v1 = _playbook_from(_gb("employee-offboarding"), 1, "x", None)
steps = [dict(s.__dict__) for s in v1.steps if s.key != "exit_survey"]
steps += [{"step_no": 20, "kind": "task", "title": "Exit survey!", "responsible_group": "hr"},
{"step_no": 21, "kind": "task", "title": "Exit survey?", "responsible_group": "hr"}]
for s in steps:
if s["step_no"] >= 20:
s.pop("key", None)
v2 = _playbook_from({**_gb("employee-offboarding"), "steps": steps}, 2, "x", None, v1)
assert [v2.step(20).key, v2.step(21).key] == ["exit_survey_2", "exit_survey_3"]
with pytest.raises(pbk.RuleBroken, match="two steps have the key"):
bad = _playbook_from({**_gb("employee-offboarding"), "steps": [
{"step_no": 1, "key": "same", "kind": "task", "title": "a", "responsible_group": "hr"},
{"step_no": 2, "key": "same", "kind": "task", "title": "b", "responsible_group": "hr"}]}, 1, "x", None)
bad.validate(known=lambda c: True, destructive=lambda c: False, kind_of_event=lambda e: "Left",
groups=lambda g: True)
def test_the_bpmn_carries_the_key_and_the_apqc_pcf_id_and_stays_xsd_valid():
root = check(_xml(_gb("employee-onboarding")))
proc = root.find(q("bpmn:process"))
assert proc.get(f"{DODIL}pcf") == "10469" # Manage employee onboarding
by = {e.get("id"): e for e in proc}
assert by["Task_send_contract"].get(f"{DODIL}step") == "send_contract"
assert by["Task_send_contract"].get(f"{DODIL}pcf") == "11052" # agreements/contracts
assert by["Task_create_account"].get(f"{DODIL}pcf") == "20758" # IT user directory
assert by["Task_payroll_setup"].get(f"{DODIL}pcf") == "10497" # Administer Payroll
assert by["Task_laptop_ready"].get(f"{DODIL}pcf") is None # no clear PCF activity: none
off = check(_xml(_gb("employee-offboarding"))).find(q("bpmn:process"))
assert off.get(f"{DODIL}pcf") == "21710"
# ── the doors: a world on SQLite ───────────────────────────────────────────────────────────
@pytest.fixture
def world(engine, api, drafter):
work.set_clock(at=work.parse_instant("2026-10-19T10:00 Europe/London"), freeze=True)
seed(lambda p, b: api("POST", p, HR, b))
work.set_clock(at=work.parse_instant("2024-01-10T10:00 Europe/London"), freeze=True)
s, lead = api("POST", "/hires", HR, hire_body("Lena", "Lead", "2024-01-08", job="studio_lead",
position_id="P-STUDIO"))
assert s == 200, lead
work.set_clock(at=work.parse_instant("2026-10-19T10:00 Europe/London"), freeze=True)
sign_in(lambda p, b: api("POST", p, HR, b), lead["person_id"], "manager.7", get=lambda p: api("GET", p, HR))
return {"lead": lead}
def _devs(api, pid="employee-onboarding"):
s, out = api("GET", f"/playbooks/{pid}/deviations", HR)
assert s == 200, out
return out
def _bpmn(api, pid="employee-onboarding"):
s, out = api("GET", f"/playbooks/{pid}/bpmn", HR, headers={"Accept": "application/json"})
assert s == 200, out
return check(out["bpmn"]), out["bpmn"]
def _change(api, body, pid="employee-onboarding", status=200):
s, out = api("POST", f"/playbooks/{pid}/changes", HR, body)
assert s == status, out
return out
def _task(root, key):
return next(e for e in root.find(q("bpmn:process")) if e.get("id") == f"Task_{key}")
def test_the_reference_itself_deviates_in_nothing_through_both_doors(world, api, tool):
out = _devs(api)
assert out["reference"] == {"pack": "gb", "playbook_id": "employee-onboarding"}
assert out["deviations"] == [] and out["summary"] == {"variation": 0, "warning": 0, "foundation": 0}
assert tool(HR, "playbook_deviations", {"playbook_id": "employee-onboarding"}) == ("ok", out)
s, refused = api("GET", "/playbooks/employee-onboarding/deviations", MGR)
assert (s, refused["detail"]) == (403, "missing permission: hr:config:write")
def test_change_add_a_step_a_director_approves_before_the_contract(world, api, tool):
kind, out = tool(HR, "change_playbook", {"playbook_id": "employee-onboarding", "change": "add_step",
"step": {"kind": "approval", "title": "Director approves the contract",
"responsible_group": "director"},
"before": ["send_contract"]})
assert kind == "ok" and out["version"] == 2 and out["key"] == "director_approves_the_contract"
added = [d for d in out["deviations"] if d["change"] == "added"]
assert [(d["key"], d["class"]) for d in added] == [("director_approves_the_contract", "variation")]
wired = next(d for d in out["deviations"] if d["key"] == "send_contract" and d["field"] == "depends_on")
assert wired["after"] == ["approve_rate", "director_approves_the_contract"] and wired["class"] == "variation"
root, xml = _bpmn(api)
t = _task(root, "director_approves_the_contract")
assert t.get(f"{DODIL}deviation") == "variation" and "Deviation from the reference" in t.findtext(
q("bpmn:documentation"))
assert 'sourceRef="Task_director_approves_the_contract"' in xml
assert _task(root, "create_account").get(f"{DODIL}deviation") is None # untouched steps are not marked
def test_change_remove_a_step(world, api):
out = _change(api, {"change": "remove_step", "key": "buddy_coffee"})
assert [(d["key"], d["change"], d["class"]) for d in out["deviations"]] == [("buddy_coffee", "removed", "variation")]
root, xml = _bpmn(api)
assert 'id="Task_buddy_coffee"' not in xml and "Not in this playbook, from the reference: Buddy coffee" in xml
out = _change(api, {"change": "remove_step", "key": "laptop_ready"})
lap = next(d for d in out["deviations"] if d["key"] == "laptop_ready")
assert lap["class"] == "variation"
_change(api, {"change": "remove_step", "key": "no_such"}, status=404)
def test_change_the_owner_group_of_a_step(world, api):
out = _change(api, {"change": "set_owner", "key": "laptop_ready", "responsible_group": "finance"})
d = next(d for d in out["deviations"] if d["key"] == "laptop_ready")
assert (d["field"], d["before"], d["after"], d["class"]) == ("responsible_group", "it", "finance", "variation")
root, _ = _bpmn(api)
lane = next(ln for ln in root.iter(q("bpmn:lane")) if any(r.text == "Task_laptop_ready"
for r in ln.findall(q("bpmn:flowNodeRef"))))
assert lane.get("id") == "Lane_finance"
_change(api, {"change": "set_owner", "key": "laptop_ready", "responsible_group": "nobody"}, status=409)
def test_change_a_timing_rule(world, api):
out = _change(api, {"change": "set_timing", "key": "create_account", "timing": {"offset": -5}})
d = next(d for d in out["deviations"] if d["key"] == "create_account")
assert d["field"] == "timing" and d["after"]["offset"] == -5 and d["class"] == "variation"
root, _ = _bpmn(api)
timer = next(e for e in root.find(q("bpmn:process")) if e.get("id") == "Timer_create_account")
assert timer.findtext(f"{q('bpmn:timerEventDefinition')}/{q('bpmn:timeDate')}").startswith(
"5 working days before start_date, 09:00")
# taking access away later than the last day is allowed, and flagged
off = _change(api, {"change": "set_timing", "key": "suspend_account", "timing": {"offset": 2}},
pid="employee-offboarding")
w = next(d for d in off["deviations"] if d["key"] == "suspend_account")
assert w["class"] == "warning" and "access lingers" in w["reason"]
root, _ = _bpmn(api, "employee-offboarding")
assert _task(root, "suspend_account").get(f"{DODIL}deviation") == "warning"
# a step:<key>.done anchor is stored as the step's number and shown by its title
_change(api, {"change": "set_timing", "key": "first_project_brief",
"timing": {"anchor": "step:laptop_ready.done", "offset": 0, "unit": "days", "time": None}})
_, xml = _bpmn(api)
assert "when 'Laptop ready' is done" in xml or "when 'Laptop ready' is done" in xml
def test_change_remove_an_approval_warns_or_is_refused_when_it_guards_something_destructive(world, api):
out = _change(api, {"change": "remove_approval", "key": "approve_rate"})
w = next(d for d in out["deviations"] if d["key"] == "approve_rate")
assert (w["change"], w["class"]) == ("removed", "warning") and "Contract sent and signed" in w["reason"]
assert out["summary"]["warning"] >= 1 and out["summary"]["foundation"] == 0
# the approval before access removal guards destructive steps: the kept foundation, refused at save
s, refused = api("POST", "/playbooks/employee-offboarding/changes", HR,
{"change": "remove_approval", "key": "approve_access_removal"})
assert s == 409 and "must depend on an approval step" in refused["detail"]
_change(api, {"change": "remove_approval", "key": "laptop_returned"}, pid="employee-offboarding", status=409)
def test_dropping_a_pack_gate_is_the_kept_foundation_and_refused_at_save(world, api):
s, out = api("GET", "/playbooks", HR, params={"playbook_id": "employee-onboarding"})
pb = out["playbooks"][0]
steps = copy.deepcopy(pb["steps"])
acc = next(x for x in steps if x["key"] == "create_account")
acc["condition"] = None # no right-to-work gate before the account
s, refused = api("POST", "/playbooks", HR, {**{k: pb[k] for k in ("playbook_id", "name", "event_types",
"worker_types")}, "steps": steps})
assert s == 409, refused
assert "drops the right_to_work_check gate" in refused["detail"] and "Immigration, Asylum and Nationality Act" \
in refused["detail"] and "kept foundation" in refused["detail"]
assert _devs(api)["version"] == 1 # nothing stored
def test_change_who_signs_the_contract(world, api):
lead = world["lead"]["person_id"]
s, _ = api("POST", "/configuration/settings", HR, {"rows": [{"key": "contract_signer_person_id", "value": lead}]})
assert s == 200
d = next(d for d in _devs(api)["deviations"] if d["field"] == "contract_signer_person_id")
assert (d["change"], d["after"], d["class"]) == ("setting", lead, "variation")
root, _ = _bpmn(api)
assert "Signed for the company by Lena Lead (the contract signer setting)" in _task(root, "send_contract").findtext(
q("bpmn:documentation"))
s, out = api("POST", "/configuration/settings", HR, {"rows": [{"key": "contract_signer_person_id", "value": "x"}]})
assert s == 422
# and the contract hr drafts names that signer for the company
s, anna = api("POST", "/hires", MGR, hire_body("Anna", "Berg", "2026-11-02"))
assert s == 200, anna
with use_cases.repository() as uow:
view = S.View(uow, caller("clock"))
run = uow.runs.get(anna["run_id"])
inp = E.contract_input(view, run, run.steps["3"], uow.persons.get(int(anna["person_id"])))
assert inp["signers"][0]["full_name"] == "Lena Lead" and inp["signers"][0]["side"] == "US"
def test_change_the_chase_cadence(world, api):
s, _ = api("POST", "/configuration/settings", HR, {"rows": [{"key": "approval_chase_every_days", "value": "2"},
{"key": "task_chase_max", "value": "5"}]})
assert s == 200
rows = {d["field"]: d for d in _devs(api)["deviations"] if d["change"] == "setting"}
assert rows["approval_chase_every_days"]["after"] == "2" and rows["task_chase_max"]["class"] == "variation"
root, _ = _bpmn(api)
assert "Chased every 2 working days, up to 3 times" in _task(root, "approve_rate").findtext(q("bpmn:documentation"))
assert "up to 5 times, then a task for HR: task overdue" in _task(root, "laptop_ready").findtext(
q("bpmn:documentation"))
api("POST", "/configuration/settings", HR, {"rows": [{"key": "approval_chase_max", "value": "0"}]})
never = next(d for d in _devs(api)["deviations"] if d["field"] == "approval_chase_max")
assert never["class"] == "warning"
# ── scenarios as paths ─────────────────────────────────────────────────────────────────────
def test_every_declared_scenario_path_walks_the_reference_and_names_a_real_scenario():
catalogue = (ROOT / "tests" / "scenarios" / "catalogue.py").read_text()
known = set(re.findall(r'@CAT\.scenario\("(SC-HR-\d+)"', catalogue))
refs = {b["playbook_id"]: _playbook_from(b, 1, "x", None) for b in gb.starter_playbooks()}
assert process.declared_paths(), "no paths declared"
for p in process.declared_paths():
assert p["scenario"] in known, p
assert pbk.walk_problem(refs[p["playbook"]], p["path"]) is None, p
# and the check can fail: a step out of order is not a walk
assert "does not wait for" in pbk.walk_problem(refs["employee-onboarding"], ["create_account", "send_contract"])
def test_scenario_paths_through_both_doors_and_a_change_that_breaks_one(world, api, tool):
s, out = api("GET", "/playbooks/employee-offboarding/scenario-paths", HR)
assert s == 200 and tool(HR, "scenario_paths", {"playbook_id": "employee-offboarding"}) == ("ok", out)
by = {p["scenario_id"]: p for p in out["paths"]}
assert set(by) == {"SC-HR-12", "SC-HR-13"} and all(p["walks"] for p in out["paths"])
assert by["SC-HR-12"]["elements"] == ["Task_approve_access_removal", "Task_transfer_drive", "Task_delete_account"]
_change(api, {"change": "remove_step", "key": "send_contract"})
s, on = api("GET", "/playbooks/employee-onboarding/scenario-paths", HR)
broken = [p for p in on["paths"] if not p["walks"]]
assert broken and all("send_contract" in p["problem"] for p in broken)
# ── stored before keys existed ─────────────────────────────────────────────────────────────
def test_a_step_stored_without_a_key_is_keyed_on_read_and_the_backfill_writes_it(world, api, engine):
from sqlalchemy import select, update
from brick_hr.infrastructure import tables as T
with engine.begin() as conn:
conn.execute(update(T.playbook_step).values(step_key=None))
_, xml = _bpmn(api)
assert 'id="Task_send_contract"' in xml # the designed key, on read
with use_cases.repository() as uow:
assert uow.playbooks.backfill_keys() > 0
with use_cases.repository() as uow:
assert uow.playbooks.backfill_keys() == 0 # idempotent
with engine.begin() as conn:
keys = {r[0] for r in conn.execute(select(T.playbook_step.c.step_key).where(
T.playbook_step.c.playbook_id == "employee-onboarding"))}
assert "send_contract" in keys and None not in keys
def test_the_process_doors_name_what_exists_and_need_the_configuration_permission(world, api, tool):
for path in ("/playbooks/no-such/deviations", "/playbooks/no-such/scenario-paths"):
s, out = api("GET", path, HR)
assert (s, out["detail"]) == (404, "no playbook no-such")
s, out = api("GET", path.replace("no-such", "employee-onboarding"), MGR)
assert (s, out["detail"]) == (403, "missing permission: hr:config:write")
s, out = api("POST", "/playbooks/no-such/changes", HR, {"change": "remove_step", "key": "x"})
assert (s, out["detail"]) == (404, "no playbook no-such")
assert tool(MGR, "change_playbook", {"playbook_id": "employee-onboarding", "change": "remove_step",
"key": "buddy_coffee"}) == ("refused", "403: missing permission: hr:config:write")
assert tool(HR, "scenario_paths", {"playbook_id": "no-such"}) == ("refused", "404: no playbook no-such")
def test_hr_03_my_tasks_is_the_tool_my_hr_tasks_so_hr_composes_with_activity(world, api, tool):
"""Issue #5: brick-activity has its own my_tasks; one composition cannot hold two tools of one name."""
names = {u.name for u in use_cases.HR.use_cases}
assert "my_hr_tasks" in names and "my_tasks" not in names
kind, out = tool(MGR, "my_hr_tasks", {})
s, via_api = api("GET", "/my-tasks", MGR) # the route is unchanged
assert kind == "ok" and s == 200 and out == via_api
def test_hr_42_the_offboarding_bpmn_shows_the_outplacement_offer_behind_its_reason_gateway():
xml = _xml(_gb("employee-offboarding"))
by = {e.get("id"): e for e in check(xml).find(q("bpmn:process"))}
assert by["Task_offer_outplacement"].get(f"{DODIL}pcf") == "10516" # 7.6.2.4 outplacement
assert "Gateway_offer_outplacement__if" in by
assert "the reason is redundancy" in xml and "accepted or declined" in xml
assert "Task_offer_outplacement" not in _xml(_gb("freelancer-offboarding")) # a freelancer is never redundant
# ── the signed process (hr v4, signed by seemo 2026-10-26, erp-method PR #21, modules/hr/signoff/hr/) ─────────────────
# The five processes drawn from module.yaml's workflows are re-pinned here; the two hand-drawn playbooks
# (employee-offboarding, employee-retirement) are the gb pack's rows and are checked by show_playbook_bpmn, unchanged.
# A missing erp-method checkout is an error, never a skip.
METHOD_DIR = pathlib.Path(os.environ.get("ERP_METHOD_DIR") or ROOT.parent / "erp-method")
SPEC = METHOD_DIR / "modules" / "hr" / "module.yaml"
SIGNED = METHOD_DIR / "modules" / "hr" / "signoff" / "hr"
SIGNED_VERSION = 4
PINNED = { # signoff/hr/hr_<workflow>.json file_sha256 — a changed signature is noticed here
"run": "f35a8a0ade7a074642784411b89d294729e14d8606c3dc3ab1edb51d4553c97e",
"step": "d1b1edeccce5d2226d326add99f817b36267d7cf0bd6f022ca3b03e649aef4bc",
"offboarding_playbook": "ac65958acad5e8a51ca44d3078478ca161d20201d7b6b5013724a8801bc338cb",
"engagement": "d1f165e6613545f656aedf7c9bb21afdf53a5b840e98b3e4387541ecf7bb4d40",
"intake": "fe511aa55847be8c5d10d94a696de40d1c6063ca2601556dd0a4eba009218b08",
}
def _spec() -> dict:
assert SPEC.exists(), f"the signed spec is needed: set ERP_METHOD_DIR (tried {SPEC})"
return yaml.safe_load(SPEC.read_text())
def test_the_signed_hr_v4_processes_are_pinned_and_drawn_byte_for_byte():
spec = _spec()
assert (spec["id"], spec["version"]) == ("hr", SIGNED_VERSION)
for entity, pinned in PINNED.items():
meta = json.loads((SIGNED / f"hr_{entity}.json").read_text())
assert (meta["version"], meta["file_sha256"]) == (SIGNED_VERSION, pinned), entity
one = platform_bpmn.from_module(spec, "module.yaml")
one.workflows = [w for w in one.workflows if w.entity == entity]
xml = platform_bpmn.to_bpmn(one)
assert hashlib.sha256(xml.encode()).hexdigest() == pinned, entity
assert xml == (SIGNED / f"hr_{entity}.bpmn").read_text(), entity # every frozen Task_/Flow_ id
def test_the_brick_s_step_states_are_the_signed_ones():
step = next(w for w in _spec()["workflows"] if w["entity"] == "step")
assert tuple(step["states"]) == run_domain.STATES