dodil/erp-brick-hr
tests/test_rules.py
"""The rules with no database: one test per sentence a business person would agree to (modules/hr stories),
named for the story. The domain is plain Python, so these are fast and exact."""
from datetime import date, datetime, timedelta, timezone
from decimal import Decimal
import pytest
from hypothesis import given, settings, strategies as st
from erp_platform import work
from erp_platform.money import Rate
from brick_hr.domain import events as ev
from brick_hr.domain import playbook as pbk
from brick_hr.domain import run as R
from brick_hr.domain import services
from brick_hr.domain.configuration import CATALOGUES, Config, validate_row
from brick_hr.domain.errors import Incomplete, RuleBroken
from brick_hr.domain.intake import status as intake_status
from brick_hr.domain.org import Org
from brick_hr.domain.packs import gb
from brick_hr.domain.payroll import Batch, build, lines_for, payable
from brick_hr.domain.person import HireSpec, Person
from brick_hr.domain.privacy import due_on, erasure_plan, retention_due
from brick_hr.domain.slices import Timeline
from brick_hr.domain.values import OPEN, PersonKey, rate
from brick_hr.infrastructure import models_adapter
NOW = datetime(2026, 10, 17, 9, tzinfo=timezone.utc)
D = date.fromisoformat
CFG = Config(gb.catalogues())
EMP = CFG.worker_type("employee_ft")
SOLE = CFG.worker_type("freelancer_sole")
CO = CFG.worker_type("freelancer_company")
def spec(start="2026-11-02", wt="employee_ft", unit="STUDIO", r=None, supplier=None, **kw):
return HireSpec(wt, D(start), "LE", {"org_id": "acme", "org_unit_id": unit, "job_code": "designer",
"location_code": "LON", "schedule_code": "FT", **kw}, None,
r, None, None, supplier, "hire")
def anna(start="2026-11-02", **kw):
return Person.from_hire(PersonKey.of(None, " Anna@Example.org "),
{"legal_given_name": "Anna", "legal_family_name": "Berg", "personal_email": "anna@example.org"},
spec(start, **kw), EMP, today=D("2026-10-17"), now=NOW, by="hr")
# ── slices ─────────────────────────────────────────────────────────────────────────────────
def test_a_change_delimits_the_slice_before_and_nothing_is_overwritten():
t = Timeline("pay")
t.start(D("2026-03-01"), {"rate_amount": Decimal("42000")}, now=NOW, by="hr")
t.change(D("2026-04-01"), {"rate_amount": Decimal("45000")}, now=NOW, by="hr")
a, b = t.active()
assert (a.valid_to, a.data["rate_amount"], b.valid_from, b.valid_to) == (D("2026-03-31"), 42000, D("2026-04-01"), OPEN)
def test_a_correction_supersedes_and_keeps_what_we_believed_before():
t = Timeline("pay")
t.start(D("2026-03-01"), {"r": 1}, now=NOW, by="hr")
t.change(D("2026-04-01"), {"r": 2}, now=NOW + timedelta(days=1), by="hr")
t.correct(D("2026-04-01"), {"r": 3}, now=NOW + timedelta(days=10), by="hr")
assert t.at(D("2026-04-15")).data["r"] == 3
assert t.believed(D("2026-04-15"), NOW + timedelta(days=5)).data["r"] == 2 # before the correction
old = [s for s in t.slices if not s.active][0]
assert old.superseded_by == 2 and old.data["r"] == 2 # still readable
t.check_tc1()
def test_two_changes_on_one_day_are_refused_correct_instead():
t = Timeline("pay")
t.start(D("2026-03-01"), {"r": 1}, now=NOW, by="hr")
with pytest.raises(RuleBroken, match="correct it instead"):
t.change(D("2026-03-01"), {"r": 2}, now=NOW, by="hr")
def test_tc1_refuses_a_gap_and_an_overlap():
t = Timeline("x")
t.start(D("2026-01-01"), {}, now=NOW, by="hr", valid_to=D("2026-01-31"))
t.start(D("2026-02-02"), {}, now=NOW, by="hr")
with pytest.raises(RuleBroken, match="gap"):
t.check_tc1()
with pytest.raises(RuleBroken, match="already covers"):
t.start(D("2026-01-15"), {}, now=NOW, by="hr")
@settings(max_examples=150, deadline=None)
@given(st.lists(st.tuples(st.sampled_from(["change", "correct", "end", "reopen"]), st.integers(0, 400)), max_size=25))
def test_no_gap_and_no_overlap_after_any_sequence_of_writes(ops):
"""The invariant as a property: random changes, corrections, ends and reopenings never leave a gap or an
overlap in the active slices."""
t = Timeline("x")
start = D("2026-01-01")
t.start(start, {"v": 0}, now=NOW, by="t")
for i, (op, off) in enumerate(ops):
d = start + timedelta(days=off)
try:
if op == "change":
t.change(d, {"v": i}, now=NOW, by="t")
elif op == "correct":
s = t.at(d)
if s:
t.correct(s.valid_from, {"v": -i}, now=NOW, by="t")
elif op == "end" and d >= start:
t.end_at(d, now=NOW)
elif op == "reopen" and t.last():
t.reopen(t.last().valid_to, OPEN)
except RuleBroken:
pass
t.check_tc1(start if t.active() else None)
# ── HR-01: the hire ────────────────────────────────────────────────────────────────────────
def test_hr_01_one_person_from_the_normalised_personal_email():
p = anna()
assert p.person_key == "email:anna@example.org" and p.person_id == PersonKey.of(None, "ANNA@example.org ").person_id
assert 0 < p.person_id < 2 ** 56
assert PersonKey.of("A-17", "x@y.z").value == "staff:a-17"
def test_hr_01_a_hire_writes_engagement_assignment_and_starts_an_onboarding_run():
p = anna(r=rate("42000", "year", "GBP"))
e = p.events[0]
assert (e.kind, e.direction, e.run_id) == ("Hired", "onboarding", f"onboarding:{p.person_id}:1:1")
assert p.assignments[1].first().valid_from == D("2026-11-02") and p.pay[1].first().data["rate_amount"] == Decimal("42000.0000")
assert p.engagements[1].status(D("2026-10-17")) == "future"
def test_hr_01_submitting_twice_is_the_same_hire():
p = anna()
assert p.same_hire(spec()) is p.engagements[1]
def test_hr_01_a_company_freelancer_with_no_company_is_refused():
with pytest.raises(Incomplete, match="needs the company"):
Person.from_hire(PersonKey.of(None, "c@x.y"), {"legal_given_name": "C", "legal_family_name": "D"},
spec(wt="freelancer_company"), CO, today=D("2026-10-17"), now=NOW, by="hr")
def test_an_employee_engagement_never_names_a_supplier():
with pytest.raises(RuleBroken, match="not via a company"):
anna(supplier=123)
def test_hr_01_a_second_open_engagement_is_refused_naming_it():
p = anna()
with pytest.raises(RuleBroken, match="engagement 1 .* is still open"):
p.rehire(spec("2027-01-04"), EMP, now=NOW, by="hr")
# ── HR-06, HR-15, HR-16, HR-17, HR-14 ──────────────────────────────────────────────────────
def test_hr_06_a_leave_delimits_assignment_and_pay_at_last_day_plus_one():
p = anna("2026-01-05", r=rate("42000", "year", "GBP"))
e = p.leave(D("2026-11-30"), "resignation", "leave", now=NOW, by="m")
assert p.engagements[1].actual_end == D("2026-11-30")
assert p.assignments[1].last().valid_to == D("2026-11-30") and p.pay[1].last().valid_to == D("2026-11-30")
assert e.direction == "offboarding" and e.takes_effect_on == D("2026-12-01")
def test_a_leave_with_reason_converted_starts_no_offboarding_run():
assert ev.direction_of("Left", "converted") is None
def test_hr_15_a_rehire_is_engagement_n_plus_1_and_never_offboards():
p = anna("2026-01-05")
p.leave(D("2026-11-30"), "resignation", "leave", now=NOW, by="hr")
with pytest.raises(RuleBroken, match="engagement 1 is still active on 30 Nov; start the new one on 1 Dec"):
p.rehire(spec("2026-11-30"), EMP, now=NOW, by="hr")
e = p.rehire(spec("2026-12-01"), EMP, now=NOW, by="hr")
assert (e.n, e.kind, e.direction) == (2, "Rehired", "onboarding")
assert sorted(p.engagements) == [1, 2] and len(p.events) == 3
def test_hr_16_a_conversion_is_one_event_ending_n_and_starting_n_plus_1():
p = Person.from_hire(PersonKey.of(None, "f@x.y"), {"legal_given_name": "F", "legal_family_name": "G"},
spec("2026-06-01", wt="freelancer_sole", r=rate("450", "day", "GBP")), SOLE,
today=D("2026-06-01"), now=NOW, by="hr")
e = p.convert(spec("2026-12-01", r=rate("52000", "year", "GBP")), EMP, now=NOW, by="hr")
assert p.engagements[1].actual_end == D("2026-11-30") and p.engagements[1].leave_reason == "converted"
assert p.engagements[2].start_date == D("2026-12-01") and e.kind == "Converted" and e.direction == "conversion"
assert [x.kind for x in p.events] == ["Hired", "Converted"]
def test_hr_17_a_transfer_delimits_the_old_assignment():
p = anna("2026-01-05")
p.transfer(D("2026-11-16"), {"org_unit_id": "CAMP"}, "transfer", None, now=NOW, by="m")
a, b = p.assignments[1].active()
assert (a.valid_to, b.valid_from, b.data["org_unit_id"]) == (D("2026-11-15"), D("2026-11-16"), "CAMP")
def test_hr_14_a_pay_change_needs_a_reason_and_is_its_own_slice():
p = anna("2026-03-01", r=rate("42000", "year", "GBP"))
with pytest.raises(Incomplete, match="needs a reason"):
p.change_pay(D("2026-04-01"), "", "pay_change", now=NOW, by="hr", rate=rate("45000", "year", "GBP"))
p.change_pay(D("2026-04-01"), "annual review", "pay_change", now=NOW, by="hr", rate=rate("45000", "year", "GBP"))
p.change_pay(D("2026-05-01"), "car", "pay_change", now=NOW, by="hr", rate=rate("300", "month", "GBP"),
component="car_allowance")
assert p.pay[1].first().valid_to == D("2026-03-31") and p.components[(1, "car_allowance")].first().valid_to == OPEN
def test_hr_14_a_rate_carries_four_places_and_is_never_a_float():
assert rate("13.3333", "hour", "GBP").amount == Decimal("13.3333")
with pytest.raises(Incomplete):
rate("13.33333", "hour", "GBP")
with pytest.raises(Incomplete):
rate(13.33, "hour", "GBP")
def test_hr_08_moving_the_start_moves_the_first_slices():
p = anna(r=rate("42000", "year", "GBP"))
old = p.move_start(1, D("2026-11-09"), now=NOW, by="m")
assert old == D("2026-11-02") and p.engagements[1].start_date == D("2026-11-09")
assert p.assignments[1].first().valid_from == D("2026-11-09") and p.pay[1].first().valid_from == D("2026-11-09")
def test_hr_31_the_work_email_is_written_back_as_a_new_detail_slice():
p = anna("2026-10-01")
assert p.write_back_work_email("Anna.Berg@co.example", D("2026-10-17"), now=NOW, by="it")
assert p.detail.at(D("2026-10-17")).data["work_email"] == "anna.berg@co.example"
assert not p.write_back_work_email("anna.berg@co.example", D("2026-10-17"), now=NOW, by="it")
# ── HR-12: the org and the derived manager ─────────────────────────────────────────────────
def org():
o = Org("acme", "Acme", "LE")
o.keep_unit("OPS", D("2020-01-01"), {"name": "Operations"})
o.keep_unit("STUDIO", D("2020-01-01"), {"name": "Studio", "parent_org_unit_id": "OPS"})
o.keep_unit("CAMP", D("2020-01-01"), {"name": "Campaigns", "parent_org_unit_id": "OPS"})
for p, u in (("P-OPS", "OPS"), ("P-STUDIO", "STUDIO")):
o.keep_position(p, D("2020-01-01"), {"org_unit_id": u})
o.keep_unit(u, D("2020-01-01"), {"chief_position_id": p})
return o
def seat(pid, unit, pos=None, frm="2020-01-01", to=OPEN, override=None):
return services.Seat(pid, 1, D(frm), to if isinstance(to, date) else D(to), "acme", unit, pos, "x", "LON", "FT",
None, override, "employee_ft")
def test_hr_12_a_move_that_would_make_a_cycle_is_refused():
o = org()
with pytest.raises(RuleBroken, match="cycle"):
o.keep_unit("OPS", D("2026-12-01"), {"parent_org_unit_id": "STUDIO"})
def test_hr_12_a_second_chief_for_one_unit_on_one_date_is_refused():
o = org()
o.keep_position("P-2", D("2020-01-01"), {"org_unit_id": "STUDIO"})
with pytest.raises(RuleBroken, match="already has a chief"):
o.keep_unit("STUDIO", D("2026-12-01"), {"chief_position_id": "P-2"})
def test_hr_12_a_unit_moved_from_a_date_changes_the_tree_only_from_then():
o = org()
o.keep_unit("STUDIO", D("2026-12-01"), {"parent_org_unit_id": "CAMP"})
assert o.unit_at("STUDIO", D("2026-11-30")).parent_org_unit_id == "OPS"
assert o.unit_at("STUDIO", D("2026-12-01")).parent_org_unit_id == "CAMP"
assert o.unit_at("STUDIO", D("2026-12-01")).chief_position_id == "P-STUDIO" # carried into the new slice
def test_hr_39_manager_of_walks_up_and_the_override_wins():
"""SC-HR-05: Lena holds Studio's chief until 30 Nov; from 1 Dec Studio's chief is vacant → Omar (Operations)."""
o = org()
d = services.Directory([seat(1, "STUDIO", "P-STUDIO", to="2026-11-30"), seat(2, "OPS", "P-OPS"), seat(3, "STUDIO")])
orgs = {"acme": o}
assert services.manager_of(3, D("2026-11-30"), d, orgs) == 1
assert services.manager_of(3, D("2026-12-01"), d, orgs) == 2
assert services.manager_of(1, D("2026-11-30"), d, orgs) == 2 # a chief's manager is one level up
d2 = services.Directory(d.seats + [seat(4, "STUDIO", override=9, frm="2026-01-01")])
assert services.manager_of(4, D("2026-11-30"), d2, orgs) == 9
assert services.line_of(2, D("2026-11-30"), d, orgs, {1, 2, 3}) == {1}
def test_hr_18_fte_is_schedule_minutes_over_the_base_times_the_percentage():
cfg = Config({"work_schedule": [{"schedule_code": "PT3", "weekly_minutes": 1350, "full_time_weekly_minutes": 2250}]})
s = services.Seat(1, 1, D("2026-01-01"), OPEN, "acme", "S", None, "x", None, "PT3", Decimal("80"), None, "e")
assert services.fte_of(s, cfg) == Decimal("0.4800")
# ── HR-11: playbooks ───────────────────────────────────────────────────────────────────────
def validate(p, kinds=None):
p.validate(known=lambda c: True, destructive=lambda c: c in ("identity.delete_user", "identity.suspend_user"),
kind_of_event=lambda et: (kinds or {}).get(et, "Left"), groups=lambda g: True)
def step(n, kind="task", cap=None, deps=(), anchor="run.created", **kw):
return pbk.PlaybookStep(n, kind, f"step {n}", "it", {"anchor": anchor, "offset": 0, "unit": "days", "time": None},
cap, {} if cap else None, list(deps), **{"key": f"step_{n}", **kw})
def test_hr_11_a_destructive_step_with_no_approval_before_it_is_refused_in_words():
with pytest.raises(RuleBroken, match="must depend on an approval step"):
validate(pbk.Playbook("p", 1, "P", ["leave"], steps=[step(1, "action", "identity.suspend_user")]))
def test_hr_11_a_delete_with_no_transfer_before_it_is_refused():
with pytest.raises(RuleBroken, match="drive.transfer_ownership"):
validate(pbk.Playbook("p", 1, "P", ["leave"], steps=[step(1, "approval", approval_group="it"),
step(2, "action", "identity.delete_user", [1])]))
def test_hr_11_a_dependency_cycle_is_refused():
with pytest.raises(RuleBroken, match="circle"):
validate(pbk.Playbook("p", 1, "P", ["leave"], steps=[step(1, deps=[2]), step(2, deps=[1])]))
def test_a_conversion_playbook_removes_no_access():
with pytest.raises(RuleBroken, match="keeps access"):
validate(pbk.Playbook("p", 1, "P", ["conversion"], steps=[step(1, "approval", approval_group="it"),
step(2, "action", "groups.remove_member", [1])]),
kinds={"conversion": "Converted"})
def test_the_most_specific_playbook_wins_and_a_tie_is_detected():
a = pbk.Playbook("a", 1, "A", ["hire"], steps=[step(1)])
b = pbk.Playbook("b", 1, "B", ["hire"], worker_types=["employee_ft"], steps=[step(1)])
c = pbk.Playbook("c", 1, "C", ["hire"], org_unit_id="STUDIO", worker_types=["employee_ft"], steps=[step(1)])
assert pbk.select([a, b, c], "hire", "employee_ft", None, ["STUDIO", "OPS"]).playbook_id == "c"
assert pbk.select([a, b, c], "hire", "intern", None, ["STUDIO", "OPS"]).playbook_id == "a"
assert pbk.ties(b, pbk.Playbook("b2", 1, "B2", ["hire"], worker_types=["employee_ft", "intern"], steps=[step(1)]))
assert not pbk.ties(a, b)
# ── runs ───────────────────────────────────────────────────────────────────────────────────
def ctx(**kw):
cal = work.WorkCalendar(holidays=frozenset(CFG.holidays("gb-eaw")))
base = dict(anchors={"start_date": D("2026-10-28"), "last_day": D("2026-10-30"), "event_date": D("2026-10-19"),
"run.created": NOW},
timezone="Europe/London", calendar=cal, values={"account_email": "a@co.example", "target_role": "member"},
access_add=[{"kind": "group", "target": "team@co.example", "role": "member"},
{"kind": "group", "target": "all@co.example", "role": "member"}],
facts={"worker_type": "employee_ft", "via_company": False},
is_destructive=lambda c: c in ("identity.delete_user", "identity.suspend_user"))
base.update(kw)
return R.ExpansionContext(**base)
def starter_run():
pb = pbk.Playbook("employee-onboarding", 3, "EO", ["hire"], steps=[
pbk.PlaybookStep(**{**s, "timing": s["timing"]}) for s in gb.onboarding("x", "x", [], "employment_contract")["steps"]])
return R.expand("onboarding:1:1:1", "onboarding", 1, 1, 1, pb, ctx(), now=NOW)
def test_sc_hr_14_due_times_are_wall_clock_in_london_across_the_clock_change():
r = starter_run()
assert r.steps["5"].due_at == datetime(2026, 10, 23, 8, tzinfo=timezone.utc) # 09:00 BST
off = gb.offboarding("x", "x", [], False)["steps"][2]
o = R.expand("offboarding:1:1:2", "offboarding", 1, 1, 2,
pbk.Playbook("eo", 1, "x", ["leave"], steps=[pbk.PlaybookStep(**{**off, "depends_on": []})]), ctx(),
now=NOW)
assert o.steps["3"].due_at == datetime(2026, 10, 30, 17, 30, tzinfo=timezone.utc) # 17:30 GMT
def test_a_step_fans_out_per_access_row_and_its_dependants_wait_on_all():
r = starter_run()
assert {"6:team@co.example", "6:all@co.example"} <= set(r.steps)
assert r.steps["6:all@co.example"].input == {"group": "all@co.example", "member_email": "a@co.example",
"role": "member"}
assert r.steps["2"].state == R.SKIPPED # rate inside the band: no approval asked
def test_hr_10_the_version_is_frozen_in_the_run():
r = starter_run()
assert (r.playbook_id, r.playbook_version) == ("employee-onboarding", 3)
def test_hr_07_a_refused_approval_cancels_what_depends_on_it_and_a_granted_one_lets_it_act():
pb = pbk.Playbook("o", 1, "o", ["leave"], steps=[step(1, "approval", approval_group="it"),
step(2, "action", "identity.delete_user", [1])])
r = R.expand("offboarding:1:1:2", "offboarding", 1, 1, 2, pb, ctx(), now=NOW)
assert not r.approval_granted(r.steps["2"])
with pytest.raises(Incomplete, match="reason"):
r.approve("1", False, reason="", by="it", at=NOW)
r.approve("1", False, reason="legal hold", by="it", at=NOW)
assert r.steps["2"].state == R.CANCELLED and "legal hold" in r.steps["2"].message
r2 = R.expand("offboarding:2:1:2", "offboarding", 2, 1, 2, pb, ctx(), now=NOW)
assert r2.approve("1", True, reason="", by="it", at=NOW) and not r2.approve("1", True, reason="", by="it", at=NOW)
assert r2.approval_granted(r2.steps["2"])
def test_hr_08_retiming_moves_only_steps_not_done():
r = starter_run()
r.complete("15", by="fin", at=NOW, calendar=ctx().calendar, timezone="Europe/London")
before = r.steps["15"].due_at
r.retime({**r.anchors, "start_date": D("2026-11-04")}, ctx().calendar, "Europe/London")
assert r.steps["15"].due_at == before and r.steps["5"].due_at == datetime(2026, 10, 30, 9, tzinfo=timezone.utc)
def test_a_done_step_never_acts_again():
r = starter_run()
assert r.complete("15", by="f", at=NOW) and not r.complete("15", by="f", at=NOW)
assert R.apply_outcome(r, r.steps["15"], R.Outcome("done"), now=NOW, max_attempts=3,
backoff=timedelta(minutes=5)) == "already-finished"
def test_hr_36_every_outcome_of_an_action():
r = starter_run()
s = r.steps["5"]
assert R.apply_outcome(r, s, R.Outcome("in_progress"), now=NOW, max_attempts=3, backoff=timedelta(minutes=5)) == "in_progress"
assert s.attempts == 0 and s.state == R.PENDING # not a failure
for i in range(2):
assert R.apply_outcome(r, s, R.Outcome("failed", retryable=True, message="503"), now=NOW, max_attempts=3,
backoff=timedelta(minutes=5)) == "retry-later"
assert R.apply_outcome(r, s, R.Outcome("failed", retryable=True, message="503"), now=NOW, max_attempts=3,
backoff=timedelta(minutes=5)) == "manual-after-failures"
assert s.manual and s.responsible_group == "it" and "503" in s.manual_reason
t = r.steps["10"]
R.apply_outcome(r, t, R.Outcome("manual", manual_input={"a": 1}, plan="do it", manual_reason="no provider"),
now=NOW, max_attempts=3, backoff=timedelta(minutes=5))
assert t.manual and t.input == {"a": 1} and t.state == R.WAITING
def test_hr_23_cancelling_creates_a_reversal_task_per_done_action_and_reverses_nothing():
r = starter_run()
for no in ("5", "6:team@co.example"):
R.apply_outcome(r, r.steps[no], R.Outcome("done", provider="gws"), now=NOW, max_attempts=3,
backoff=timedelta(minutes=1))
rev = r.cancel("withdrew", by="hr", at=NOW)
assert [x.input["capability"] for x in rev] == ["identity.suspend_user", "groups.remove_member"]
assert all(x.kind == "task" and x.state == R.WAITING for x in rev) and r.state == "cancelled"
assert r.steps["12"].state == R.CANCELLED and r.cancel("again", by="hr", at=NOW) == []
def test_the_contract_gate_blocks_until_the_status_determination_is_recorded():
cond = gb.onboarding("x", "x", [], "freelance_agreement")["steps"][2]["condition"]
facts = {"via_company": True, "client_size": "medium", "compliance": []}
assert R.blocking_requirements(cond, facts) == ["status_determination"]
assert R.blocking_requirements(cond, {**facts, "compliance": ["status_determination"]}) == []
assert R.blocking_requirements(cond, {**facts, "client_size": "small"}) == []
# ── money, payroll, privacy, intake, the partner ───────────────────────────────────────────
def test_sc_hr_16_rounding_once_at_the_payable_line():
r = Rate(Decimal("13.3333"), "hour", "GBP")
assert str(payable(r, minutes=450).amount) == "100.00"
per_day = sum((payable(r, minutes=150).amount for _ in range(3)), Decimal(0))
assert per_day == Decimal("99.99") # what rounding per day would pay
def test_hr_19_freelancers_are_never_in_the_extract_and_a_sent_batch_is_frozen():
p = anna("2026-10-01", r=rate("42000", "year", "GBP"))
f = Person.from_hire(PersonKey.of(None, "f@x.y"), {"legal_given_name": "F", "legal_family_name": "G"},
spec("2026-10-01", wt="freelancer_sole", r=rate("450", "day", "GBP")), SOLE,
today=D("2026-10-01"), now=NOW, by="hr")
lines = lines_for(p, "2026-11", "LE", {"employee_ft"}) + lines_for(f, "2026-11", "LE", {"employee_ft"})
assert [l.person_id for l in lines] == [p.person_id]
sent = Batch("LE|2026-11", "LE", "2026-11", "sent", sent_at=NOW)
with pytest.raises(RuleBroken, match="frozen"):
build("LE", "2026-11", {"2026-11": lines}, [], now=NOW, existing=sent)
def test_hr_19_a_change_to_a_sent_period_is_a_retro_line_in_the_next_batch():
p = anna("2026-10-01", r=rate("42000", "year", "GBP"))
nov = lines_for(p, "2026-11", "LE", {"employee_ft"})
p.change_pay(D("2026-11-01"), "review", "pay_change", now=NOW, by="hr", rate=rate("44000", "year", "GBP"))
dec = build("LE", "2026-12", {"2026-12": lines_for(p, "2026-12", "LE", {"employee_ft"}),
"2026-11": lines_for(p, "2026-11", "LE", {"employee_ft"})}, nov, now=NOW)
retro = [l for l in dec.lines if l.change_kind == "retro"]
assert len(retro) == 1 and retro[0].for_period == "2026-11" and retro[0].rate_amount == Decimal("44000.0000")
def test_hr_21_a_subject_access_request_is_due_one_month_from_receipt():
assert due_on(D("2026-10-01")) == D("2026-11-01") and due_on(D("2026-01-31")) == D("2026-02-28")
def test_hr_22_a_legal_obligation_hold_keeps_a_category_and_says_why():
plan = erasure_plan(CFG, D("2025-01-31"), D("2026-10-20"))
assert plan["payroll"][0] == "kept" and "legal obligation" in plan["payroll"][1]
assert plan["sensitive_bank"][0] == "purge" and plan["person_detail"][0] == "anonymise"
def test_hr_40_retention_is_due_only_past_its_period():
rule = {"keep_months": 12}
assert retention_due(rule, D("2025-09-15"), D("2026-10-20")) and not retention_due(rule, D("2025-11-25"), D("2026-10-20"))
assert not retention_due(rule, None, D("2026-10-20"))
def test_hr_02_the_intake_status_names_fields_and_never_a_value():
fields = CFG.intake_fields("gb", "employee_ft")
st = intake_status(fields, {"home_address": {"value": {"home_address": "1 Lane"}, "has_file": False}})
assert st["done"] == 1 and st["total"] == len(fields) and "1 Lane" not in str(st)
assert "right_to_work" in st["missing"]
def test_hr_50_the_partner_card_carries_a_name_and_a_work_contact_only():
p = anna("2026-10-01")
p.write_back_work_email("anna@co.example", D("2026-10-17"), now=NOW, by="it")
card = services.person_card(p, D("2026-10-17"), CFG, "GB")
assert set(card) == {"person_id", "legal_name", "preferred_name", "work_email", "work_phone", "country", "roles",
"works_for"}
assert card["roles"] == ["employee"] and "anna@example.org" not in str(card)
def test_hr_33_the_draft_adapter_refuses_a_sensitive_field_before_sending():
with pytest.raises(models_adapter.DraftFailed, match="never given"):
models_adapter.ModelsDrafter().draft("x", {"preferred_name": "A", "bank": "1"}, "kimi-k2.6")
def test_hr_06_less_than_statutory_notice_is_warned_not_refused():
assert services.statutory_notice_weeks(D("2020-01-06"), D("2026-11-30")) == 6
assert "s.86" in services.notice_warning(D("2020-01-06"), D("2026-11-30"), D("2026-11-20"))
assert services.notice_warning(D("2026-11-01"), D("2026-11-20"), D("2026-11-19")) is None
def test_hr_28_catalogue_rules():
cfg = Config(gb.catalogues())
with pytest.raises(RuleBroken, match="no event type"):
validate_row(CATALOGUES["event_reason"], {"event_type": "nope", "reason": "x"}, cfg)
with pytest.raises(RuleBroken, match="cannot write"):
validate_row(CATALOGUES["infogroup"], {"event_type": "transfer", "record": "pay"}, cfg)
with pytest.raises(Incomplete, match="payroll"):
validate_row(CATALOGUES["worker_type"], {"worker_type": "x", "label": "X", "category": "intern",
"bp_roles": []}, cfg)
with pytest.raises(RuleBroken, match="never given"):
validate_row(CATALOGUES["draft_instruction"], {"draft_code": "x", "instructions": "y",
"fields": ["bank"]}, cfg)
assert validate_row(CATALOGUES["event_reason"], {"event_type": "leave", "reason": "end_of_project_2"}, cfg)
def test_the_gb_pack_loads_the_published_bank_holidays():
h = CFG.holidays("gb-eaw")
assert D("2026-12-25") in h and D("2026-08-31") in h and len(h) > 50
def test_tc1_refuses_an_overlap_however_it_got_there():
from brick_hr.domain.slices import Slice
t = Timeline("x", [Slice(D("2026-01-01"), D("2026-01-31"), {}), Slice(D("2026-01-20"), OPEN, {})])
with pytest.raises(RuleBroken, match="overlap"):
t.check_tc1()
p = anna("2026-01-05")
p.assignments[1].slices.append(Slice(D("2026-03-01"), OPEN, {"org_unit_id": "X", "job_code": "y"}))
with pytest.raises(RuleBroken, match="overlap"):
p.check()
# ── HR-11: a playbook as BPMN 2.0 (domain/bpmn.py) ─────────────────────────────────────────────────────────────
def _bpmn(body: dict, version: int = 1) -> str:
from brick_hr.application.organisation import _playbook_from, bpmn_setting
from brick_hr.domain import bpmn
pb = _playbook_from(body, version, "hr_admin.1", NOW)
return bpmn.to_bpmn(pb, bpmn_setting(CFG, pb))
def _gb(playbook_id: str) -> dict:
return next(b for b in gb.starter_playbooks() if b["playbook_id"] == playbook_id)
@pytest.mark.parametrize("playbook_id", [b["playbook_id"] for b in gb.starter_playbooks()])
def test_hr_11_every_gb_playbook_is_valid_bpmn_2_with_its_dependencies_as_flows(playbook_id):
from bpmn_check import check, upstream_steps
body = _gb(playbook_id)
root = check(_bpmn(body)) # the OMG XSD + references, lanes, DI, reachability
pb = __import__("brick_hr.application.organisation", fromlist=["x"])._playbook_from(body, 1, "x", NOW)
up = upstream_steps(root)
assert set(up) == {s.key for s in pb.steps}
for s in pb.steps: # what runs before a step is exactly what it depends on
assert up[s.key] == set(pb.keys_of(pb.ancestors(s.step_no))), s.key
def test_hr_11_the_onboarding_bpmn_says_what_the_checklist_does():
from bpmn_check import q
import xml.etree.ElementTree as ET
xml = _bpmn(_gb("employee-onboarding"))
root = ET.fromstring(xml)
proc = root.find(q("bpmn:process"))
by_id = {e.get("id"): e for e in root.iter() if e.get("id")}
doc = lambda i: by_id[i].findtext(q("bpmn:documentation")) or "" # noqa: E731
lanes = [ln.get("id") for ln in proc.iter(q("bpmn:lane"))]
assert lanes == ["Lane_hr", "Lane_it", "Lane_manager", "Lane_buddy", "Lane_finance"]
assert by_id["Lane_hr"].get("name") == "Operations / HR (hr)"
assert "onboarding playbook employee-onboarding, version 1" in proc.findtext(q("bpmn:documentation"))
# a task is a user task; an action a service task naming its capability and the manual path
assert by_id["Task_laptop_ready"].tag == q("bpmn:userTask") and by_id["Task_laptop_ready"].get("name") == "Laptop ready"
assert by_id["Task_create_account"].tag == q("bpmn:serviceTask")
assert "identity.provision_user" in doc("Task_create_account") and "manual task for IT / ops carrying the same input" in doc("Task_create_account")
ins = [d.get("name") for d in by_id["Task_create_account"].iter(q("bpmn:dataInput"))]
assert ins == ["given_name", "family_name", "primary_email", "recovery_email", "org_unit",
"initial_password_delivery"]
assert [d.get("name") for d in by_id["Task_create_account"].iter(q("bpmn:dataOutput"))] == ["account_id", "primary_email",
"created"]
# a draft is drafted, then approved by a person
assert by_id["Task_welcome_email__draft"].get("name") == "Draft: Welcome email"
assert by_id["Task_welcome_email"].tag == q("bpmn:userTask") and by_id["Task_welcome_email"].get("name").startswith("Approve: ")
# the condition, in words, on an exclusive gateway; the skip is its default flow
gw = by_id["Gateway_approve_rate__if"]
assert gw.tag == q("bpmn:exclusiveGateway") and gw.get("default")
conds = [f.findtext(q("bpmn:conditionExpression")) for f in proc.iter(q("bpmn:sequenceFlow"))
if f.get("sourceRef") == "Gateway_approve_rate__if" and f.find(q("bpmn:conditionExpression")) is not None]
assert conds == ["the agreed rate is outside the pay band"]
assert by_id["Task_create_account__record_right_to_work_check"].get("name") == "Record the right to work check"
assert "worker type is one of employee_ft" in by_id["Gateway_create_account__requires_right_to_work_check"].findtext(q("bpmn:documentation"))
# the timing rule as a timer before the step
assert by_id["Timer_create_account"].findtext(f"{q('bpmn:timerEventDefinition')}/{q('bpmn:timeDate')}") == \
"3 working days before start_date, 09:00 Europe/London"
# outside systems by capability family, never a vendor
parts = {p.get("name") for p in root.iter(q("bpmn:participant"))}
assert {"Identity (work accounts)", "Groups", "Mail", "Contracts (signature)", "The starter (public form)"} <= parts
assert not any(v in xml.lower() for v in ("google", "microsoft", "slack", "docusign"))
assert by_id["MessageFlow_Task_create_account"].get("targetRef") == "Participant_identity"
assert by_id["Task_add_to_groups"].find(q("bpmn:multiInstanceLoopCharacteristics")) is not None # one per group row
# inputs and outputs: the starter's details, the draft the email sends, the contract
names = {d.get("name") for d in proc.iter(q("bpmn:dataObjectReference"))}
assert names == {"Starter's details", "Draft: Welcome email",
"Contract: Employment contract (with the written statement)"}
reads = [a for a in by_id["Task_send_welcome_email"].iter(q("bpmn:dataInputAssociation"))]
assert [a.findtext(q("bpmn:sourceRef")) for a in reads] == ["DataObjectReference_draft_welcome_email"]
assert by_id[reads[0].findtext(q("bpmn:targetRef"))].get("name") == "body_text"
written = [r.text for a in by_id["Task_starter_details"].iter(q("bpmn:dataOutputAssociation"))
for r in a.findall(q("bpmn:sourceRef"))]
assert "Task_starter_details_out_ni_number" in written and len(written) == 7 # an employee: no company or VAT number
assert "Task_starter_details_out_company_number" not in written
def test_hr_11_offboarding_waits_on_both_branches_before_the_delete():
from bpmn_check import q
import xml.etree.ElementTree as ET
root = ET.fromstring(_bpmn(_gb("employee-offboarding")))
by_id = {e.get("id"): e for e in root.iter() if e.get("id")}
assert by_id["Gateway_delete_account__join"].tag == q("bpmn:parallelGateway")
into = sorted(f.get("sourceRef") for f in root.iter(q("bpmn:sequenceFlow")) if f.get("targetRef") == "Gateway_delete_account__join")
assert into == ["Task_approve_account_deletion", "Task_transfer_drive"] # the drive transfer AND the delete approval
text = by_id["Task_delete_account"].findtext(q("bpmn:documentation"))
assert "Destructive: it runs only after an approval it depends on was GRANTED" in text
assert "drive.transfer_status" in text
def test_hr_11_the_bpmn_is_the_same_for_the_same_playbook():
import copy
import random
body = _gb("employee-onboarding")
shuffled = copy.deepcopy(body)
random.Random(7).shuffle(shuffled["steps"])
assert _bpmn(body) == _bpmn(body) == _bpmn(shuffled)
def test_hr_11_a_playbook_change_shows_in_the_bpmn():
import copy
body = copy.deepcopy(_gb("employee-onboarding"))
before = _bpmn(body)
body["steps"][4]["timing"]["offset"] = -4 # create the account a day earlier
body["steps"].append({"step_no": 16, "kind": "task", "title": "Desk booked", "responsible_group": "it",
"depends_on": [11]})
after = _bpmn(body, version=2)
assert "3 working days before start_date, 09:00" in before and "4 working days before start_date, 09:00" in after
assert "Desk booked" not in before and 'name="Desk booked"' in after
assert 'sourceRef="Task_laptop_ready" targetRef="Task_desk_booked"' in after
assert 'id="Process_employee-onboarding_v2"' in after
# ── HR-29: monitoring the onboarding programme ─────────────────────────────────────────────
def _mrun(rid, created, steps, version=1, start="2026-11-02", cancelled=False):
from brick_hr.domain.monitoring import RunFacts # noqa: F401
r = R.Run(rid, "onboarding", int(rid[-1]), 1, "1", "employee-onboarding", version, created,
anchors={"start_date": D(start)}, cancelled_at=created if cancelled else None)
for i, (group, cap, state, due, done) in enumerate(steps, 1):
r.steps[str(i)] = R.RunStep(str(i), i, "action" if cap else "task", f"s{i}", group, capability=cap,
state=state, due_at=due, done_at=done)
return r
def test_hr_29_the_monitor_derives_lead_time_overdue_first_access_and_versions_in_the_business_zone():
from brick_hr.domain.monitoring import RunFacts, monitor
utc = lambda s: datetime.fromisoformat(s).replace(tzinfo=timezone.utc) # noqa: E731
now = utc("2026-11-10T12:00")
done = _mrun("r1", utc("2026-10-20T08:00"), [
("it", "identity.provision_user", "done", utc("2026-10-28T09:00"), utc("2026-10-29T23:30")), # GMT after 25 Oct: 29 Oct
("manager", None, "done", utc("2026-11-03T09:00"), utc("2026-11-04T23:30"))])
late = _mrun("r2", utc("2026-11-01T09:00"), [
("it", "groups.add_member", "done", utc("2026-11-01T09:00"), utc("2026-11-03T10:00")),
("it", None, "pending", utc("2026-11-05T09:00"), None),
("buddy", None, "waiting", utc("2026-11-06T09:00"), None),
("hr", None, "pending", utc("2026-12-01T09:00"), None)], version=2)
gone = _mrun("r3", utc("2026-11-01T09:00"), [("it", None, "pending", utc("2026-11-02T09:00"), None)],
cancelled=True)
units = {"r1": "STUDIO", "r2": "CAMP", "r3": "CAMP"}
out = monitor([done, late, gone], lambda r: RunFacts(r.anchors["start_date"], units[r.run_id]), now,
"Europe/London", lambda pid, v: {"variation": v - 1, "warning": 0})
assert out["lead_time"] == {"count": 1, "median_days": 15, "max_days": 15, "open_runs": 1,
"open_median_age_days": 9} # 20 Oct → 4 Nov; 1 Nov → 10 Nov
assert out["first_access"] == {"count": 2, "median_days": -4, "max_days": 1} # 29 Oct (-4) and 3 Nov (+1)
assert out["steps_overdue"] == {"total": 2, "by_owner": [{"responsible_group": "buddy", "overdue": 1},
{"responsible_group": "it", "overdue": 1}],
"by_team": [{"org_unit_id": "CAMP", "overdue": 2}]} # a cancelled run counts nothing
assert [(v["version"], v["runs"], v["open"], v["deviations"]) for v in out["playbook_versions"]] == [
(1, 1, 0, {"variation": 0, "warning": 0}), (2, 1, 1, {"variation": 1, "warning": 0})]
def test_hr_29_a_late_evening_step_counts_on_its_business_date_not_the_utc_one():
from brick_hr.domain.monitoring import RunFacts, monitor
utc = lambda s: datetime.fromisoformat(s).replace(tzinfo=timezone.utc) # noqa: E731
# 23:30 UTC on 1 Jul is 00:30 on 2 Jul in London (BST): access came the day after the start
r = _mrun("r1", utc("2026-06-20T08:00"), [("it", "identity.provision_user", "done", None, utc("2026-07-01T23:30"))],
start="2026-07-01")
out = monitor([r], lambda x: RunFacts(D("2026-07-01"), "STUDIO"), utc("2026-07-05T12:00"), "Europe/London",
lambda pid, v: None)
assert out["first_access"]["median_days"] == 1 and out["lead_time"]["median_days"] == 12
def test_hr_55_the_dated_setting_in_force_is_the_reasons_latest_row_then_the_defaults_then_last_day_end():
from datetime import date as d
from brick_hr.domain.configuration import Config, revoke_access_timing
k = "offboarding.revoke_access_at"
row = lambda scope, day, v: {"setting_key": k, "scope": scope, "valid_from": d(2026, 10, day), "value": v} # noqa: E731
cfg = Config(dated=[row("", 10, "immediate"), row("", 20, "last_day_end"), row("dismissal", 15, "immediate")])
assert Config().dated_setting(k, d(2026, 10, 1)) == "last_day_end" # no rows: the default
assert cfg.dated_setting(k, d(2026, 10, 9)) == "last_day_end" # before the first row
assert cfg.dated_setting(k, d(2026, 10, 12)) == "immediate"
assert cfg.dated_setting(k, d(2026, 10, 20), "resignation") == "last_day_end" # no override: the default's
assert cfg.dated_setting(k, d(2026, 10, 14), "dismissal") == "immediate" # the default's, before its own
assert cfg.dated_setting(k, d(2026, 10, 21), "dismissal") == "immediate" # its own beats the default's
assert revoke_access_timing("immediate")["anchor"] == "run.created" and revoke_access_timing("last_day_end") is None