dodil/erp-brick-hr
tests/test_doors.py
"""Both doors over the REAL app (HR.app(): the routes and /mcp), storage on SQLite: the same outcome through the API
and through MCP, the 404 for a row the caller may not see, the 403 without the permission — and the security
properties of the spec's tests.access list."""
import os
import subprocess
import sys
import anyio
import httpx
import pytest
from sqlalchemy import func, select
from erp_platform import idempotency, public, work
from brick_hr.application import use_cases
from brick_hr.infrastructure import tables as T
from conftest import b64, caller, headers
from seed import hire_body, seed, sign_in
HR = headers("hr_admin")
FIN = headers("finance")
IT = headers("it")
TICK = headers("clock")
MGR = headers("manager", sub="manager.7")
OTHER_MGR = headers("manager", sub="manager.8")
EMP = headers("employee", sub="emp.1")
ASSIGNEE = headers("assignee")
@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))
s, anna = api("POST", "/hires", MGR, hire_body("Anna", "Berg", "2026-11-02"))
assert s == 200, anna
return {"lead": lead, "anna": anna}
def test_ids_leave_both_doors_as_strings(world, api, tool):
pid = world["anna"]["person_id"]
assert isinstance(pid, str) and pid.isdigit()
s, via_api = api("GET", f"/people/{pid}", HR)
kind, via_mcp = tool(HR, "show_person", {"person_id": pid})
assert s == 200 and kind == "ok"
assert via_api == via_mcp # the same answer through both doors
assert via_api["bp_id"] == world["anna"]["bp_id"] and isinstance(via_api["bp_id"], str)
def test_hr_01_a_hire_through_mcp_is_the_same_hire(world, tool, api):
body = hire_body("Anna", "Berg", "2026-11-02")
kind, again = tool(MGR, "record_hire", body)
assert kind == "ok" and again == world["anna"]
def test_another_managers_person_is_not_found_through_both_doors(world, api, tool):
pid = world["anna"]["person_id"]
assert api("GET", f"/people/{pid}", OTHER_MGR)[0] == 404
assert tool(OTHER_MGR, "show_person", {"person_id": pid})[0] == "refused"
assert tool(OTHER_MGR, "show_person", {"person_id": pid})[1].startswith("404")
assert api("GET", f"/people/{pid}", MGR)[0] == 200 # her own manager sees her
def test_a_missing_permission_is_403_naming_it(world, api, tool):
s, out = api("POST", "/hires", FIN, hire_body("X", "Y", "2026-11-02"))
assert (s, out["detail"]) == (403, "missing permission: hr:hire:record")
kind, msg = tool(FIN, "record_hire", hire_body("X", "Y", "2026-11-02"))
assert kind == "refused" and msg == "403: missing permission: hr:hire:record"
def test_hr_13_a_manager_gets_404_on_every_sensitive_route_for_their_own_report_both_doors(world, api, tool):
pid = world["anna"]["person_id"]
assert api("POST", f"/people/{pid}/sensitive", FIN, {"kind": "bank", "value": {"bank_sort_code": "1"}})[0] == 200
assert api("GET", f"/people/{pid}/sensitive", FIN)[0] == 200
for who in (MGR, IT, ASSIGNEE, EMP):
assert api("GET", f"/people/{pid}/sensitive", who)[0] == 404
assert api("POST", f"/people/{pid}/sensitive", who, {"kind": "bank", "value": "x"})[0] == 404
assert tool(who, "show_sensitive", {"person_id": pid})[1].startswith("404")
assert tool(who, "record_sensitive", {"person_id": pid, "kind": "bank", "value": "x"})[1].startswith("404")
s, me = api("GET", f"/people/{pid}", MGR)
assert "sort_code" not in str(me) and "bank" not in str(me.get("documents"))
def test_it_cannot_read_pay_and_sensitive_is_404(world, api):
pid = world["anna"]["person_id"]
api("POST", "/run-due-steps", TICK, {})
work.set_clock(at=work.parse_instant("2026-10-28T09:05 Europe/London"), freeze=True)
s, p = api("GET", f"/people/{pid}", HR)
assert "pay" in p
s, p_it = api("GET", f"/people/{pid}", IT)
assert s == 404 or "pay" not in p_it # IT sees her only with an open IT step, never pay
assert api("GET", f"/people/{pid}/sensitive", IT)[0] == 404
def test_hr_20_documents_are_streamed_after_their_class_check(world, api):
pid = world["anna"]["person_id"]
s, d = api("POST", f"/people/{pid}/documents", HR, {"document_type": "nda", "filename": "nda.pdf",
"content_base64": b64(b"%PDF nda")})
assert s == 200 and d["access_class"] == "personnel"
s, rtw = api("POST", f"/people/{pid}/documents", HR, {"document_type": "right_to_work", "filename": "pp.pdf",
"content_base64": b64(b"%PDF passport")})
assert s == 200 and rtw["access_class"] == "sensitive"
s, got = api("GET", f"/people/{pid}/documents/{d['doc_id']}", MGR)
assert s == 200 and got["content_base64"] == b64(b"%PDF nda") and "url" not in str(got).lower()
assert api("GET", f"/people/{pid}/documents/{rtw['doc_id']}", MGR)[0] == 404
assert api("GET", f"/people/{pid}/documents/{rtw['doc_id']}", HR)[1]["content_base64"] == b64(b"%PDF passport")
listed = api("GET", f"/people/{pid}/documents", MGR)[1]["documents"]
assert [x["doc_id"] for x in listed] == [d["doc_id"]]
def test_hr_24_an_employee_sees_only_their_own_record(world, api):
pid = world["anna"]["person_id"]
sign_in(lambda p, b: api("POST", p, HR, b), pid, "emp.1", get=lambda p: api("GET", p, HR))
s, me = api("GET", "/me", EMP)
assert s == 200 and me["person_id"] == pid and "first_week" in me
assert api("GET", "/me", headers("employee", sub="emp.2"))[0] == 404
assert api("GET", f"/people/{world['lead']['person_id']}", EMP)[0] == 403
def test_hr_03_an_assignee_completing_another_groups_step_gets_404(world, api):
rid = world["anna"]["run_id"]
assert api("POST", f"/runs/{rid}/steps/5/complete", ASSIGNEE, {})[0] == 404
assert api("POST", f"/runs/{rid}/steps/15/complete", IT, {})[0] == 404 # finance's step
s, mt = api("GET", "/my-tasks", ASSIGNEE)
assert s == 200 and sum(mt["counts"].values()) == 0
def test_no_identity_is_401_on_both_doors(world, api):
assert api("GET", "/people", None)[0] == 401
from brick_hr.app import app
async def go():
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://hr") as c:
async with app.router.lifespan_context(app):
return await c.post("/mcp", json={"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {}},
headers={"accept": "application/json"})
assert anyio.run(go).status_code == 401
def test_if_match_carries_the_version_and_a_stale_one_is_409(world, api):
pid = world["anna"]["person_id"]
s, p = api("GET", f"/people/{pid}", HR)
body = {"effective_date": "2026-12-01", "reason": "review", "rate_amount": "45000.0000", "rate_basis": "year"}
assert api("POST", f"/people/{pid}/pay", HR, body, headers={"If-Match": f'"{p["version"]}"'})[0] == 200
s, out = api("POST", f"/people/{pid}/pay", HR, {**body, "effective_date": "2027-01-01"},
headers={"If-Match": f'"{p["version"]}"'})
assert s == 409 and "changed since you opened it" in out["detail"]
def test_an_idempotency_key_replays_a_leave_instead_of_recording_two(world, api, engine):
idempotency.md.create_all(engine)
use_cases.HR.idempotency_connection = engine.begin
try:
pid = world["lead"]["person_id"]
body = {"last_day": "2026-12-31", "reason": "resignation"}
a = api("POST", f"/people/{pid}/leave", HR, body, headers={"Idempotency-Key": "leave-lena-1"})
b = api("POST", f"/people/{pid}/leave", HR, body, headers={"Idempotency-Key": "leave-lena-1"})
assert a == b and a[0] == 200
assert api("POST", f"/people/{pid}/leave", HR, body)[0] == 409 # without the key: already left
finally:
use_cases.HR.idempotency_connection = None
def test_hr_10_a_preview_writes_claims_and_sends_nothing(world, api, engine):
def counts():
with engine.connect() as c:
return {t.name: c.execute(select(func.count()).select_from(t)).scalar() for t in T.md.sorted_tables}
before = counts()
s, out = api("POST", "/runs/preview", HR, {"hire": {**hire_body("New", "Person", "2026-11-09"),
"personal_email": "new@home.example"}})
assert s == 200 and out["dry_run"] and len(out["steps"]) > 5
acc = next(x for x in out["steps"] if x["step_no"] == "5")
assert acc["input"]["primary_email"] == "new.person@co.example" and acc["plan"].startswith("manual: no provider")
s, out2 = api("POST", "/runs/preview", HR, {"run_id": world["anna"]["run_id"]})
assert s == 200
assert counts() == before
def test_the_intake_form_through_the_public_door_never_echoes_a_value(world, api, engine):
iid = world["anna"]["intake_id"]
s, link = api("POST", f"/intakes/{iid}/link", HR, {})
assert s == 200 and link["url"].startswith("https://public.hr.test/p/")
assert link["url"] not in str(api("GET", f"/people/{world['anna']['person_id']}", HR)[1])
papp = public.app(use_cases.HR.public_handlers)
token_path = link["url"].split("https://public.hr.test", 1)[1]
async def go(method, data=None, files=None, path=token_path):
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=papp), base_url="https://public.hr.test") as c:
return await c.request(method, path, data=data, files=files)
page = anyio.run(go, "GET")
assert page.status_code == 200 and "Home address" in page.text
r = anyio.run(go, "POST", {"home_address": "7 Secret Lane"})
assert r.status_code == 200 and "Still missing" in r.text and "7 Secret Lane" not in r.text
assert "7 Secret Lane" not in anyio.run(go, "GET").text
assert api("GET", f"/intakes/{iid}/form", HR)[0] == 404 # the staff app gets no form
assert anyio.run(go, "GET", None, None, "/p/" + "x" * 43).status_code == 404
def test_the_demo_clock_route_exists_only_when_enabled_and_never_live():
env = {k: v for k, v in os.environ.items() if k not in ("HR_DEMO_CLOCK", "DODIL_ENV")}
code = "from brick_hr.application.use_cases import HR; print(sorted(u.name for u in HR.use_cases if 'clock' in u.name))"
off = subprocess.run([sys.executable, "-c", code], env=env, capture_output=True, text=True)
on = subprocess.run([sys.executable, "-c", code], env={**env, "HR_DEMO_CLOCK": "enabled"}, capture_output=True, text=True)
live = subprocess.run([sys.executable, "-c", code], env={**env, "HR_DEMO_CLOCK": "enabled", "DODIL_ENV": "live"},
capture_output=True, text=True)
assert off.stdout.strip() == "[]"
assert "set_hr_demo_clock" in on.stdout
assert live.returncode != 0 and "refusing to start" in live.stderr
def test_the_mcp_tools_carry_a_docstring_and_the_destructive_hint(world):
names = {u.name: u for u in use_cases.HR.use_cases}
assert all(u.description for u in names.values())
assert names["cancel_run"].destructive and names["apply_erasure"].destructive
assert names["show_sensitive"].permission is None # checked inside: 404, never 403
def test_hr_17_a_transfer_outside_my_line_is_refused(world, api):
pid = world["anna"]["person_id"]
s, out = api("POST", f"/people/{pid}/transfer", MGR, {"effective_date": "2026-11-16", "org_unit_id": "CAMP"})
assert s == 409 and "outside your line" in out["detail"]
assert api("POST", f"/people/{pid}/transfer", HR, {"effective_date": "2026-11-16", "org_unit_id": "CAMP"})[0] == 200
def test_hr_41_an_engagement_about_to_end_warns_its_manager_once(world, api):
s, h = api("POST", "/hires", HR, hire_body("Tem", "Porary", "2026-10-19", worker_type="employee_fixed",
planned_end="2026-10-30"))
assert s == 200, h
api("POST", "/run-due-steps", TICK, {})
api("POST", "/run-due-steps", TICK, {})
s, board = api("GET", "/runs", HR)
warn = [r for r in board["runs"] if r["direction"] == "warning" and r["person_id"] == h["person_id"]]
assert len(warn) == 1
s, mt = api("GET", "/my-tasks", MGR)
assert any("extend it or record a leaver" in x["title"] for x in mt["overdue"] + mt["today"])
def test_rtw_a_check_after_the_start_date_is_flagged_never_refused(world, api):
"""IANA 2006 s.15: the statutory excuse needs the check BEFORE employment starts. hr records what was done and
makes no legal judgement — so a late check is recorded and flagged, and an on-time one carries no flag."""
lead, anna = world["lead"]["person_id"], world["anna"]["person_id"] # started 2024-01-08 / starts 2026-11-02
body = {"kind": "right_to_work_check", "checked_by": "hr", "check_type": "manual_document_check"}
s, late = api("POST", f"/people/{lead}/compliance", HR, {**body, "checked_on": "2026-10-19"})
assert s == 200, late
assert late["flags"] == ["the start date (2024-01-08) is before the check (2026-10-19)"]
s, on_time = api("POST", f"/people/{anna}/compliance", HR, {**body, "checked_on": "2026-10-19"})
assert s == 200 and on_time["flags"] == []
s, out = api("POST", f"/people/{anna}/compliance", HR, {"kind": "right_to_work_check", "checked_by": "hr",
"checked_on": "2026-10-19"})
assert s == 422 and "check_type" in out["detail"] # how it was done is required
def test_hr_41_a_time_limited_right_to_work_warns_hr_once_before_the_follow_up(world, api):
"""A time-limited right to work: the follow-up check date is on the record; within warning_days_right_to_work
(28) of it the tick opens ONE warning task for HR, however often it runs."""
pid = world["anna"]["person_id"]
s, r = api("POST", f"/people/{pid}/compliance", HR, {"kind": "right_to_work_check", "checked_on": "2026-10-19",
"checked_by": "hr", "check_type": "home_office_online_check",
"follow_up_on": "2026-11-10"})
assert s == 200, r
api("POST", "/run-due-steps", TICK, {})
api("POST", "/run-due-steps", TICK, {})
s, board = api("GET", "/runs", HR)
warn = [x for x in board["runs"] if x["direction"] == "warning" and x["person_id"] == pid]
assert len(warn) == 1
s, mt = api("GET", "/my-tasks", HR)
assert any("Follow-up right-to-work check" in x["title"] and "2026-11-10" in x["title"]
for x in mt["overdue"] + mt["today"])
def _raw(app, path, who, accept=None, params=None):
async def go():
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://hr") as c:
return await c.get(path, headers={**who, **({"Accept": accept} if accept else {})}, params=params)
return anyio.run(go)
def test_hr_11_the_playbook_bpmn_is_the_same_document_through_both_doors(world, api, tool):
from bpmn_check import check
r = _raw(api.app, "/playbooks/employee-onboarding/bpmn", HR)
assert r.status_code == 200 and r.headers["content-type"].startswith("application/xml")
assert r.headers["etag"] == '"1"' and 'filename="employee-onboarding.v1.bpmn"' in r.headers["content-disposition"]
kind, via_mcp = tool(HR, "show_playbook_bpmn", {"playbook_id": "employee-onboarding"})
assert kind == "ok" and via_mcp["bpmn"] == r.text and via_mcp["version"] == 1
check(r.text)
j = _raw(api.app, "/playbooks/employee-onboarding/bpmn", HR, accept="application/json")
assert j.headers["content-type"].startswith("application/json") and j.json() == via_mcp
def test_hr_11_a_new_playbook_version_shows_in_its_bpmn_and_the_old_one_stays(world, api):
s, out = api("GET", "/playbooks", HR, params={"playbook_id": "employee-onboarding"})
pb = out["playbooks"][0]
steps = pb["steps"] + [{"step_no": 16, "kind": "task", "title": "Desk booked", "responsible_group": "it",
"depends_on": [11]}]
s, kept = api("POST", "/playbooks", HR, {"playbook_id": pb["playbook_id"], "name": pb["name"],
"event_types": pb["event_types"], "worker_types": pb["worker_types"],
"steps": steps})
assert s == 200 and kept["version"] == 2, kept
v2 = _raw(api.app, "/playbooks/employee-onboarding/bpmn", HR)
v1 = _raw(api.app, "/playbooks/employee-onboarding/bpmn", HR, params={"version": 1})
assert v2.headers["etag"] == '"2"' and 'name="Desk booked"' in v2.text
assert v1.headers["etag"] == '"1"' and "Desk booked" not in v1.text
def test_hr_11_the_playbook_bpmn_needs_the_configuration_permission_and_names_what_exists(world, api, tool):
for who in (MGR, IT, FIN, EMP):
s, out = api("GET", "/playbooks/employee-onboarding/bpmn", who)
assert (s, out["detail"]) == (403, "missing permission: hr:config:write")
assert tool(who, "show_playbook_bpmn", {"playbook_id": "employee-onboarding"}) == \
("refused", "403: missing permission: hr:config:write")
s, out = api("GET", "/playbooks/no-such/bpmn", HR)
assert (s, out["detail"]) == (404, "no playbook no-such")
assert tool(HR, "show_playbook_bpmn", {"playbook_id": "employee-onboarding", "version": 9}) == \
("refused", "404: no playbook employee-onboarding version 9")
def test_hr_29_the_onboarding_monitor_through_both_doors_with_the_runs_visibility(world, api, tool):
s, out = api("GET", "/reports/onboarding", HR)
assert s == 200, out
kind, via_mcp = tool(HR, "onboarding_monitor", {})
assert kind == "ok" and via_mcp == out # the same numbers through both doors
assert {r["person_id"] for r in out["runs"]} >= {world["anna"]["person_id"], world["lead"]["person_id"]}
assert out["playbook_versions"][0]["playbook_id"] == "employee-onboarding"
assert out["playbook_versions"][0]["deviations"] == {"variation": 0, "warning": 0, "foundation": 0}
mine = api("GET", "/reports/onboarding", MGR)[1] # the manager: what the run board shows them
board = api("GET", "/runs", MGR, params={"include_done": True, "direction": "onboarding"})[1]
assert {r["run_id"] for r in mine["runs"]} == {r["run_id"] for r in board["runs"]}
assert world["anna"]["person_id"] in {r["person_id"] for r in mine["runs"]}
assert api("GET", "/reports/onboarding", OTHER_MGR)[1]["runs"] == []
s, refused = api("GET", "/reports/onboarding", IT)
assert (s, refused["detail"]) == (403, "missing permission: hr:run:monitor")
assert tool(IT, "onboarding_monitor", {})[1] == "403: missing permission: hr:run:monitor"
def _step_by(run, key_title):
return next(s for s in run["steps"] if s["title"] == key_title)
def test_hr_42_a_redundancy_offers_outplacement_and_records_the_answer(world, api, tool):
pid = world["anna"]["person_id"]
s, left = api("POST", f"/people/{pid}/leave", HR, {"last_day": "2026-12-31", "reason": "redundancy"})
assert s == 200, left
run = api("GET", f"/runs/{left['run_id']}", HR)[1]
offer = _step_by(run, "Offer outplacement support")
assert offer["state"] != "skipped" and offer["responsible_group"] == "hr"
assert offer["outcomes"] == ["accepted", "declined"] and offer["outcome"] is None
path = f"/runs/{left['run_id']}/steps/{offer['step_no']}/complete"
s, refused = api("POST", path, HR, {"output": {"outcome": "maybe"}})
assert s == 422 and "accepted, declined" in refused["detail"]
assert tool(HR, "complete_step", {"run_id": left["run_id"], "step_no": offer["step_no"]})[1].startswith("422")
s, done = api("POST", path, HR, {"output": {"outcome": "declined"}})
assert s == 200 and done["state"] == "done" and done["outcome"] == "declined"
s, again = api("POST", path, HR, {"output": {"outcome": "accepted"}})
assert s == 200 and again["outcome"] == "declined" # a done step never changes its answer
def test_hr_42_a_resignation_skips_the_outplacement_offer(world, api):
pid = world["anna"]["person_id"]
s, left = api("POST", f"/people/{pid}/leave", HR, {"last_day": "2026-12-31", "reason": "resignation"})
assert s == 200, left
assert _step_by(api("GET", f"/runs/{left['run_id']}", HR)[1], "Offer outplacement support")["state"] == "skipped"
def test_hr_43_a_retirement_runs_the_retirement_playbook_with_its_own_steps(world, api):
pid = world["anna"]["person_id"]
s, left = api("POST", f"/people/{pid}/leave", HR, {"last_day": "2026-12-31", "reason": "retirement"})
assert s == 200, left
run = api("GET", f"/runs/{left['run_id']}", HR)[1]
assert run["playbook"] == "employee-retirement v1"
titles = {x["title"]: x for x in run["steps"]}
for t in ("Tell the pension scheme they are retiring", "Knowledge handover sessions held",
"Long-service recognition", "Suspend the account", "Approve access removal on the last day"):
assert t in titles, t
assert "Offer outplacement support" not in titles
assert titles["Tell the pension scheme they are retiring"]["responsible_group"] == "finance"
assert titles["Knowledge handover sessions held"]["depends_on"] == ["1"] # after the handover list
s, other = api("POST", f"/people/{world['lead']['person_id']}/leave", HR,
{"last_day": "2026-12-31", "reason": "redundancy"})
assert api("GET", f"/runs/{other['run_id']}", HR)[1]["playbook"].startswith("employee-offboarding ")
def _leave_lead(world, api):
s, left = api("POST", f"/people/{world['lead']['person_id']}/leave", HR,
{"last_day": "2026-12-31", "reason": "resignation"})
assert s == 200, left
return left["run_id"]
def test_hr_54_without_a_connector_revoking_the_erp_login_is_a_manual_task_with_the_exact_input(world, api):
rid = _leave_lead(world, api)
step = _step_by(api("GET", f"/runs/{rid}", HR)[1], "Revoke their ERP sign-in")
assert step["manual"] is True and step["capability"] == "identity.revoke_app_user"
assert step["input"]["sub"] == "manager.7" and step["input"]["email"] == "manager.7@scenario.test"
assert step["input"]["reason"] == "left on 2026-12-31" and step["depends_on"] == ["2"]
assert "revoke" in step["plan"].lower() or "identity.revoke_app_user" in step["plan"]
def test_hr_54_with_a_connector_the_login_is_revoked_once_on_the_last_day_after_the_approval(world, api):
from erp_platform import capabilities as caps
from brick_hr.infrastructure import capability_acl as acl
calls = []
def appid(key, data, dry_run):
calls.append((key, data))
return caps.Result("done", provider="appid-fake", provider_ref=data["sub"],
output={"disabled": True, "sessions_revoked": 2})
acl.fake_connector("identity.revoke_app_user", appid)
try:
assert api("POST", "/configuration/settings", HR, {"rows": [{"key": "user_pool", "value": "acme/erp"}]})[0] == 200
rid = _leave_lead(world, api)
no = _step_by(api("GET", f"/runs/{rid}", HR)[1], "Revoke their ERP sign-in")["step_no"]
assert api("POST", f"/runs/{rid}/steps/2/approve", HR, {"granted": True})[0] == 200
work.set_clock(at=work.parse_instant("2026-12-31T17:20 Europe/London"), freeze=True)
api("POST", "/run-due-steps", TICK, {})
assert calls == [] # not before 17:30 on the last day
work.set_clock(at=work.parse_instant("2026-12-31T17:31 Europe/London"), freeze=True)
api("POST", "/run-due-steps", TICK, {})
api("POST", "/run-due-steps", TICK, {})
assert [k for k, _ in calls] == [f"{rid}|{no}"] # once: one claim key per person operation
step = _step_by(api("GET", f"/runs/{rid}", HR)[1], "Revoke their ERP sign-in")
assert step["state"] == "done" and step["provider_ref"] == "manager.7" and step["manual"] is False
assert calls[0][1] == {"pool": "acme/erp", "email": "manager.7@scenario.test", "sub": "manager.7",
"reason": "left on 2026-12-31"}
finally:
acl.fake_connector("identity.revoke_app_user", None)
# ── HR-55: when the ERP sign-in goes is the owner's dated setting, per leave reason ─────────────────────────────
POLICY = "/dated-settings/offboarding.revoke_access_at"
def test_hr_55_by_default_the_erp_login_goes_at_the_end_of_the_last_day(world, api):
s, shown = api("GET", POLICY, HR)
assert s == 200 and shown["default"] == "last_day_end" and shown["in_force"] == {"": "last_day_end"}
rid = _leave_lead(world, api) # a resignation
step = _step_by(api("GET", f"/runs/{rid}", HR)[1], "Revoke their ERP sign-in")
assert step["due_at"] == "2026-12-31T17:30:00Z" and step["depends_on"] == ["2"] # 17:30 GMT = 17:30Z
def test_hr_55_a_dismissal_set_to_immediate_revokes_the_login_once_the_access_removal_is_approved(world, api):
from erp_platform import capabilities as caps
from brick_hr.infrastructure import capability_acl as acl
calls = []
def appid(key, data, dry_run):
calls.append(key)
return caps.Result("done", provider="appid-fake", provider_ref=data["sub"],
output={"disabled": True, "sessions_revoked": 1})
s, set_ = api("POST", POLICY, HR, {"value": "immediate", "scope": "dismissal", "note": "at the meeting"})
assert s == 200, set_
assert set_["in_force"] == {"": "last_day_end", "dismissal": "immediate"} # the first row may start today
assert set_["history"][0]["set_by"] == "hr_admin.1@scenario.test" and set_["history"][0]["valid_from"] == "2026-10-19"
acl.fake_connector("identity.revoke_app_user", appid)
try:
assert api("POST", "/configuration/settings", HR, {"rows": [{"key": "user_pool", "value": "acme/erp"}]})[0] == 200
s, left = api("POST", f"/people/{world['lead']['person_id']}/leave", HR,
{"last_day": "2026-12-31", "reason": "dismissal"})
assert s == 200, left
rid = left["run_id"]
step = _step_by(api("GET", f"/runs/{rid}", HR)[1], "Revoke their ERP sign-in")
assert step["due_at"] == "2026-10-19T09:00:00Z" and step["depends_on"] == ["2"] # recorded 10:00 BST
api("POST", "/run-due-steps", TICK, {})
assert calls == [] # due, but the access removal is not approved
assert api("POST", f"/runs/{rid}/steps/2/approve", HR, {"granted": True})[0] == 200
api("POST", "/run-due-steps", TICK, {})
api("POST", "/run-due-steps", TICK, {})
assert calls == [f"{rid}|{step['step_no']}"] # once, on the day, months before the last day
finally:
acl.fake_connector("identity.revoke_app_user", None)
def test_hr_55_only_the_owner_sets_it_rows_are_dated_and_the_history_is_kept(world, api):
s, out = api("POST", POLICY, IT, {"value": "immediate"})
assert (s, out["detail"]) == (403, "missing permission: hr:offboarding:policy")
assert api("POST", POLICY, HR, {"value": "sometime"})[0] == 422
assert api("POST", POLICY, HR, {"value": "immediate", "scope": "holiday"})[0] == 422 # not a leave reason
assert api("POST", POLICY, HR, {"value": "last_day_end"})[0] == 200 # the first row: today
s, out = api("POST", POLICY, HR, {"value": "immediate", "valid_from": "2026-10-19"}) # never edited
assert s == 409 and out["gate"] == "dated_row_exists"
s, out = api("POST", POLICY, HR, {"value": "immediate", "valid_from": "2026-10-18"}) # never backdated
assert s == 409 and out["gate"] == "dated_from_tomorrow"
s, out = api("POST", POLICY, HR, {"value": "immediate"}) # a second row: tomorrow
assert s == 200 and out["in_force"][""] == "last_day_end" and out["history"][-1]["valid_from"] == "2026-10-20"
assert api("POST", POLICY, HR, {"value": "immediate", "valid_from": "2026-10-20"})[0] == 200 # a replay
assert api("POST", POLICY, HR, {"value": "last_day_end", "valid_from": "2026-10-20"})[0] == 409
rid = _leave_lead(world, api) # recorded today: the default in force today is last_day_end
work.set_clock(at=work.parse_instant("2026-10-20T10:00 Europe/London"), freeze=True)
s, shown = api("GET", POLICY, HR)
assert shown["in_force"][""] == "immediate"
assert [(r["valid_from"], r["value"], r["set_by"]) for r in shown["history"]] == [
("2026-10-19", "last_day_end", "hr_admin.1@scenario.test"), ("2026-10-20", "immediate", "hr_admin.1@scenario.test")]
step = _step_by(api("GET", f"/runs/{rid}", HR)[1], "Revoke their ERP sign-in")
assert step["due_at"] == "2026-12-31T17:30:00Z" # a run already made keeps its timing
# ── hr v4 seams: other modules read where a person belongs (HR-63) ──────────────────────────────────────────────
EXPENSES = headers("expenses_client")
def test_hr_63_the_organisational_assignment_by_person_id_through_both_doors(world, api, tool):
lead, anna = world["lead"]["person_id"], world["anna"]["person_id"]
s, got = api("GET", f"/persons/{anna}/assignment", EXPENSES, params={"on": "2026-11-10"})
assert s == 200, got
assert got == {"person_id": anna, "on": "2026-11-10", "employed": True, "company_code": "ACME-UK",
"org_unit_id": "STUDIO", "org_unit_code": "OPS", "team_code": "STUDIO", "manager_person_id": lead,
"work_email": None, "pool_user_sub": None}
kind, via_mcp = tool(EXPENSES, "show_assignment", {"person_id": anna, "on": "2026-11-10"})
assert kind == "ok" and via_mcp == got
s, mine = api("GET", f"/persons/{lead}/assignment", EXPENSES, params={"on": "2026-11-10"})
assert (mine["work_email"], mine["pool_user_sub"], mine["team_code"]) == (
"manager.7@scenario.test", "manager.7", "STUDIO")
# before the start date: employed false with nulls, never a 404
s, before = api("GET", f"/persons/{anna}/assignment", EXPENSES, params={"on": "2026-10-19"})
assert s == 200 and before["employed"] is False and before["company_code"] is None and before["manager_person_id"] is None
assert api("GET", "/persons/123456789/assignment", EXPENSES)[0] == 404
s, out = api("GET", f"/persons/{anna}/assignment", HR) # a machine permission: no person's grant reaches it
assert s == 403 and "hr:assignment:read" in str(out)
assert api("GET", f"/persons/{anna}/assignment", EXPENSES, params={"on": "2026-13-01"})[0] == 422
def test_hr_64_the_roster_of_a_company_on_a_date_paged_through_both_doors(world, api, tool):
lead, anna = world["lead"]["person_id"], world["anna"]["person_id"]
q = {"company_code": "ACME-UK", "on": "2026-11-10"}
s, everyone = api("GET", "/roster", EXPENSES, params=q)
assert s == 200, everyone
assert [p["person_id"] for p in everyone["people"]] == sorted([lead, anna], key=int) and everyone["next_offset"] is None
row = next(p for p in everyone["people"] if p["person_id"] == anna)
assert row == {"person_id": anna, "company_code": "ACME-UK", "org_unit_code": "OPS", "team_code": "STUDIO",
"manager_person_id": lead, "work_email": None, "fte": "1.0000", "employed_from": "2026-11-02",
"employed_to": None}
kind, via_mcp = tool(EXPENSES, "show_roster", q)
assert kind == "ok" and via_mcp == everyone
s, first = api("GET", "/roster", EXPENSES, params={**q, "limit": 1})
s, second = api("GET", "/roster", EXPENSES, params={**q, "limit": 1, "offset": first["next_offset"]})
assert first["next_offset"] == 1 and second["next_offset"] is None
assert first["people"] + second["people"] == everyone["people"]
team = api("GET", "/roster", EXPENSES, params={**q, "manager_person_id": lead})[1]
assert [p["person_id"] for p in team["people"]] == [anna] # direct reports of the derived manager
assert len(api("GET", "/roster", EXPENSES, params={**q, "org_unit_code": "OPS"})[1]["people"]) == 2 # and below
assert api("GET", "/roster", EXPENSES, params={**q, "org_unit_code": "CAMP"})[1]["people"] == []
# a starter after `on` is not listed yet; another company is an empty page
assert [p["person_id"] for p in api("GET", "/roster", EXPENSES, params={**q, "on": "2026-10-19"})[1]["people"]] == [lead]
assert api("GET", "/roster", EXPENSES, params={**q, "company_code": "ACME-DE"})[1]["people"] == []
assert api("GET", "/roster", HR, params=q)[0] == 403
assert api("GET", "/roster", EXPENSES, params={**q, "limit": 201})[0] == 422
# ── hr v4: every hr grant is scoped per company (hr_company_grant rows; none = every company) ─────────────────────
UK_ADMIN = headers("hr_admin", 2) # sub hr_admin.2: scoped to ACME-UK below
UK_EXPENSES = headers("expenses_client") # sub expenses_client.1: scoped to ACME-UK below
def _two_companies(api):
s, de = api("POST", "/hires", HR, hire_body("Dora", "Dahl", "2026-11-02", legal_entity_ref="ACME-DE"))
assert s == 200, de
s, out = api("POST", "/configuration/company_grant", HR, {"rows": [
{"principal": "hr_admin.2", "company_code": "ACME-UK"},
{"principal": "expenses_client.1@scenario.test", "company_code": "ACME-UK"}]})
assert s == 200, out
return de["person_id"]
def test_v4_a_grant_scoped_to_company_1_cannot_read_company_2_through_both_doors(world, api, tool):
de, anna = _two_companies(api), world["anna"]["person_id"]
# the person of company 2 does not exist for user A, through every door; company 1's does
assert api("GET", f"/people/{de}", UK_ADMIN)[0] == 404
assert tool(UK_ADMIN, "show_person", {"person_id": de})[0] == "refused"
assert api("GET", f"/people/{anna}", UK_ADMIN)[0] == 200
assert api("GET", f"/people/{de}/sensitive", UK_ADMIN)[0] == 404
assert api("POST", "/privacy/requests", UK_ADMIN, {"person_id": de, "kind": "access",
"received_on": "2026-10-19"})[0] == 404
listed = {p["person_id"] for p in api("GET", "/people", UK_ADMIN)[1]["people"]}
assert anna in listed and de not in listed
head = api("GET", "/reports/headcount", UK_ADMIN, params={"on": "2026-11-10"})[1]
assert head["headcount"] == 2 # Lena and Anna; never Dora
s, out = api("POST", "/hires", UK_ADMIN, hire_body("Eva", "Ek", "2026-11-09", legal_entity_ref="ACME-DE"))
assert s == 404, out # cannot hire into company 2 either
assert api("POST", "/payroll/extracts", UK_ADMIN, {"legal_entity_ref": "ACME-DE", "period": "2026-11"})[0] == 404
# a service client's reads stop at its companies too (matched on its email here)
assert api("GET", f"/persons/{de}/assignment", UK_EXPENSES)[0] == 404
assert api("GET", f"/persons/{anna}/assignment", UK_EXPENSES)[0] == 200
q = {"on": "2026-11-10"}
assert api("GET", "/roster", UK_EXPENSES, params={**q, "company_code": "ACME-DE"})[1]["people"] == []
assert len(api("GET", "/roster", UK_EXPENSES, params={**q, "company_code": "ACME-UK"})[1]["people"]) == 2
# a grant no row names covers every company (the spec's "none named")
assert api("GET", f"/people/{de}", HR)[0] == 200
assert [p["person_id"] for p in api("GET", "/roster", EXPENSES_DE, params={**q, "company_code": "ACME-DE"})[1]
["people"]] == [de]
EXPENSES_DE = headers("expenses_client", 2) # no row names it
def test_v4_a_scoped_grant_cannot_widen_itself_and_nobody_changes_their_own(world, api):
_two_companies(api)
s, out = api("POST", "/configuration/company_grant", UK_ADMIN,
{"rows": [{"principal": "hr_admin.2", "company_code": "ACME-DE"}]})
assert s == 409 and "every company" in str(out)
s, out = api("POST", "/configuration/company_grant", UK_ADMIN,
{"rows": [{"principal": "hr_admin.3", "company_code": "ACME-DE"}]})
assert s == 409
s, out = api("POST", "/configuration/company_grant", HR, # the unscoped admin, on themselves: by email
{"rows": [{"principal": "HR_ADMIN.1@scenario.test", "company_code": "ACME-UK"}]})
assert s == 409 and "own company grant" in str(out)
s, out = api("POST", "/configuration/company_grant", HR, {"remove": [{"principal": "hr_admin.1",
"company_code": "ACME-UK"}]})
assert s == 409 # and by sub
rows = api("GET", "/configuration/company_grant", HR)[1]["rows"]
assert {(r["principal"], r["company_code"], r["granted_by"]) for r in rows} == {
("hr_admin.2", "ACME-UK", "hr_admin.1@scenario.test"),
("expenses_client.1@scenario.test", "ACME-UK", "hr_admin.1@scenario.test")}