"""hr's scenario catalogue: SC-HR-01 … SC-HR-32 (modules/hr tests.scenarios), each mapped to its stories, written
ONCE over the composition's doors — routes, MCP tools, the public link — never a table.
erp-scenarios list tests/scenarios/catalogue.py
erp-scenarios run tests/scenarios/catalogue.py # in process: hr + bp + a test provider
Every scenario makes its own people, its own org units (team()) and, where money is compared, its own legal entity,
so none leans on another's rows or order (--shuffle). Scenarios about the first demo's "no provider" path make every
capability absent (no_providers). LIVE: run against a demo deployment with the dates relative to today; the clock
scenarios are skipped there unless the deployment mounts the demo clock (HR_DEMO_CLOCK=enabled), and the fault
scenarios are always skipped (tag failure).
"""
import base64
import pathlib
import sys
from datetime import date, datetime, timedelta
HERE = pathlib.Path(__file__).resolve().parent
sys.path.insert(0, str(HERE.parent / "scenario_bricks"))
sys.path.insert(0, str(HERE.parent))
from erp_platform.scenarios import Catalogue, Rows # noqa: E402
from seed import DOMAIN, ORG, SINCE, seed # noqa: E402
class _Drafter:
"""In process the welcome email, the handover and the leaver letter are drafted by this stand-in for Models
(live, the real adapter at api.dodil.io answers). It is given what the real one would be — and the scenarios
check it is never given a sensitive field."""
seen: list = []
def draft(self, instructions, fields, model):
self.seen.append(dict(fields))
return f"Dear {fields.get('preferred_name') or 'colleague'}, welcome. ({model})"
def _install_drafter():
from brick_hr.infrastructure import models_adapter
models_adapter.DRAFTER = DRAFTER
DRAFTER = _Drafter()
_install_drafter()
CAT = Catalogue("hr", compose=HERE / "compose.yaml", roles=HERE / "roles.yaml", ticks=["POST /hr/run-due-steps"],
tick_as="clock", clock="hr", zone="Europe/London")
CAPS = ("identity.provision_user", "identity.suspend_user", "identity.delete_user", "identity.invite_app_user",
"groups.add_member", "groups.remove_member", "chat.add_to_space", "chat.remove_from_space",
"drive.create_folder", "drive.transfer_ownership", "drive.transfer_status", "mail.send", "mail.set_autoreply",
"calendar.create_event", "contracts.draft", "contracts.send", "contracts.status", "contracts.fetch_executed")
ALL = ["hr_admin", "manager", "it", "finance", "assignee", "employee", "clock"]
# ── helpers (all through the doors) ────────────────────────────────────────────────────────
def admin(s):
return s.as_("hr_admin")
def setup(s):
a = admin(s)
seed(lambda p, b: (lambda r: (r.status_code, r.body))(a.post(p, b)), prefix="/hr")
def no_providers(s):
"""The first demo: no connector anywhere — every action is a manual task (rule 8)."""
if s.mode == "inprocess":
for c in CAPS:
s.fault.capability(c, "absent")
def team(s, tag="T"):
"""A fresh org subtree of this scenario's own: OPS-k ← STUDIO-k, CAMP-k, each with a chief position."""
a = admin(s)
k = s.key(tag).lower()
t = {"ops": f"OPS-{k}", "studio": f"STUDIO-{k}", "camp": f"CAMP-{k}", "k": k}
for uid, name, parent in ((t["ops"], f"Operations {k}", None), (t["studio"], f"Studio {k}", t["ops"]),
(t["camp"], f"Campaigns {k}", t["ops"])):
a.post("/hr/org", {"org_id": ORG, "kind": "unit", "id": uid, "valid_from": SINCE,
"values": {"name": name, "parent_org_unit_id": parent, "location_code": "LON"}}).ok()
a.post("/hr/org", {"org_id": ORG, "kind": "position", "id": f"P-{uid}", "valid_from": SINCE,
"values": {"org_unit_id": uid, "job_code": "studio_lead"}}).ok()
a.post("/hr/org", {"org_id": ORG, "kind": "unit", "id": uid, "valid_from": SINCE,
"values": {"chief_position_id": f"P-{uid}"}}).ok()
a.post("/hr/configuration/unit_access", {"rows": [
{"org_unit_id": t["studio"], "kind": "group", "target": f"team-studio-{k}@{DOMAIN}", "role": "member"},
{"org_unit_id": t["studio"], "kind": "chat_space", "target": f"Studio {k}"},
{"org_unit_id": t["camp"], "kind": "group", "target": f"team-camp-{k}@{DOMAIN}", "role": "member"},
{"org_unit_id": t["camp"], "kind": "chat_space", "target": f"Campaigns {k}"}]}).ok()
return t
def body(s, given, start, unit, *, job="designer", wt="employee_ft", family=None, **extra):
b = {"legal_given_name": given, "legal_family_name": family or s.key("F").split("-")[-1].title(),
"worker_type": wt, "job_code": job, "org_unit_id": unit, "start_date": start, "location_code": "LON",
"schedule_code": "FT", "personal_email": s.email(given.lower())}
if wt.startswith("employee") and "rate_amount" not in extra:
extra.update(rate_amount="42000.0000", rate_basis="year", currency="GBP")
b.update(extra)
return b
def hire(s, given, start, unit, who="hr_admin", n=1, **kw):
return s.as_(who, n).post("/hr/hires", body(s, given, start, unit, **kw)).ok()
def sign_in(s, person_id, sub):
a = admin(s)
p = a.get(f"/hr/people/{person_id}").ok()
first = p["detail_history"][0]["valid_from"]
a.post(f"/hr/people/{person_id}/corrections", {"record": "detail", "valid_from": first,
"values": {"work_email": f"{sub}@scenario.test"}}).ok()
inv = a.post(f"/hr/people/{person_id}/invite", {}).ok()
if inv["state"] != "done":
a.post(f"/hr/runs/{inv['run_id']}/steps/1/complete",
{"output": {"sub": sub, "email": f"{sub}@scenario.test", "created": True}}).ok()
def lead(s, t, unit_key, n, start="2024-01-08", given="Lena"):
"""A person holding the unit's chief position since `start`, signed in as persona manager#n. The hire is
recorded a few days after the start (the clock is put back after), as HR would have."""
unit = t[unit_key]
back = s.clock.now()
s.clock.set(f"{start}T12:00Z")
s.clock.advance("+3d")
p = hire(s, given, start, unit, job="studio_lead", position_id=f"P-{unit}")
s.clock.set(back)
sign_in(s, p["person_id"], f"manager.{n}")
return p["person_id"]
def run(s, rid, who="hr_admin"):
return s.as_(who).get(f"/hr/runs/{rid}").ok()
def steps(r):
return {x["step_no"]: x for x in r["steps"]}
def attach_contract(s, rid, no="3"):
return admin(s).post(f"/hr/runs/{rid}/steps/{no}/complete", {"attachment": {
"filename": "signed.pdf", "content_type": "application/pdf",
"content_base64": base64.b64encode(b"%PDF signed").decode()}}).ok()
def rtw(s, pid, on):
return admin(s).post(f"/hr/people/{pid}/compliance", {"kind": "right_to_work_check", "checked_on": on,
"checked_by": "hr", "check_type": "manual_document_check"}).ok()
def partner(s, bp_id):
return s.as_("finance").get(f"/business-partner/partners/{bp_id}")
def pay_slices(p, n="1"):
return [x for x in p["pay"][n]]
# ── SC-HR-01 … 28 ──────────────────────────────────────────────────────────────────────────
@CAT.scenario("SC-HR-01", "future-dated hire", stories=["HR-01", "HR-38", "HR-39"], tags=["clock"],
personas=["hr_admin", "manager", "finance", "clock"])
def future_hire(s):
s.given("today is 2026-10-17; no person with Anna's personal email")
s.clock.set("2026-10-17T10:00 Europe/London")
setup(s)
t = team(s, "A")
lead(s, t, "studio", 101)
s.when("a manager records a hire starting 2026-11-02")
h = hire(s, "Anna", "2026-11-02", t["studio"], who="manager", n=101)
s.then("person, engagement 1 (future), assignment from 2026-11-02 and an onboarding run exist today")
p = admin(s).get(f"/hr/people/{h['person_id']}").ok()
p.field("engagements.0.status", "future").field("engagements.0.assignments.0.valid_from", "2026-11-02")
s.check(h["run_id"].startswith("onboarding:"), "an onboarding run id", h)
run(s, h["run_id"]).field("state", "open")
s.then("headcount on 2026-10-31 excludes her, on 2026-11-02 includes her")
before = admin(s).get("/hr/reports/headcount", on="2026-10-31").ok()
after = admin(s).get("/hr/reports/headcount", on="2026-11-02").ok()
count = lambda r: next((g["headcount"] for g in r["groups"] if g["group"] == t["studio"]), 0) # noqa: E731
s.check(count(after) - count(before) == 1, "she counts from her start date", (before.body, after.body))
s.then("her business partner exists now, with no employee role until the 2026-11-02 tick")
partner(s, h["bp_id"]).ok().field("roles", []).field("kind", "person")
s.clock.set("2026-11-02T08:00 Europe/London")
s.tick()
partner(s, h["bp_id"]).ok().field("roles", ["employee"])
hist = s.as_("finance").get(f"/business-partner/partners/{h['bp_id']}/history").ok()["events"]
s.tick()
again = s.as_("finance").get(f"/business-partner/partners/{h['bp_id']}/history").ok()["events"]
s.check(len(again) == len(hist), "a second tick that day changes nothing", again)
@CAT.scenario("SC-HR-02", "a change versus a correction", stories=["HR-14", "HR-39"], tags=["money"],
personas=["hr_admin"])
def change_vs_correction(s):
s.given("a pay slice 42,000.0000 GBP/year from 2026-03-01")
s.clock.set("2026-03-05T10:00 Europe/London")
setup(s)
t = team(s, "B")
h = hire(s, "Ben", "2026-03-01", t["studio"])
pid = h["person_id"]
s.when("(a) a PayChanged to 45,000.0000 from 2026-04-01")
s.clock.set("2026-03-20T10:00 Europe/London")
admin(s).post(f"/hr/people/{pid}/pay", {"effective_date": "2026-04-01", "reason": "annual review",
"rate_amount": "45000.0000", "rate_basis": "year"}).ok()
s.then("the March slice is delimited at 2026-03-31; a new slice opens 2026-04-01")
p = admin(s).get(f"/hr/people/{pid}").ok()
pay = pay_slices(p)
s.check(pay[0]["valid_to"] == "2026-03-31" and pay[1]["valid_from"] == "2026-04-01"
and pay[1]["rate_amount"] == "45000.0000", "delimited, not overwritten", pay)
s.when("(b) then a correction of the 2026-04-01 slice to 45,500.0000")
s.clock.set("2026-04-20T10:00 Europe/London")
admin(s).post(f"/hr/people/{pid}/corrections", {"record": "pay", "valid_from": "2026-04-01", "values": {
"rate_amount": "45500.0000", "rate_basis": "year", "currency": "GBP"}, "reason": "typo"}).ok()
pay = pay_slices(admin(s).get(f"/hr/people/{pid}").ok())
april = [x for x in pay if x["valid_from"] == "2026-04-01"]
s.then("the first 2026-04-01 slice is superseded (still readable); the new one holds 45,500.0000")
s.check(len(april) == 2 and april[0]["superseded_by"] == 2 and not april[0]["active"]
and april[1]["active"] and april[1]["rate_amount"] == "45500.0000", "a correction keeps the old", april)
s.then("what we believed on 2026-04-15, before the correction (asked on 2026-04-10), is 45,000.0000")
t0 = "2026-04-10T00:00:00Z"
believed = [x for x in pay if x["valid_from"] <= "2026-04-15" <= (x["valid_to"] if x["valid_to"] != "open" else "9999")
and x["recorded_at"] <= t0 and (x["superseded_at"] is None or x["superseded_at"] > t0)]
s.check(len(believed) == 1 and believed[0]["rate_amount"] == "45000.0000", "the belief at the time", believed)
s.then("TC1 holds after both")
active = [x for x in pay if x["active"]]
s.check(all(date.fromisoformat(b["valid_from"]) == date.fromisoformat(a["valid_to"]) + timedelta(days=1)
for a, b in zip(active, active[1:])), "no gap, no overlap", active)
@CAT.scenario("SC-HR-03", "retro pay after the batch was sent", stories=["HR-14", "HR-19"], tags=["money"],
personas=["hr_admin", "finance"])
def retro(s):
s.clock.set("2026-10-05T10:00 Europe/London")
setup(s)
t = team(s, "C")
le = s.key("LE")
h = hire(s, "Cara", "2026-10-01", t["studio"], legal_entity_ref=le)
s.given("the November 2026 batch for the entity is sent")
s.clock.set("2026-11-28T10:00 Europe/London")
fin = s.as_("finance")
nov = fin.post("/hr/payroll/extracts", {"legal_entity_ref": le, "period": "2026-11"}).ok()
fin.post(f"/hr/payroll/extracts/{nov['batch_id']}/sent", {}).ok().field("state", "sent")
s.when("on 2026-12-10 a PayChanged effective 2026-11-01 is recorded")
s.clock.set("2026-12-10T10:00 Europe/London")
admin(s).post(f"/hr/people/{h['person_id']}/pay", {"effective_date": "2026-11-01", "reason": "annual review",
"rate_amount": "44000.0000", "rate_basis": "year"}).ok()
dec = fin.post("/hr/payroll/extracts", {"legal_entity_ref": le, "period": "2026-12"}).ok()
s.then("the November batch is unchanged; December carries a retro line for 2026-11 with its proration dates")
fin.get(f"/hr/payroll/extracts/{nov['batch_id']}").ok().same_as(
fin.get(f"/hr/payroll/extracts/{nov['batch_id']}"))
nov_now = fin.get(f"/hr/payroll/extracts/{nov['batch_id']}").ok()
s.check(nov_now["lines"] == nov["lines"], "the sent batch is frozen", nov_now["lines"])
retro_lines = [x for x in dec["lines"] if x["change_kind"] == "retro"]
s.check(len(retro_lines) == 1 and retro_lines[0]["for_period"] == "2026-11"
and retro_lines[0]["prorate_from"] == "2026-11-01" and retro_lines[0]["prorate_to"] == "2026-11-30"
and retro_lines[0]["rate_amount"] == "44000.0000", "one retro line", dec["lines"])
seen = [(x["for_period"], x["slice_from"], x["rate_amount"]) for b in (nov, dec) for x in b["lines"]]
s.check(len(seen) == len(set(seen)), "no line appears twice across batches", seen)
@CAT.scenario("SC-HR-04", "mid-month transfer", stories=["HR-17", "HR-18", "HR-19"], tags=["clock"],
personas=["hr_admin", "finance"])
def mid_month(s):
s.given("a Designer in Studio since 2026-01-05")
s.clock.set("2026-01-10T10:00 Europe/London")
setup(s)
t = team(s, "D")
le = s.key("LE")
h = hire(s, "Dana", "2026-01-05", t["studio"], legal_entity_ref=le)
s.when("a Transferred event to Campaigns from 2026-11-16 is recorded on 2026-11-10")
s.clock.set("2026-11-10T10:00 Europe/London")
tr = admin(s).post(f"/hr/people/{h['person_id']}/transfer", {
"effective_date": "2026-11-16", "org_unit_id": t["camp"], "job_code": "producer"}).ok()
s.then("the Studio slice ends 2026-11-15 and the Campaigns slice starts 2026-11-16")
a = admin(s).get(f"/hr/people/{h['person_id']}").ok()["engagements"][0]["assignments"]
s.check(a[0]["valid_to"] == "2026-11-15" and a[1]["valid_from"] == "2026-11-16"
and a[1]["org_unit_id"] == t["camp"], "two slices", a)
s.then("headcount counts her in Studio on the 15th and in Campaigns on the 16th")
g15 = {g["group"]: g["headcount"] for g in admin(s).get("/hr/reports/headcount", on="2026-11-15").ok()["groups"]}
g16 = {g["group"]: g["headcount"] for g in admin(s).get("/hr/reports/headcount", on="2026-11-16").ok()["groups"]}
s.check(g15.get(t["studio"]) == 1 and not g15.get(t["camp"]) and g16.get(t["camp"]) == 1
and not g16.get(t["studio"]), "as the org was on each day", (g15, g16))
s.then("the November extract carries both slices with their proration dates")
lines = s.as_("finance").post("/hr/payroll/extracts", {"legal_entity_ref": le, "period": "2026-11"}).ok()["lines"]
spans = sorted((x["prorate_from"], x["prorate_to"], x["org_unit_id"]) for x in lines if x["component"] == "base")
s.check(spans == [("2026-11-01", "2026-11-15", t["studio"]), ("2026-11-16", "2026-11-30", t["camp"])],
"two lines", lines)
s.then("the change run's access delta is due on 2026-11-16, not on the 10th")
r = steps(run(s, tr["run_id"]))
adds = [x for x in r.values() if x["capability"] == "groups.add_member"]
s.check(adds and all(x["due_at"] == "2026-11-16T09:00:00Z" for x in adds)
and any(x["target"] == f"team-camp-{t['k']}@{DOMAIN}" for x in adds), "the delta on the day", adds)
removes = [x for x in r.values() if x["capability"] == "groups.remove_member"]
s.check(any(x["target"] == f"team-studio-{t['k']}@{DOMAIN}" for x in removes), "the old team removed", removes)
@CAT.scenario("SC-HR-05", "a reorganisation changes the derived manager", stories=["HR-12", "HR-39"],
tags=["clock", "permissions"], personas=["hr_admin", "manager"])
def reorg(s):
s.given("Studio (chief held by Lena) under Operations (chief held by Omar); Ben assigned to Studio")
s.clock.set("2026-11-20T10:00 Europe/London")
setup(s)
t = team(s, "E")
lena = lead(s, t, "studio", 51, given="Lena")
lead(s, t, "ops", 52, given="Omar")
s.clock.set("2025-01-10T10:00 Europe/London")
ben = hire(s, "Ben", "2025-01-06", t["studio"])
s.clock.set("2026-11-20T10:00 Europe/London")
s.when("Lena leaves on 2026-11-30, and Studio has no chief holder from 2026-12-01")
admin(s).post(f"/hr/people/{lena}/leave", {"last_day": "2026-11-30", "reason": "resignation"}).ok()
s.then("ManagerOf(Ben, 2026-11-30) is Lena; ManagerOf(Ben, 2026-12-01) is Omar — no write to Ben's record")
s.clock.set("2026-11-30T10:00 Europe/London")
b1 = admin(s).get(f"/hr/people/{ben['person_id']}").ok()
b1.field("manager.person_id", lena)
omar = s.as_("manager", 52)
s.check(not any(r["person_id"] == ben["person_id"] for r in omar.get("/hr/runs").ok()["runs"]),
"Omar does not see Ben's runs before 2026-12-01")
s.clock.set("2026-12-01T10:00 Europe/London")
b2 = admin(s).get(f"/hr/people/{ben['person_id']}").ok()
s.check(b2["manager"]["name"].startswith("Omar"), "Omar, walking up", b2["manager"])
s.check(b2["version"] == b1["version"], "Ben's record was not written", (b1["version"], b2["version"]))
s.check(any(r["person_id"] == ben["person_id"] for r in omar.get("/hr/runs").ok()["runs"]),
"Omar sees Ben's runs from 2026-12-01")
s.then("a manager override on Ben's assignment wins over both")
zed = hire(s, "Zed", "2026-12-01", t["camp"])
admin(s).post(f"/hr/people/{ben['person_id']}/transfer", {"effective_date": "2026-12-02",
"manager_override_person_id": zed["person_id"]}).ok()
s.clock.set("2026-12-02T10:00 Europe/London")
admin(s).get(f"/hr/people/{ben['person_id']}").ok().field("manager.person_id", zed["person_id"])
@CAT.scenario("SC-HR-06", "leave, then rehire", stories=["HR-06", "HR-15"], personas=["hr_admin"])
def leave_rehire(s):
s.given("an employee on engagement 1")
s.clock.set("2026-03-05T10:00 Europe/London")
setup(s)
t = team(s, "F")
h = hire(s, "Finn", "2026-03-02", t["studio"])
pid = h["person_id"]
s.when("a Left event (last day 2026-11-30) and, the same afternoon, a Rehired event")
s.clock.set("2026-11-20T14:00 Europe/London")
left = admin(s).post(f"/hr/people/{pid}/leave", {"last_day": "2026-11-30", "reason": "resignation"}).ok()
re_body = {"start_date": "2026-11-30", "worker_type": "employee_ft", "job_code": "designer",
"org_unit_id": t["studio"], "rate_amount": "43000.0000", "rate_basis": "year"}
s.then("a rehire starting on 2026-11-30 is refused, naming the first day the new one can start")
admin(s).post(f"/hr/people/{pid}/rehire", re_body).refused(
409, "engagement 1 is still active on 30 Nov; start the new one on 1 Dec")
again = admin(s).post(f"/hr/people/{pid}/rehire", {**re_body, "start_date": "2026-12-01"}).ok()
s.then("engagement 1 ends 2026-11-30, engagement 2 starts 2026-12-01; one offboarding and one onboarding run")
p = admin(s).get(f"/hr/people/{pid}").ok()
s.check(p["engagements"][0]["actual_end"] == "2026-11-30" and p["engagements"][1]["start_date"] == "2026-12-01",
"two engagements", p["engagements"])
runs = [r["run_id"] for r in p["runs"]]
s.check(sum(r.startswith("offboarding:") for r in runs) == 1 and again["run_id"] in runs
and again["run_id"].startswith(f"onboarding:{pid}:2:"), "the rehire made no offboarding run", runs)
s.check(left["run_id"].startswith("offboarding:"), "the leave's run", left)
@CAT.scenario("SC-HR-07", "freelancer converted to employee", stories=["HR-16", "HR-50"], tags=["clock"],
personas=["hr_admin", "finance", "clock"])
def conversion(s):
s.given("a sole-trader freelancer on engagement 1")
s.clock.set("2026-10-05T10:00 Europe/London")
setup(s)
t = team(s, "G")
le = s.key("LE")
h = hire(s, "Gus", "2026-10-01", t["studio"], wt="freelancer_sole", rate_amount="450.0000", rate_basis="day",
currency="GBP", legal_entity_ref=le)
s.tick()
partner(s, h["bp_id"]).ok().field("roles", ["supplier", "worker"])
s.when("a Converted event effective 2026-12-01")
s.clock.set("2026-11-20T10:00 Europe/London")
c = admin(s).post(f"/hr/people/{h['person_id']}/conversion", {
"effective_date": "2026-12-01", "worker_type": "employee_ft", "rate_amount": "52000.0000",
"rate_basis": "year", "legal_entity_ref": le}).ok()
s.then("engagement 1 ends 2026-11-30 (converted), engagement 2 starts 2026-12-01, in one event")
p = admin(s).get(f"/hr/people/{h['person_id']}").ok()
s.check(p["engagements"][0]["actual_end"] == "2026-11-30" and p["engagements"][0]["leave_reason"] == "converted"
and p["engagements"][1]["start_date"] == "2026-12-01" and p["engagements"][1]["worker_type"] == "employee_ft",
"converted", p["engagements"])
s.check(sum(e["kind"] == "Converted" for e in p["events"]) == 1, "one event", p["events"])
runs = [r["run_id"] for r in p["runs"]]
s.check([r.split(":")[0] for r in runs].count("conversion") == 1 and not any(r.startswith("offboarding") for r in runs),
"exactly one conversion run and no offboarding", runs)
for rid in runs:
caps = {x["capability"] for x in run(s, rid)["steps"]}
s.check(not caps & {"identity.suspend_user", "groups.remove_member", "identity.delete_user",
"chat.remove_from_space"}, "no access-removal step anywhere", (rid, caps))
s.then("the partner loses worker and supplier and gains employee on 2026-12-01")
s.clock.set("2026-12-01T08:00 Europe/London")
s.tick()
partner(s, h["bp_id"]).ok().field("roles", ["employee"])
s.then("the November extract has no line for engagement 1 (a freelancer is paid as a supplier)")
lines = s.as_("finance").post("/hr/payroll/extracts", {"legal_entity_ref": le, "period": "2026-11"}).ok()["lines"]
s.check(not [x for x in lines if x["person_id"] == h["person_id"] and x["n"] == 1], "no freelancer line", lines)
del c
@CAT.scenario("SC-HR-08", "two admins change the same person at once", stories=["HR-14", "HR-13"],
tags=["concurrency"], personas=["hr_admin", "hr_admin#2", "finance", "finance#2"])
def two_admins(s):
s.clock.set("2026-10-05T10:00 Europe/London")
setup(s)
t = team(s, "H")
h = hire(s, "Hana", "2026-10-01", t["studio"])
pid = h["person_id"]
s.given("two HR admins open the same person")
v = admin(s).get(f"/hr/people/{pid}").ok()["version"]
s.when("both save a different pay slice at the same moment")
rs = s.concurrently(2, lambda i: s.as_("hr_admin", i + 1).post(f"/hr/people/{pid}/pay", {
"effective_date": "2026-11-01", "reason": "review", "rate_amount": f"4{i + 5}000.0000", "rate_basis": "year",
"expected_version": v}))
s.then('one commits; the other is refused 409 "changed since you opened it" and nothing of it is written')
rs.expect({200: 1, 409: 1})
rs.where(409)[0].refused(409, "changed since you opened it")
pay = [x for x in pay_slices(admin(s).get(f"/hr/people/{pid}").ok()) if x["active"]]
s.check(len(pay) == 2 and pay[0]["valid_to"] == "2026-10-31", "TC1 holds: one November slice", pay)
s.then("retrying after a reload succeeds as a normal change")
v2 = admin(s).get(f"/hr/people/{pid}").ok()["version"]
admin(s).post(f"/hr/people/{pid}/pay", {"effective_date": "2026-12-01", "reason": "review",
"rate_amount": "47000.0000", "rate_basis": "year",
"expected_version": v2}).ok()
s.then("the same holds on the Sensitive root")
fin = s.as_("finance")
fin.post(f"/hr/people/{pid}/sensitive", {"kind": "bank", "value": {"sort_code": "10-20-30"}}).ok()
sv = fin.get(f"/hr/people/{pid}/sensitive").ok()["version"]
rs = s.concurrently(2, lambda i: s.as_("finance", i + 1).post(f"/hr/people/{pid}/sensitive", {
"kind": "bank", "value": {"sort_code": f"10-20-3{i}"}, "correction": True, "expected_version": sv}))
rs.expect({200: 1, 409: 1})
@CAT.scenario("SC-HR-09", "no provider wired", stories=["HR-31", "HR-36", "HR-03"], tags=["failure", "capability"],
personas=["hr_admin", "it", "clock"])
def no_provider(s):
s.clock.set("2026-10-19T10:00 Europe/London")
setup(s)
t = team(s, "I")
with s.fault.scope() as f:
s.given("a composition mapping no provider to identity.provision_user (nor anything else)")
for c in CAPS:
f.capability(c, "absent")
h = hire(s, "Ivo", "2026-11-02", t["studio"])
s.tick()
attach_contract(s, h["run_id"])
rtw(s, h["person_id"], "2026-10-20")
s.when("the account step falls due (start −3 working days, 09:00)")
s.clock.set("2026-10-28T09:05 Europe/London")
s.tick()
s.then("a manual task for IT with the capability, the exact validated input and the plan in words")
tasks = s.as_("it").get("/hr/my-tasks").ok()
acc = next(x for x in tasks["overdue"] + tasks["today"] if x["run_id"] == h["run_id"] and x["step_no"] == "5")
s.check(acc["manual"] and acc["capability"] == "identity.provision_user"
and acc["input"]["primary_email"].endswith(f"@{DOMAIN}") and acc["input"]["given_name"] == "Ivo"
and acc["input"]["org_unit"].endswith(f"/Studio {t['k']}") and "work account" in acc["plan"].lower(),
"the exact input and the plan", acc)
s.when("IT marks it done with the provider_ref and the work email")
made = f"ivo.made@{DOMAIN}"
s.as_("it").post(f"/hr/runs/{h['run_id']}/steps/5/complete", {
"provider_ref": "acct-1", "output": {"account_id": "acct-1", "primary_email": made, "created": True}}).ok()
s.then("the email is written back as a person-detail slice")
admin(s).get(f"/hr/people/{h['person_id']}").ok().field("detail.work_email", made)
s.then("with a provider wired, the same playbook on a new starter makes the step automatic — no playbook change")
s.clock.set("2026-10-19T10:00 Europe/London")
h2 = hire(s, "Ida", "2026-11-02", t["studio"])
rtw(s, h2["person_id"], "2026-10-20")
s.tick()
s.clock.set("2026-10-20T10:00 Europe/London")
s.tick() # the test provider signs the contract
s.clock.set("2026-10-28T09:05 Europe/London")
s.tick()
st = steps(run(s, h2["run_id"]))["5"]
s.check(st["state"] == "done" and not st["manual"] and st["provider_ref"], "automatic", st)
s.check(run(s, h["run_id"])["playbook"] == run(s, h2["run_id"])["playbook"], "the same playbook version")
def _due_account(s, tag, given="Pat"):
"""A starter whose account step is due now, with providers present (contract signed by the test provider)."""
s.clock.set("2026-10-19T10:00 Europe/London")
setup(s)
t = team(s, tag)
h = hire(s, given, "2026-11-02", t["studio"])
rtw(s, h["person_id"], "2026-10-20")
s.tick() # intake mailed, contract drafted + sent
s.clock.set("2026-10-20T10:00 Europe/London")
s.tick() # contract signed → filed
return t, h
@CAT.scenario("SC-HR-10", "provider failing", stories=["HR-36"], tags=["failure", "capability"],
personas=["hr_admin", "it", "clock"])
def provider_failing(s):
t, h = _due_account(s, "J")
with s.fault.scope() as f:
s.given("a provider that answers failed retryable three times, with max attempts 3")
f.capability("identity.provision_user", "failing")
s.when("three ticks pass, past each backoff")
s.clock.set("2026-10-28T09:01 Europe/London")
for _ in range(3):
s.tick()
s.clock.advance("+30m")
s.then("three attempts in the action log, then a manual task for IT with the provider's message in words")
log = s.as_("it").get("/hr/action-log", run_id=h["run_id"]).ok()["entries"]
fails = [e for e in log if e["step_no"] == "5" and e["what"] == "attempt failed"]
s.check(len(fails) == 3, "three attempts", fails)
st = steps(run(s, h["run_id"]))["5"]
s.check(st["state"] == "waiting" and st["manual"] and st["responsible_group"] == "it"
and "not answering" in st["manual_reason"], "a manual task for IT", st)
board = admin(s).get("/hr/runs").ok()["runs"]
mine = next(r for r in board if r["run_id"] == h["run_id"])
s.check(any(x["why"] == "failed" and x["step_no"] == "5" for x in mine["stuck"]), "the run shows stuck", mine)
@CAT.scenario("SC-HR-11", "provider needs a connection", stories=["HR-36"], tags=["failure", "capability"],
personas=["hr_admin", "it", "clock"])
def connection_required(s):
t, h = _due_account(s, "K")
with s.fault.scope() as f:
s.given("a provider that answers connection_required")
f.capability("identity.provision_user", "connection_required")
s.when("the step falls due")
s.clock.set("2026-10-28T09:01 Europe/London")
s.tick()
st = steps(run(s, h["run_id"]))["5"]
s.then("a manual task at once, naming what is missing; no retries counted")
s.check(st["manual"] and "connection" in st["manual_reason"] and st["attempts"] == 0, "at once", st)
s.then("a later tick after the connection exists does not act behind the person's back")
s.clock.advance("+1h")
s.tick()
st2 = steps(run(s, h["run_id"]))["5"]
s.check(st2["state"] == "waiting" and st2["manual"], "the manual task stands", st2)
s.as_("it").post(f"/hr/runs/{h['run_id']}/steps/5/complete", {"provider_ref": "by-hand"}).ok().field("state", "done")
def _leaver(s, tag, given, n=None):
"""An employee with an account and a manager, leaving Friday 2026-11-27, offboarding run open, removal
approved."""
s.clock.set("2026-01-07T10:00 Europe/London")
setup(s)
t = team(s, tag)
if n is not None:
lead(s, t, "studio", n, given="Mia")
h = hire(s, given, "2026-01-05", t["studio"])
sign_in(s, h["person_id"], f"leaver.{s.key('L').lower()}")
s.clock.set("2026-11-16T10:00 Europe/London")
left = admin(s).post(f"/hr/people/{h['person_id']}/leave", {"last_day": "2026-11-27", "reason": "resignation"}).ok()
admin(s).post(f"/hr/runs/{left['run_id']}/steps/2/approve", {"granted": True}).ok()
return t, h, left
@CAT.scenario("SC-HR-12", "provider still working", stories=["HR-35", "HR-36"], tags=["clock", "capability"],
personas=["hr_admin", "it", "clock"])
def still_working(s):
t, h, left = _leaver(s, "M", "Mo", n=121)
s.given("drive.transfer_status answers in_progress for two ticks, then completed")
s.clock.set("2026-11-27T17:31 Europe/London")
s.tick()
st = steps(run(s, left["run_id"]))
s.check(st["4"]["state"] == "done", "the transfer was requested on the last day", st["4"])
s.when("the delete step is approved and due")
s.clock.set("2026-12-20T09:01 Europe/London")
s.tick()
s.as_("it").post(f"/hr/runs/{left['run_id']}/steps/8/approve", {"granted": True}).ok()
s.clock.set("2026-12-27T09:01 Europe/London")
seen = []
for _ in range(4):
s.tick()
x = steps(run(s, left["run_id"]))["9"]
seen.append((x["state"], x["last_status"], x["attempts"]))
s.clock.advance("+2m")
s.then("identity.delete_user is called only after completed; in_progress never counts as a failure")
s.check(seen[0][1] == "in_progress" and seen[1][1] == "in_progress" and seen[-1][0] == "done"
and all(a == 0 for _st, _ls, a in seen), "waits, then deletes", seen)
@CAT.scenario("SC-HR-13", "approvals pending, declined, timed out", stories=["HR-07", "HR-37"],
tags=["clock", "failure"], personas=["hr_admin", "it", "clock"])
def approvals(s):
no_providers(s)
out = [_leaver(s, f"N{i}", g) for i, g in enumerate(("Ann", "Dec", "Una"))]
s.given("three runs with an approval before identity.delete_user; IT did the last-day removals by hand")
s.clock.set("2026-11-27T17:31 Europe/London")
s.tick()
for _t, _h, left in out:
for x in steps(run(s, left["run_id"])).values():
if x["state"] == "waiting" and x["capability"] in ("identity.suspend_user", "drive.transfer_ownership",
"mail.set_autoreply", "groups.remove_member",
"chat.remove_from_space"):
s.as_("it").post(f"/hr/runs/{left['run_id']}/steps/{x['step_no']}/complete", {}).ok()
s.clock.set("2026-12-20T09:01 Europe/London")
s.tick()
s.when("one is approved, one is declined with a reason, one is never answered")
it = s.as_("it")
it.post(f"/hr/runs/{out[0][2]['run_id']}/steps/8/approve", {"granted": True}).ok()
it.post(f"/hr/runs/{out[1][2]['run_id']}/steps/8/approve", {"granted": False, "reason": "legal hold on mail"}).ok()
s.then("a step is approved once: the same decision again changes nothing")
it.post(f"/hr/runs/{out[0][2]['run_id']}/steps/8/approve", {"granted": True}).ok().field("approved", True)
for _ in range(10):
s.clock.advance("+1d")
s.tick()
s.then("approved: delete runs when due (a manual task here, no provider)")
a = steps(run(s, out[0][2]["run_id"]))["9"]
s.check(a["state"] == "waiting" and a["manual"], "the delete fell due behind its approval", a)
s.then("declined: delete is cancelled and the reason is in the run log")
d = run(s, out[1][2]["run_id"])
s.check(steps(d)["9"]["state"] == "cancelled" and any("legal hold" in (l["to"] or "") for l in d["log"]),
"cancelled with the reason", d["log"][-3:])
s.then("unanswered: chased per cadence, then a task to HR 'approval overdue'; the delete never runs")
u = run(s, out[2][2]["run_id"])
us = steps(u)
chases = [l for l in u["log"] if l["what"] == "chase" and l["from"] == "8"]
s.check(len(chases) == 3 and "8:escalate" in us and "approval overdue" in us["8:escalate"]["title"]
and us["9"]["state"] == "pending", "chased, escalated, never deleted", (chases, us.get("8:escalate"), us["9"]))
@CAT.scenario("SC-HR-14", "daylight saving", stories=["HR-30", "HR-31", "HR-34"], tags=["clock"],
personas=["hr_admin"])
def dst(s):
s.given("org timezone Europe/London; British Summer Time ends on 2026-10-25")
s.clock.set("2026-10-19T10:00 Europe/London")
setup(s)
t = team(s, "O")
s.when("a start on Wednesday 2026-10-28, account step at start −3 working days 09:00")
h = hire(s, "Ola", "2026-10-28", t["studio"])
s.then("the account step is due 2026-10-23T08:00Z (09:00 BST)")
acc = steps(run(s, h["run_id"]))["5"]
s.check(acc["due_at"] == "2026-10-23T08:00:00Z", "09:00 BST is 08:00Z", acc)
s.when("a last day on Friday 2026-10-30, removal at 17:30")
h2 = hire(s, "Oli", "2026-10-01", t["studio"])
left = admin(s).post(f"/hr/people/{h2['person_id']}/leave", {"last_day": "2026-10-30", "reason": "resignation"}).ok()
s.then("the removal is due 2026-10-30T17:30Z (17:30 GMT)")
st = steps(run(s, left["run_id"]))
s.check(st["3"]["due_at"] == "2026-10-30T17:30:00Z", "17:30 GMT", st["3"])
@CAT.scenario("SC-HR-15", "leap day and month end", stories=["HR-14", "HR-19"], tags=["money"],
personas=["hr_admin", "finance"])
def leap(s):
s.clock.set("2028-02-10T10:00 Europe/London")
setup(s)
t = team(s, "P")
le = s.key("LE")
s.given("a pay slice starting 2028-02-29, and a leaver whose last day is 2028-02-29")
a = hire(s, "Pia", "2028-02-01", t["studio"], legal_entity_ref=le)
admin(s).post(f"/hr/people/{a['person_id']}/pay", {"effective_date": "2028-02-29", "reason": "promotion",
"rate_amount": "48000.0000", "rate_basis": "year"}).ok()
b = hire(s, "Per", "2028-02-01", t["studio"], legal_entity_ref=le)
admin(s).post(f"/hr/people/{b['person_id']}/leave", {"last_day": "2028-02-29", "reason": "resignation"}).ok()
s.then("the previous slice ends 2028-02-28; the leaver's pay is delimited at 2028-03-01 (ends 2028-02-29)")
pa = [x for x in pay_slices(admin(s).get(f"/hr/people/{a['person_id']}").ok()) if x["active"]]
pb = [x for x in pay_slices(admin(s).get(f"/hr/people/{b['person_id']}").ok()) if x["active"]]
s.check(pa[0]["valid_to"] == "2028-02-28" and pa[1]["valid_from"] == "2028-02-29", "no gap on the 29th", pa)
s.check(pb[-1]["valid_to"] == "2028-02-29", "delimited at 1 March", pb)
lines = s.as_("finance").post("/hr/payroll/extracts", {"legal_entity_ref": le, "period": "2028-02"}).ok()["lines"]
leaver = [x for x in lines if x["person_id"] == b["person_id"]]
s.check(leaver and leaver[-1]["prorate_to"] == "2028-02-29" and leaver[-1]["change_kind"] in ("leaver", "starter"),
"the extract carries prorate_to 2028-02-29", leaver)
@CAT.scenario("SC-HR-16", "proration inputs and rounding once", stories=["HR-19"], tags=["money"],
personas=["hr_admin", "finance"])
def rounding(s):
s.given("an hourly rate of 13.3333 GBP")
s.clock.set("2026-10-05T10:00 Europe/London")
setup(s)
t = team(s, "Q")
le = s.key("LE")
h = hire(s, "Quin", "2026-10-01", t["studio"], wt="employee_pt", rate_amount="13.3333", rate_basis="hour",
legal_entity_ref=le)
s.then("the rate goes out to four places, exactly, and no amount is made without a quantity (the time brick's)")
lines = s.as_("finance").post("/hr/payroll/extracts", {"legal_entity_ref": le, "period": "2026-10"}).ok()["lines"]
mine = [x for x in lines if x["person_id"] == h["person_id"]]
s.check(mine and mine[0]["rate_amount"] == "13.3333" and mine[0]["payable_amount"] is None, "exact rate", mine)
s.note("13.3333 × 450 min = 100.00 (rounded once) is asserted by tests/test_rules.py::test_sc_hr_16_rounding_once: "
"v1 has no door for quantities (they come from the future time brick)")
@CAT.scenario("SC-HR-17", "retention purge", stories=["HR-40"], tags=["clock"],
personas=["hr_admin", "finance", "clock"])
def retention(s):
s.given('a retention rule "bank details, purge 12 months after the engagement ends"; a leaver ended 13 months ago')
s.clock.set("2025-08-04T10:00 Europe/London")
setup(s)
t = team(s, "R")
old = hire(s, "Rae", "2025-08-01", t["studio"])
recent = hire(s, "Rob", "2025-08-01", t["studio"])
for p in (old, recent):
s.as_("finance").post(f"/hr/people/{p['person_id']}/sensitive", {
"kind": "bank", "value": {"bank_sort_code": "10-20-30"}, "valid_from": "2025-08-01"}).ok()
admin(s).post(f"/hr/people/{old['person_id']}/leave", {"last_day": "2025-09-15", "reason": "resignation"}).ok()
s.clock.set("2025-11-10T10:00 Europe/London")
admin(s).post(f"/hr/people/{recent['person_id']}/leave", {"last_day": "2025-11-25", "reason": "resignation"}).ok()
s.when("the daily sweep runs, twice")
s.clock.set("2026-10-20T02:30 Europe/London")
s.tick()
s.tick()
s.then("the bank slices are purged; one retention-log row per category; the second run wrote nothing")
gone = s.as_("finance").get(f"/hr/people/{old['person_id']}/sensitive").ok()
s.check("bank" not in gone["current"] and "bank" not in gone["history"], "purged, history too", gone.body)
logrows = admin(s).get(f"/hr/people/{old['person_id']}/retention-log").ok()["entries"]
bank = [e for e in logrows if e["category"] == "sensitive_bank"]
s.check(len(bank) == 1 and bank[0]["action"] == "purged", "one row", logrows)
s.then("a leaver who ended 11 months ago is untouched")
kept = s.as_("finance").get(f"/hr/people/{recent['person_id']}/sensitive").ok()
s.check("bank" in kept["current"], "untouched", kept.body)
@CAT.scenario("SC-HR-18", "subject-access export", stories=["HR-21"], personas=["hr_admin", "finance"])
def subject_access(s):
s.given("a former freelancer with two engagements, a correction, sensitive data and documents")
s.clock.set("2026-01-10T10:00 Europe/London")
setup(s)
t = team(s, "S")
h = hire(s, "Sol", "2026-01-05", t["studio"], wt="freelancer_sole", rate_amount="400.0000", rate_basis="day")
pid = h["person_id"]
a = admin(s)
a.post(f"/hr/people/{pid}/corrections", {"record": "detail", "valid_from": "2026-01-05",
"values": {"preferred_name": "Solly"}}).ok()
s.as_("finance").post(f"/hr/people/{pid}/sensitive", {"kind": "bank", "value": {"bank_account_number": "12345678"}}).ok()
a.post(f"/hr/people/{pid}/documents", {"document_type": "nda", "filename": "nda.pdf",
"content_base64": base64.b64encode(b"nda").decode()}).ok()
a.post(f"/hr/people/{pid}/leave", {"last_day": "2026-03-31", "reason": "end_of_project"}).ok()
s.clock.set("2026-05-20T10:00 Europe/London")
a.post(f"/hr/people/{pid}/rehire", {"start_date": "2026-06-01", "worker_type": "freelancer_sole",
"job_code": "designer", "org_unit_id": t["studio"], "rate_amount": "420.0000",
"rate_basis": "day"}).ok()
s.clock.set("2026-08-20T10:00 Europe/London")
a.post(f"/hr/people/{pid}/leave", {"last_day": "2026-08-31", "reason": "end_of_project"}).ok()
s.when("HR exports their data")
s.clock.set("2026-10-02T10:00 Europe/London")
req = a.post("/hr/privacy/requests", {"person_id": pid, "kind": "access", "received_on": "2026-10-01"}).ok()
s.then("the request shows its due date one month from receipt")
req.field("due_on", "2026-11-01")
out = a.post(f"/hr/privacy/requests/{req['request_id']}/export", {}).ok()
s.check(out["done_by"] and out["export_object_key"], "the export is logged with who and when", out.body)
arc = a.get(f"/hr/privacy/requests/{req['request_id']}/archive").ok()
import json as _json
data = _json.loads(base64.b64decode(arc["content_base64"]))
s.then("the archive holds every slice incl. superseded, every event, sensitive data, documents, runs, partner")
s.check(any(not x["active"] for x in data["detail"]) and len(data["engagements"]) == 2
and len(data["events"]) >= 4 and "bank" in data["sensitive"] and data["documents"]
and data["runs"] and data["business_partner"] and data["business_partner"]["kind"] == "person",
"everything", {k: (len(v) if isinstance(v, (list, dict)) else v) for k, v in data.items()})
@CAT.scenario("SC-HR-19", "erasure with a legal hold", stories=["HR-22"], personas=["hr_admin", "finance"])
def erasure(s):
s.given("an erasure request, and a retention rule holding payroll records on a legal-obligation basis")
s.clock.set("2024-12-10T10:00 Europe/London")
setup(s)
t = team(s, "U")
h = hire(s, "Uma", "2024-12-02", t["studio"])
pid = h["person_id"]
s.as_("finance").post(f"/hr/people/{pid}/sensitive", {"kind": "bank", "value": {"bank_sort_code": "11-22-33"}}).ok()
admin(s).post(f"/hr/people/{pid}/leave", {"last_day": "2025-01-31", "reason": "resignation"}).ok()
s.clock.set("2026-10-20T10:00 Europe/London")
req = admin(s).post("/hr/privacy/requests", {"person_id": pid, "kind": "erasure", "received_on": "2026-10-20"}).ok()
s.when("HR applies it")
out = admin(s).post(f"/hr/privacy/requests/{req['request_id']}/erase", {}).ok()
s.then("payroll is kept with its basis; the rest erased or anonymised; the partner anonymised through the seam")
o = out["outcome"]
s.check(o["payroll"]["action"] == "kept" and "legal obligation" in o["payroll"]["basis"]
and o["sensitive_bank"]["action"] == "purge" and o["person_detail"]["action"] == "anonymise", "the plan", o)
partner(s, h["bp_id"]).ok().field("legal_name", "Erased person").field("work_email", "")
s.check("bank" not in s.as_("finance").get(f"/hr/people/{pid}/sensitive").ok()["current"], "bank purged")
again = admin(s).post(f"/hr/privacy/requests/{req['request_id']}/erase", {}).ok()
again.same_as(out)
@CAT.scenario("SC-HR-21", "double submits everywhere", stories=["HR-01", "HR-05", "HR-07", "HR-30"],
tags=["concurrency", "idempotency"], personas=["hr_admin", "manager", "it", "clock"])
def double_submits(s):
s.clock.set("2026-10-19T10:00 Europe/London")
setup(s)
t = team(s, "V")
lead(s, t, "studio", 211)
b = body(s, "Vic", "2026-11-02", t["studio"])
s.when("the hire form arrives twice at once — one by the API, one by an agent")
rs = s.concurrently(2, lambda i: admin(s).post("/hr/hires", b) if i else admin(s).tool("record_hire", b))
s.then("one person, one engagement, one run")
rs.expect({200: 2})
s.check(rs[0]["person_id"] == rs[1]["person_id"] and rs[0]["run_id"] == rs[1]["run_id"], "the same", rs)
h = rs[0]
p = admin(s).get(f"/hr/people/{h['person_id']}").ok()
s.check(len(p["engagements"]) == 1 and len([r for r in p["runs"] if r["run_id"].startswith("onboarding")]) == 1,
"one of each", p.body)
rtw(s, h["person_id"], "2026-10-20")
s.when("two ticks run at once, twice over")
s.tick()
s.clock.set("2026-10-26T09:05 Europe/London")
rs = s.concurrently(2, lambda i: s.as_("clock").post("/hr/run-due-steps", {}))
rs.expect({200: 2})
s.then("the draft is made once and approved once")
m = s.as_("manager", 211)
m.post(f"/hr/runs/{h['run_id']}/steps/8/approve-draft", {"text": "Welcome!"}).ok()
m.post(f"/hr/runs/{h['run_id']}/steps/8/approve-draft", {"text": "Welcome!"}).ok().field("draft_text", "Welcome!")
s.clock.set("2026-10-28T09:05 Europe/London")
rs = s.concurrently(2, lambda i: s.as_("clock").post("/hr/run-due-steps", {}))
rs.expect({200: 2})
s.then("one capability call per step: each action done once in the action log")
log = s.as_("it").get("/hr/action-log", run_id=h["run_id"]).ok()["entries"]
done = [e for e in log if e["what"] == "action done"]
s.check(done and len(done) == len({e["step_no"] for e in done}), "once per step",
[(k, x["state"], x["last_status"], x["message"]) for k, x in steps(run(s, h["run_id"])).items()])
@CAT.scenario("SC-HR-22", "intake link lifecycle", stories=["HR-02", "HR-37", "HR-53"],
tags=["public", "clock", "failure"], personas=["hr_admin", "manager", "finance", "clock"])
def intake_lifecycle(s):
no_providers(s)
s.clock.set("2026-10-19T10:00 Europe/London")
setup(s)
t = team(s, "W")
lead(s, t, "studio", 221)
h = hire(s, "Wen", "2026-11-02", t["studio"])
s.tick()
s.given("an intake link issued with a 7-day validity")
first = admin(s).post(f"/hr/intakes/{h['intake_id']}/link", {}).ok()
link = s.link(first["url"], "the starter")
s.when("the starter opens it (seen), and submits half")
page = link.open().status(200)
s.check("Your details" in page.text and "Home address" in page.text, "the form", page.text[:300])
link.submit({"home_address": "1 Lane, London", "ni_number": "QQ123456C"}).status(200)
s.then("the manager's view shows field names complete or missing, never a value")
mv = s.as_("manager", 221).get(f"/hr/people/{h['person_id']}").ok()
s.check(mv["intake"]["done"] == 2 and "home_address" in mv["intake"]["complete"]
and "1 Lane" not in str(mv.body) and "QQ123456C" not in str(mv.body), "names only", mv["intake"])
s.check("1 Lane" not in link.open().text, "the form never shows a stored value back")
s.when("2 working days pass: the starter is chased (no mail provider: a reminder task for HR)")
s.clock.advance("+2 working days")
s.tick()
s.tick()
r = run(s, h["run_id"])
chases = [l for l in r["log"] if l["what"] == "chase" and l["from"] == "1"]
s.check(len(chases) == 1 and "1:chase:1" in steps(r), "once per slot", chases)
s.when("HR reissues; the old link is used; the new link completes it with the right-to-work upload")
second = admin(s).post(f"/hr/intakes/{h['intake_id']}/link", {}).ok()
link.open().status(404)
new = s.link(second["url"], "the starter")
form = {"bank_account_name": "W Wen", "bank_sort_code": "01-02-03", "bank_account_number": "12345678",
"emergency_contact": "Mum 07000"}
new.submit(form, files={"right_to_work": ("passport.pdf", b"%PDF passport", "application/pdf")}).status(200)
s.then("seen and submitted are recorded apart; the upload lands in the Sensitive aggregate; the step is done")
sens = s.as_("finance").get(f"/hr/people/{h['person_id']}/sensitive").ok()
s.check(sens["current"]["right_to_work_evidence"]["has_file"] and sens["current"]["bank"]["value"]["bank_sort_code"]
== "01-02-03", "sensitive", list(sens["current"]))
s.check(steps(run(s, h["run_id"]))["1"]["state"] == "done", "the intake step completed by the starter")
new.open().status(404)
@CAT.scenario("SC-HR-23", "the clock through a whole onboarding",
stories=["HR-01", "HR-02", "HR-30", "HR-31", "HR-32", "HR-33", "HR-51"],
tags=["clock", "failure"], personas=["hr_admin", "it", "manager", "finance", "assignee", "clock"])
def whole_onboarding(s):
no_providers(s)
s.given("a freelance motion designer hired on Monday 2026-10-19 to start Monday 2026-11-02, no providers")
s.clock.set("2026-10-19T10:00 Europe/London")
setup(s)
t = team(s, "X")
lead(s, t, "studio", 231)
h = hire(s, "Xia", "2026-11-02", t["studio"], wt="freelancer_sole", job="motion_designer",
rate_amount="450.0000", rate_basis="day")
rid = h["run_id"]
def due_now():
r = run(s, rid)
st = steps(r)
return {k for k, x in st.items() if x["state"] == "waiting" and not k.startswith(("1:chase", "R"))
and ":escalate" not in k}
s.when("the run is created and the tick runs")
s.tick()
s.check(due_now() == {"1", "3"}, "intake and contract only", due_now())
s.when("the contract is signed (attached by hand: no contracts brick)")
attach_contract(s, rid)
s.check(due_now() == {"1"}, "nothing else is due yet", due_now())
s.when("start −5 working days: the draft and the purchase order")
s.clock.set("2026-10-26T09:01 Europe/London")
s.tick()
s.check({"8", "15"} <= due_now(), "the draft and the PO", due_now())
s.when("start −3 working days 09:00: account and access")
s.clock.set("2026-10-28T09:01 Europe/London")
s.tick()
now = due_now()
s.check({"5"} <= now and not any(k.startswith("6:") for k in now), "the account first; groups after it", now)
s.as_("it").post(f"/hr/runs/{rid}/steps/5/complete", {"output": {"primary_email": f"xia@{DOMAIN}"}}).ok()
s.tick()
now = due_now()
s.check(any(k.startswith("6:") for k in now) and any(k.startswith("7:") for k in now), "groups and spaces", now)
s.then("each manual task carries its exact input")
tasks = s.as_("it").get("/hr/my-tasks").ok()
g = [x for x in tasks["overdue"] + tasks["today"] if x["run_id"] == rid and x["capability"] == "groups.add_member"]
s.check(g and all(x["input"]["member_email"] == f"xia@{DOMAIN}" for x in g), "the address made", g)
for x in steps(run(s, rid)).values():
if x["state"] == "waiting" and x["capability"] and x["capability"] != "mail.send":
s.as_("it" if x["responsible_group"] == "it" else "hr_admin").post(
f"/hr/runs/{rid}/steps/{x['step_no']}/complete", {}).ok()
s.as_("manager", 231).post(f"/hr/runs/{rid}/steps/8/approve-draft", {"text": "Welcome Xia"}).ok()
s.when("start 09:30, start +1 working day 10:00, start +4 working days")
for when in ("2026-11-02T09:31", "2026-11-03T10:01", "2026-11-06T15:01"):
s.clock.set(f"{when} Europe/London")
s.tick()
for x in steps(run(s, rid)).values():
if x["state"] == "waiting" and not x["step_no"].startswith("1"):
if x["kind"] == "intake":
continue
s.as_("hr_admin").post(f"/hr/runs/{rid}/steps/{x['step_no']}/complete", {}).ok()
s.then("at the end, with the starter's details in, the run is done (derived)")
link = admin(s).post(f"/hr/intakes/{h['intake_id']}/link", {}).ok()
s.link(link["url"]).submit({"home_address": "2 Road", "ni_number": "QQ1", "bank_account_name": "X",
"bank_sort_code": "1", "bank_account_number": "2", "emergency_contact": "Y"}).status(200)
for x in steps(run(s, rid)).values():
if x["state"] == "waiting":
admin(s).post(f"/hr/runs/{rid}/steps/{x['step_no']}/complete", {}).ok()
run(s, rid).field("state", "done")
@CAT.scenario("SC-HR-24", "moving a date with steps done", stories=["HR-08"], tags=["clock"],
personas=["hr_admin", "manager", "clock"])
def move_date(s):
s.clock.set("2026-10-19T10:00 Europe/London")
setup(s)
t = team(s, "Y")
lead(s, t, "studio", 241)
h = hire(s, "Yan", "2026-11-02", t["studio"])
rid = h["run_id"]
s.tick()
s.given("steps done: the contract, the payroll set-up, the welcome draft")
s.clock.set("2026-10-26T09:05 Europe/London")
s.tick()
m = s.as_("manager", 241)
m.post(f"/hr/runs/{rid}/steps/8/approve-draft", {"text": "Hi"}).ok()
admin(s).post(f"/hr/runs/{rid}/steps/15/complete", {}).ok()
before = steps(run(s, rid))
s.when("the start moves to 2026-11-09")
m.post(f"/hr/people/{h['person_id']}/move-date", {"which": "start_date", "new_date": "2026-11-09"}).ok()
after = run(s, rid)
st = steps(after)
s.then("not-done steps are re-timed; done steps keep due_at and results; the log says who, from, to")
for no in ("8", "15"):
s.check(st[no]["due_at"] == before[no]["due_at"] and st[no]["state"] == "done", f"step {no} untouched", st[no])
s.check(st["5"]["due_at"] == "2026-11-04T09:00:00Z" and before["5"]["due_at"] == "2026-10-28T09:00:00Z",
"the account step moved a week", (before["5"]["due_at"], st["5"]["due_at"]))
moved = [l for l in after["log"] if l["what"] == "start_date moved"]
s.check(moved and moved[0]["from"] == "2026-11-02" and moved[0]["to"] == "2026-11-09"
and moved[0]["actor"].startswith("manager.241"), "who, from, to", moved)
@CAT.scenario("SC-HR-25", "playbook edited while runs are open", stories=["HR-11"], personas=["hr_admin", "clock"])
def playbook_versions(s):
s.clock.set("2026-10-19T10:00 Europe/London")
setup(s)
t = team(s, "Z")
a = admin(s)
pbid = f"pb-{t['k']}"
base = {"playbook_id": pbid, "name": "Team onboarding", "event_types": ["hire"], "worker_types": [],
"org_unit_id": t["studio"], "steps": [
{"step_no": 1, "kind": "task", "title": "Desk", "responsible_group": "it"},
{"step_no": 2, "kind": "task", "title": "Badge", "responsible_group": "hr"}]}
v1 = a.post("/hr/playbooks", base).ok()["version"]
s.given(f"two open runs on version {v1}")
r1, r2 = (hire(s, n, "2026-11-02", t["studio"])["run_id"] for n in ("Zoe", "Zak"))
s.when("a new version adds a step and removes another")
v2 = a.post("/hr/playbooks", {**base, "steps": [base["steps"][0], {
"step_no": 3, "kind": "task", "title": "Buddy coffee", "responsible_group": "buddy",
"timing": {"anchor": "start_date", "offset": 1, "unit": "working_days", "time": "10:00"}}]}).ok()["version"]
s.then("the open runs are unchanged — after ticks too; a run created afterwards uses the new version")
s.tick()
s.tick()
for rid in (r1, r2):
r = run(s, rid)
s.check(r["playbook"] == f"{pbid} v{v1}" and set(steps(r)) == {"1", "2"}, "frozen", r["playbook"])
r3 = run(s, hire(s, "Zia", "2026-11-02", t["studio"])["run_id"])
s.check(r3["playbook"] == f"{pbid} v{v2}" and set(steps(r3)) == {"1", "3"}, "the new version", r3["playbook"])
s.then("saving a version whose delete step has no approval is refused in words")
a.post("/hr/playbooks", {**base, "steps": [{"step_no": 1, "kind": "action", "title": "Delete", "responsible_group": "it",
"capability": "identity.delete_user",
"input_template": {"primary_email": "{account_email}"}}]}).refused(
409, "must depend on an approval step")
@CAT.scenario("SC-HR-26", "a run cancelled after actions were done", stories=["HR-23"], tags=["capability"],
personas=["hr_admin", "it", "clock"])
def cancel(s):
t, h = _due_account(s, "AA", given="Abe")
s.given("an onboarding run where provision_user and the group adds are done")
s.clock.set("2026-10-28T09:01 Europe/London")
s.tick()
s.tick()
st = steps(run(s, h["run_id"]))
s.check(st["5"]["state"] == "done" and any(k.startswith("6:") and x["state"] == "done" for k, x in st.items()),
"account and groups done", st)
s.when("HR cancels it")
out = admin(s).post(f"/hr/runs/{h['run_id']}/cancel", {"reason": "starter withdrew"}).ok()
s.then("pending steps are cancelled; reversal tasks (suspend, remove from the group) exist; nothing reversed")
r = run(s, h["run_id"])
st = steps(r)
rev = [x for k, x in st.items() if k.startswith("R")]
s.check(r["state"] == "cancelled" and st["12"]["state"] == "cancelled", "cancelled", r["state"])
s.check(any("identity.provision_user" in x["title"] for x in rev)
and any("groups.add_member" in x["title"] for x in rev) and all(x["state"] == "waiting" for x in rev),
"reversal TASKS, waiting for a person", rev)
s.tick()
log = s.as_("it").get("/hr/action-log", run_id=h["run_id"]).ok()["entries"]
s.check(not any(e["capability"] in ("identity.suspend_user", "groups.remove_member") and e["what"] == "action done"
for e in log), "nothing reversed automatically", log)
out.field("state", "cancelled")
@CAT.scenario("SC-HR-27", "the business partner receives nothing private", stories=["HR-50"], tags=["clock"],
personas=["hr_admin", "finance", "it", "clock"])
def nothing_private(s):
s.given("a person with personal email and phone, address, NI number, bank and right-to-work evidence")
s.clock.set("2026-10-05T10:00 Europe/London")
setup(s)
t = team(s, "AB")
h = hire(s, "Bea", "2026-10-01", t["studio"], personal_phone="+44 7700 900123")
pid = h["person_id"]
fin = s.as_("finance")
for kind, value in (("home_address", "9 Secret Road"), ("ni_number", "QQ999999Q"),
("bank", {"bank_account_number": "99999999"})):
fin.post(f"/hr/people/{pid}/sensitive", {"kind": kind, "value": value}).ok()
s.when("every sync path runs: hire, detail change, work email written back, conversion, anonymisation")
admin(s).post(f"/hr/people/{pid}/corrections", {"record": "detail", "valid_from": "2026-10-01",
"values": {"preferred_name": "Bee", "work_email": f"bea@{DOMAIN}"}}).ok()
s.tick()
snapshot = partner(s, h["bp_id"]).ok()
private = (s.email("x").split("@")[1], "+44 7700", "9 Secret", "QQ999999Q", "99999999", "home.example")
s.then("the partner holds only legal and preferred name, work email and phone, roles and the works_for edge")
s.check(set(snapshot.body) <= {"bp_id", "bp_key", "legal_name", "country", "tax_id", "domain", "status", "roles",
"can_transact", "kind", "preferred_name", "work_email", "work_phone", "anonymised",
"version"}, # bp 0.3.0: the partner's optimistic version (bp's own)
"no other field", snapshot.body)
s.check(snapshot["work_email"] == f"bea@{DOMAIN}" and snapshot["preferred_name"] == "Bee", "the work contact")
hist = fin.get(f"/business-partner/partners/{h['bp_id']}/history").ok()
for blob in (str(snapshot.body), str(hist.body)):
s.check(not any(x in blob for x in private[1:]), "nothing private reached the partner", blob)
@CAT.scenario("SC-HR-28", "a company freelancer", stories=["HR-01", "HR-50", "HR-27"], tags=["clock", "capability"],
personas=["hr_admin", "finance", "clock"])
def company_freelancer(s):
s.given("a freelancer via Studio Nine Ltd, at a client set as medium")
s.clock.set("2026-10-19T10:00 Europe/London")
setup(s)
t = team(s, "AC")
le = s.key("LE")
admin(s).post("/hr/configuration/settings", {"rows": [{"key": f"client_size:{le}", "value": "medium"}]}).ok()
s.when("the hire is recorded")
h = hire(s, "Cy", "2026-11-02", t["studio"], wt="freelancer_company", rate_amount="500.0000", rate_basis="day",
legal_entity_ref=le, company={"legal_name": f"Studio Nine {t['k']} Ltd", "company_number": t["k"][-8:]})
s.then("the company is an organisation partner with the supplier role; supplier_bp_id is on the engagement")
co = s.as_("finance").get(f"/business-partner/partners/{h['supplier_bp_id']}").ok()
co.field("kind", "organisation").contains("roles", "supplier")
s.then("works_for links them; the person's partner has worker (not supplier) once they start")
fam = s.as_("finance").get(f"/business-partner/partners/{h['supplier_bp_id']}/family", rel="works_for").ok()
s.check(any(str(m["bp_id"]) == str(h["bp_id"]) for m in fam["below"]), "works_for", fam.body)
s.then("the contract step is blocked until the status determination is recorded")
s.tick()
st = steps(run(s, h["run_id"]))["3"]
s.check(st["state"] == "pending" and "status determination" in st["message"], "blocked in words", st)
admin(s).post(f"/hr/people/{h['person_id']}/compliance", {"kind": "status_determination", "checked_on": "2026-10-19",
"checked_by": "client", "outcome": "outside",
"reasons": "own tools, substitution"}).ok()
s.clock.advance("+2m")
s.tick()
st = steps(run(s, h["run_id"]))["3"]
s.check(st["state"] != "pending" or st["contract_id"], "the contract went out", st)
s.clock.set("2026-11-02T08:00 Europe/London")
s.tick()
partner(s, h["bp_id"]).ok().field("roles", ["worker"])
# ── the contract step against what contracts can answer (HR-51) ────────────────────────────
def _contract_starter(s, tag, given, **kw):
"""A starter whose contract step (3) is due now, with the providers present (rtw recorded: nothing gates it)."""
s.clock.set("2026-10-19T10:00 Europe/London")
setup(s)
t = team(s, tag)
h = hire(s, given, "2026-11-02", t["studio"], **kw)
rtw(s, h["person_id"], "2026-10-20")
return t, h
def _step3(s, h):
return steps(run(s, h["run_id"]))["3"]
@CAT.scenario("SC-HR-29", "a contract that needs an approval", stories=["HR-51"], tags=["clock", "capability"],
personas=["hr_admin", "clock"])
def contract_approval(s):
s.given("a starter whose contract needs legal's approval in contracts (a rewritten clause)")
t, h = _contract_starter(s, "AP", "Approval")
s.when("the tick drafts it")
s.tick()
st = _step3(s, h)
s.then("hr waits for the approval: drafted, not sent, asked again on the next tick")
s.check(st["state"] == "pending" and st["phase"] == "drafted" and st["contract_id"], "drafted, waiting", st)
s.when("the approval is still open at the next tick")
s.clock.advance("+2m")
s.tick()
st = _step3(s, h)
s.check(st["phase"] == "drafted" and "approval" in (st["message"] or ""), "still waiting for the approval", st)
s.when("legal approves it in contracts, and the tick asks again")
s.clock.advance("+2m")
s.tick()
st = _step3(s, h)
s.then("hr sends it (the same key as a first send) and polls it")
s.check(st["state"] == "waiting" and st["phase"] == "sent", "sent for signature", st)
s.when("it is signed")
s.clock.advance("+1d")
s.tick()
st = _step3(s, h)
s.then("the signed copy is filed on the person and the step is done")
s.check(st["state"] == "done" and st["phase"] == "signed", "signed and filed", st)
docs = admin(s).get(f"/hr/people/{h['person_id']}").ok()["documents"]
s.check(any(d["doc_id"] == f"contract:{st['contract_id']}" for d in docs), "on the person", docs)
if s.mode == "inprocess":
import hr_test_provider.brick as prov
sends = [c for c in prov.CALLS if c["capability"] == "contracts.send" and c["key"].startswith(h["run_id"])]
s.check([c["key"] for c in sends] == [f"{h['run_id']}|3|send"], "one send, after the approval", sends)
@CAT.scenario("SC-HR-30", "a contract that ends unsigned, or is signed on paper", stories=["HR-51"],
tags=["clock", "capability"], personas=["hr_admin", "clock"])
def contract_ends(s):
s.clock.set("2026-10-19T10:00 Europe/London")
setup(s)
t = team(s, "EN")
s.given("four starters whose contracts end DECLINED, RECALLED, EXPIRED_UNSIGNED and SIGNED_ELSEWHERE")
people = {}
for given in ("Declines", "Recalled", "Lapses", "Paper"):
h = hire(s, given, "2026-11-02", t["studio"])
rtw(s, h["person_id"], "2026-10-20")
people[given] = h
s.when("the tick sends them and asks where they stand")
s.tick()
s.clock.advance("+1d")
s.tick()
s.then("each that ended unsigned is a manual task naming why; none is left polling")
for given, state in (("Declines", "DECLINED"), ("Recalled", "RECALLED"), ("Lapses", "EXPIRED_UNSIGNED")):
st = _step3(s, people[given])
s.check(st["state"] == "waiting" and st["manual"] and state in (st["manual_reason"] or "")
and "attach the signed copy" in (st["plan"] or ""), f"{state}: a manual task with the reason", st)
s.then("signed on paper (SIGNED_ELSEWHERE) is signed: the copy is filed and the step is done")
st = _step3(s, people["Paper"])
s.check(st["state"] == "done" and st["phase"] == "signed", "signed elsewhere = signed", st)
@CAT.scenario("SC-HR-31", "a contract missing a value", stories=["HR-51", "HR-28"], tags=["clock", "capability"],
personas=["hr_admin", "clock"])
def contract_missing(s):
le = s.key("LE")
s.given("a contract type that needs the client size regime, which hr has not been told for this legal entity")
t, h = _contract_starter(s, "MI", "Regime", legal_entity_ref=le)
s.when("the tick asks contracts to draft it")
s.tick()
st = _step3(s, h)
s.then("contracts stores nothing; the step waits and HR gets a task naming the field")
s.check(st["state"] == "pending" and not st["contract_id"] and "client_size_regime" in (st["message"] or ""),
"blocked in words", st)
fill = steps(run(s, h["run_id"])).get("3:fill")
s.check(fill and fill["state"] == "waiting" and "client_size_regime" in fill["title"], "a fill-in task", fill)
s.when("another tick passes with nothing filled in")
s.clock.advance("+2m")
s.tick()
s.check(not _step3(s, h)["contract_id"], "not drafted", _step3(s, h))
s.when("HR sets the client size for the legal entity and marks the task done")
admin(s).post("/hr/configuration/settings", {"rows": [{"key": f"client_size:{le}", "value": "small"}]}).ok()
admin(s).post(f"/hr/runs/{h['run_id']}/steps/3:fill/complete", {}).ok()
s.clock.advance("+2m")
s.tick()
st = _step3(s, h)
s.then("the draft is asked again with the same key, and the contract goes out")
s.check(st["contract_id"] and st["phase"] == "sent", "drafted and sent", st)
if s.mode == "inprocess":
import hr_test_provider.brick as prov
keys = {c["key"] for c in prov.CALLS if c["capability"] == "contracts.draft"
and c["key"].startswith(f"hr:{h['run_id']}")}
s.check(keys == {f"hr:{h['run_id']}#3|employment_contract"} or len(keys) == 1, "one key throughout", keys)
# ── generated probes: SC-HR-20 (every role probes every forbidden operation), idempotency, mass assignment ──
@CAT.scenario("SC-HR-32", "employee documents by access class", stories=["HR-20", "HR-13"],
personas=["hr_admin", "manager", "finance"])
def employee_documents(s):
s.given("a starter in a manager's line, and HR filing an NDA (personnel) and a right-to-work copy (sensitive)")
s.clock.set("2026-10-19T10:00 Europe/London")
setup(s)
t = team(s, "DOC")
lead(s, t, "studio", 251)
pid = hire(s, "Dora", "2026-11-02", t["studio"], who="manager", n=251)["person_id"]
a = admin(s)
nda = a.post(f"/hr/people/{pid}/documents", {"document_type": "nda", "filename": "nda.pdf",
"content_base64": base64.b64encode(b"%PDF nda").decode()}).ok()
rtw_doc = a.post(f"/hr/people/{pid}/documents", {"document_type": "right_to_work", "filename": "passport.pdf",
"content_base64": base64.b64encode(b"%PDF passport").decode()}).ok()
s.then("each document's type decides its class")
s.check(nda["access_class"] == "personnel" and rtw_doc["access_class"] == "sensitive", "the classes",
{"nda": nda.body, "rtw": rtw_doc.body})
s.when("the manager lists and opens her documents")
m = s.as_("manager", 251)
listed = m.get(f"/hr/people/{pid}/documents").ok()
s.then("the manager sees the NDA and nothing of the sensitive copy, which is not found when asked for by id")
s.check([d["doc_id"] for d in listed["documents"]] == [nda["doc_id"]], "only the personnel document",
listed.body)
got = m.get(f"/hr/people/{pid}/documents/{nda['doc_id']}").ok()
s.check(got["content_base64"] == base64.b64encode(b"%PDF nda").decode() and "url" not in str(got.body).lower(),
"streamed, never a link", got.body)
m.get(f"/hr/people/{pid}/documents/{rtw_doc['doc_id']}").refused(404)
s.then("HR sees both; HR and finance (hr:sensitive:read) open the sensitive copy")
both = a.get(f"/hr/people/{pid}/documents").ok()
s.check({d["doc_id"] for d in both["documents"]} == {nda["doc_id"], rtw_doc["doc_id"]}, "HR sees both", both.body)
for who in ("hr_admin", "finance"):
s.as_(who).get(f"/hr/people/{pid}/documents/{rtw_doc['doc_id']}").ok()
def _owned(s):
"""Records made by the probe's owner — a manager with a line — for the rows probe. Operations sets the team up
(the probe declares only the two managers, so the admin acts through a plain Actor)."""
from erp_platform.scenarios.core import Actor
s.as_ = (lambda real: lambda role, n=1: Actor(s, role if n == 1 else f"{role}#{n}"))(s.as_)
setup(s)
t = team(s, "PR")
lead(s, t, "studio", 1)
h = hire(s, "Rowan", "2027-01-04", t["studio"], who="manager", n=1)
return {"person_id": h["person_id"]}
SENSITIVE_404 = {(r, uc): 404 for r in ("manager", "it", "assignee", "employee", "clock")
for uc in ("show_sensitive", "record_sensitive", "download_document")}
PROV_DATETIME = ("the test provider's capability: the probe's placeholder 'x' fails the contract's ISO date-time "
"before the permission check (a placeholder gap in erp_platform.scenarios.probes, not hr's)")
CAT.permission_matrix(rows=Rows(make=_owned, owner="manager", other="manager#2"), expect=SENSITIVE_404, exceptions={
"set_hr_demo_clock": "the harness's own lever (mounted only with HR_DEMO_CLOCK=enabled)",
"mail_set_autoreply": PROV_DATETIME, "calendar_create_event": PROV_DATETIME})
def _person(s, wt="employee_ft", start="2026-11-02", who="hr_admin"):
s.clock.set("2026-10-19T10:00 Europe/London")
setup(s)
t = team(s, "IP")
return t, hire(s, "Ivy", start, t["studio"], wt=wt)
def _get_person(pid):
return lambda s, r: s.as_("hr_admin").get(f"/hr/people/{pid}")
def _r_hire(s):
setup(s)
t = team(s, "RH")
b = body(s, "Ray", "2027-01-04", t["studio"])
return ("hr_admin", b, lambda s, r: s.as_("hr_admin").get(f"/hr/people/{r['person_id']}"))
def _r_leave(s):
t, h = _person(s)
return ("hr_admin", {"person_id": h["person_id"], "last_day": "2027-03-31", "reason": "resignation"},
_get_person(h["person_id"]))
def _r_rehire(s):
t, h = _person(s, start="2026-10-01")
s.as_("hr_admin").post(f"/hr/people/{h['person_id']}/leave", {"last_day": "2026-10-31", "reason": "resignation"}).ok()
return ("hr_admin", {"person_id": h["person_id"], "start_date": "2026-12-01", "worker_type": "employee_ft",
"job_code": "designer", "org_unit_id": t["studio"]}, _get_person(h["person_id"]))
def _r_conversion(s):
t, h = _person(s, wt="freelancer_sole", start="2026-10-01")
return ("hr_admin", {"person_id": h["person_id"], "effective_date": "2026-12-01", "worker_type": "employee_ft"},
_get_person(h["person_id"]))
def _run_step(s, kind):
t, h = _person(s)
s.tick()
return h, steps(run(s, h["run_id"]))
def _r_complete(s):
h, st = _run_step(s, "task")
return ("hr_admin", {"run_id": h["run_id"], "step_no": "15"}, lambda s, r: run(s, h["run_id"]))
def _r_approve(s):
s.clock.set("2026-10-19T10:00 Europe/London")
setup(s)
t = team(s, "AP")
s.clock.set("2026-10-05T10:00 Europe/London")
h = hire(s, "Al", "2026-10-01", t["studio"])
left = s.as_("hr_admin").post(f"/hr/people/{h['person_id']}/leave", {"last_day": "2026-12-31", "reason": "resignation"}).ok()
return ("hr_admin", {"run_id": left["run_id"], "step_no": "2", "granted": True}, lambda s, r: run(s, left["run_id"]))
def _r_draft(s):
h, st = _run_step(s, "draft")
return ("hr_admin", {"run_id": h["run_id"], "step_no": "8", "text": "Welcome"}, lambda s, r: run(s, h["run_id"]))
def _r_party(s):
t, h = _person(s)
other = hire(s, "Bud", "2026-11-02", t["studio"])
return ("hr_admin", {"run_id": h["run_id"], "group_code": "buddy", "person_id": other["person_id"]},
lambda s, r: run(s, h["run_id"]))
def _r_cancel(s):
t, h = _person(s)
return ("hr_admin", {"run_id": h["run_id"], "reason": "withdrew"}, lambda s, r: run(s, h["run_id"]))
def _r_invite(s):
t, h = _person(s)
p = s.as_("hr_admin").get(f"/hr/people/{h['person_id']}").ok()
s.as_("hr_admin").post(f"/hr/people/{h['person_id']}/corrections", {
"record": "detail", "valid_from": p["detail_history"][0]["valid_from"],
"values": {"work_email": f"{s.key('w').lower()}@{DOMAIN}"}}).ok()
return ("hr_admin", {"person_id": h["person_id"]}, _get_person(h["person_id"]))
def _r_file(s):
t, h = _person(s)
return ("hr_admin", {"person_id": h["person_id"], "document_type": "nda", "filename": "nda.pdf",
"content_base64": base64.b64encode(s.key("doc").encode()).decode()},
lambda s, r: s.as_("hr_admin").get(f"/hr/people/{h['person_id']}/documents"))
def _r_compliance(s):
t, h = _person(s)
return ("hr_admin", {"person_id": h["person_id"], "kind": "right_to_work_check", "checked_on": "2026-10-20",
"checked_by": "hr", "check_type": "manual_document_check"}, _get_person(h["person_id"]))
def _r_extract(s):
s.clock.set("2026-10-05T10:00 Europe/London")
setup(s)
t = team(s, "EX")
le = s.key("LE")
hire(s, "Eve", "2026-10-01", t["studio"], legal_entity_ref=le)
return ("finance", {"legal_entity_ref": le, "period": "2026-10"}, None)
def _r_sent(s):
role, b, _ = _r_extract(s)
out = s.as_("finance").post("/hr/payroll/extracts", b).ok()
return ("finance", {"batch_id": out["batch_id"]}, lambda s, r: s.as_("finance").get(f"/hr/payroll/extracts/{out['batch_id']}"))
def _r_privacy(s):
t, h = _person(s)
return ("hr_admin", {"person_id": h["person_id"], "kind": "access", "received_on": "2026-10-19"}, None)
def _r_export(s):
role, b, _ = _r_privacy(s)
q = s.as_("hr_admin").post("/hr/privacy/requests", b).ok()
return ("hr_admin", {"request_id": q["request_id"]}, None)
def _r_erase(s):
s.clock.set("2026-10-05T10:00 Europe/London")
setup(s)
t = team(s, "ER")
h = hire(s, "Eli", "2026-10-01", t["studio"])
s.as_("hr_admin").post(f"/hr/people/{h['person_id']}/leave", {"last_day": "2026-10-02", "reason": "resignation"}).ok()
s.clock.set("2026-10-20T10:00 Europe/London")
q = s.as_("hr_admin").post("/hr/privacy/requests", {"person_id": h["person_id"], "kind": "erasure",
"received_on": "2026-10-20"}).ok()
return ("hr_admin", {"request_id": q["request_id"]}, None)
def _r_change(s):
setup(s)
pid = f"pb-{s.key('c').lower()}"
s.as_("hr_admin").post("/hr/playbooks", {"playbook_id": pid, "name": "x", "event_types": ["leave"],
"reasons": [s.key("r")], "steps": [
{"step_no": 1, "kind": "task", "title": "t", "responsible_group": "hr"}]}).ok()
return ("hr_admin", {"playbook_id": pid, "change": "add_step", "after": ["t"],
"step": {"kind": "task", "title": "Desk booked", "responsible_group": "it"}},
lambda s, r: s.as_("hr_admin").get("/hr/playbooks", playbook_id=pid))
RECIPES = {
"record_hire": _r_hire, "record_leave": _r_leave, "record_rehire": _r_rehire, "record_conversion": _r_conversion,
"complete_step": _r_complete, "approve_step": _r_approve, "approve_draft": _r_draft, "assign_run_party": _r_party,
"cancel_run": _r_cancel, "invite_to_sign_in": _r_invite, "file_document": _r_file,
"record_compliance": _r_compliance, "produce_payroll_extract": _r_extract, "mark_extract_sent": _r_sent,
"record_privacy_request": _r_privacy, "export_subject_access": _r_export, "apply_erasure": _r_erase,
"keep_configuration": lambda s: (setup(s), ("hr_admin", {"catalogue": "job", "rows": [
{"job_code": "copywriter", "title": "Copywriter"}]}, None))[1],
"keep_playbook": lambda s: (setup(s), ("hr_admin", {"playbook_id": f"pb-{s.key('x').lower()}", "name": "x",
"event_types": ["leave"], "reasons": [s.key("r")],
"steps": [{"step_no": 1, "kind": "task", "title": "t",
"responsible_group": "hr"}]}, None))[1],
"change_playbook": _r_change,
# HR-55: a dated row from a day of its own, so a re-run (and the other probe) meets the same row: a replay.
"set_dated_setting": lambda s: (setup(s), ("hr_admin", {"setting_key": "offboarding.revoke_access_at",
"value": "immediate", "scope": "dismissal",
"valid_from": "2027-01-04"}, None))[1],
"install_localisation_pack": lambda s: ("hr_admin", {"pack": "gb", "with_playbooks": True}, None),
"record_transfer": lambda s: (lambda t_h: ("hr_admin", {"person_id": t_h[1]["person_id"], "effective_date": "2026-12-01",
"org_unit_id": t_h[0]["camp"]}, None))(_person(s)),
"record_pay_change": lambda s: (lambda t_h: ("hr_admin", {"person_id": t_h[1]["person_id"],
"effective_date": "2026-12-01", "reason": "review",
"rate_amount": "50000.0000", "rate_basis": "year"},
None))(_person(s)),
"record_contract_extension": lambda s: (lambda t_h: ("hr_admin", {"person_id": t_h[1]["person_id"],
"new_end": "2027-06-30"}, None))(_person(s)),
"correct_slice": lambda s: (lambda t_h: ("hr_admin", {"person_id": t_h[1]["person_id"], "record": "detail",
"valid_from": "2026-10-19",
"values": {"preferred_name": "Iv"}}, None))(_person(s)),
"move_event_date": lambda s: (lambda t_h: ("hr_admin", {"person_id": t_h[1]["person_id"], "which": "start_date",
"new_date": "2026-11-09"}, None))(_person(s)),
"keep_org": lambda s: (setup(s), ("hr_admin", {"org_id": ORG, "kind": "unit", "id": f"U-{s.key('u').lower()}",
"valid_from": "2026-01-01", "values": {"name": "New"}}, None))[1],
"preview_run": lambda s: (lambda t_h: ("hr_admin", {"run_id": t_h[1]["run_id"]}, None))(_person(s)),
"record_sensitive": lambda s: (lambda t_h: ("finance", {"person_id": t_h[1]["person_id"], "kind": "ni_number",
"value": "QQ123"}, None))(_person(s)),
"reissue_intake_link": lambda s: (lambda t_h: ("hr_admin", {"intake_id": t_h[1]["intake_id"]}, None))(_person(s)),
}
BP_OWN = "business-partner's own command: its brick's suite probes it (hr only calls its seam)"
BP_COMMANDS = ("upsert_partner", "set_partner_status", "remove_role", "link_partners", "unlink_partners",
"promote_partner", "sync_person")
CAT.idempotency(recipes=RECIPES, exceptions={
**{n: BP_OWN for n in BP_COMMANDS},
"submit_intake": "reached only through the public door as the link's holder; SC-HR-22 replays it",
"set_hr_demo_clock": "the harness's own lever"},
ignore=("version", "sent_at", "done_at"))
CAT.mass_assignment(recipes=RECIPES, exceptions={
**{n: BP_OWN for n in BP_COMMANDS},
"submit_intake": "reached only through the public door; its fields are the pack's rows, extra ones refused 422",
"run_due_steps": "takes no input",
"retry_step": "takes no body",
"set_hr_demo_clock": "the harness's own lever"})