"""What makes hr fast on DataK3 must not change what it answers: the configuration cache sees the next change, a
save that writes only what changed leaves storage exactly as a full save would, and the tick finds a due step
— and never a done or not-yet-due one — by a query rather than by loading everything.
Statements are counted with an engine event on the test's SQLite (scripts/bench_live.py counts the same way, live).
"""
from contextlib import contextmanager
from datetime import date, datetime, timedelta
import pytest
from sqlalchemy import event, func, insert, select
from erp_platform import work
from erp_platform.repository import Conflict, upsert
from brick_hr.infrastructure import repositories as R
from brick_hr.infrastructure import tables as T
from brick_hr.infrastructure.tables import HrRepository
from conftest import headers
from seed import hire_body, seed
HR = headers("hr_admin")
@contextmanager
def counted(engine):
seen = []
def on(conn, cursor, statement, params, context, executemany):
seen.append(statement)
event.listen(engine, "before_cursor_execute", on)
try:
yield seen
finally:
event.remove(engine, "before_cursor_execute", on)
def _reads(seen, table):
return [s for s in seen if s.lstrip().upper().startswith("SELECT") and f"FROM {table}" in s]
@pytest.fixture
def seeded(engine, api, drafter):
work.set_clock(at=work.parse_instant("2026-10-19T10:00 Europe/London"), freeze=True)
seed(lambda p, b: api("POST", p, HR, b))
return engine
# ── the configuration cache ────────────────────────────────────────────────────────────────
def test_a_warm_request_reads_the_versions_once_and_no_catalogue(seeded, api):
api("GET", "/my-tasks", HR) # warm
with counted(seeded) as seen:
s, _ = api("GET", "/my-tasks", HR)
assert s == 200
assert len(_reads(seen, "hr_config_version")) == 1
assert not _reads(seen, "hr_job") and not _reads(seen, "hr_worker_type") and not _reads(seen, "hr_playbook")
def test_a_hire_reads_the_versions_once_for_all_of_its_units_of_work(seeded, api):
api("GET", "/my-tasks", HR)
with counted(seeded) as seen:
s, h = api("POST", "/hires", HR, hire_body("Ola", "Once", "2026-11-02"))
assert s == 200, h
assert len(_reads(seen, "hr_config_version")) == 1 # the hire, its run: one request
def test_a_configuration_change_is_seen_by_the_next_request(seeded, api):
api("GET", "/configuration/job", HR) # the snapshot is now cached
s, _ = api("POST", "/configuration/job", HR, {"rows": [{"job_code": "designer", "title": "Senior Designer"}]})
assert s == 200
with counted(seeded) as seen:
s, out = api("GET", "/configuration/job", HR)
assert {r["job_code"]: r["title"] for r in out["rows"]}["designer"] == "Senior Designer"
assert len(_reads(seen, "hr_job")) == 1 # only the changed table is read again
assert not _reads(seen, "hr_worker_type")
def test_a_change_made_by_another_replica_is_seen_by_the_next_request(seeded, api):
"""Another replica writes the row and a new token in one transaction; this process's cache must yield."""
api("GET", "/configuration/job", HR)
with seeded.begin() as conn: # what the other replica commits
upsert(conn, T.job, [{"job_code": "designer", "title": "Lead Designer"}], "job_code", insert_only=())
upsert(conn, T.config_version, [{"key": "hr_job", "token": "from-another-replica"}], "key", insert_only=())
s, out = api("GET", "/configuration/job", HR)
assert {r["job_code"]: r["title"] for r in out["rows"]}["designer"] == "Lead Designer"
def test_without_a_new_token_the_cached_rows_are_served(seeded, api):
"""The control for the test above: a row written WITHOUT its token is not read — so the cache is in use,
and the token (not luck) is what made the change visible."""
api("GET", "/configuration/job", HR)
with seeded.begin() as conn:
upsert(conn, T.job, [{"job_code": "designer", "title": "Untracked"}], "job_code", insert_only=())
s, out = api("GET", "/configuration/job", HR)
assert {r["job_code"]: r["title"] for r in out["rows"]}["designer"] == "Designer"
def test_a_playbook_and_an_org_change_are_seen_by_the_next_request(seeded, api):
s, pbs = api("GET", "/playbooks", HR)
before = {p["playbook_id"]: p["version"] for p in pbs["playbooks"]}
pb = next(p for p in pbs["playbooks"] if p["playbook_id"] == "onboarding_employee") if \
"onboarding_employee" in before else pbs["playbooks"][0]
body = {k: pb[k] for k in ("playbook_id", "name", "event_types", "worker_types", "reasons", "org_unit_id",
"status")}
body["name"] = pb["name"] + " (edited)"
body["steps"] = [{k: v for k, v in st.items() if k in ("step_no", "kind", "title", "responsible_group", "timing",
"capability", "input_template", "depends_on", "condition",
"approval_group", "draft_instruction", "template_kind",
"fan_out")} for st in pb["steps"]]
s, out = api("POST", "/playbooks", HR, body)
assert s == 200, out
s, pbs = api("GET", "/playbooks", HR, params={"playbook_id": pb["playbook_id"]})
assert pbs["playbooks"][0]["version"] == before[pb["playbook_id"]] + 1
s, _ = api("POST", "/org", HR, {"org_id": "acme", "kind": "unit", "id": "NEW", "valid_from": "2020-01-01",
"values": {"name": "New team", "parent_org_unit_id": "OPS"}})
assert s == 200
s, chart = api("GET", "/org/acme", HR)
assert "NEW" in {u["org_unit_id"] for u in chart["units"]}
def test_a_rolled_back_change_never_reaches_the_cache(seeded, api):
api("GET", "/configuration/job", HR)
try:
with seeded.begin() as conn:
uow = HrRepository(conn)
uow.config.save_rows("job", [{"job_code": "designer", "title": "Never committed"}])
assert {r["job_code"]: r["title"] for r in uow.config.load().all("job")}["designer"] == "Never committed"
raise RuntimeError("roll it back")
except RuntimeError:
pass
s, out = api("GET", "/configuration/job", HR)
assert {r["job_code"]: r["title"] for r in out["rows"]}["designer"] == "Designer"
def test_a_reader_cannot_change_the_cached_snapshot(seeded, api):
with seeded.begin() as conn:
cfg = HrRepository(conn).config.load()
cfg.job("designer")["title"] = "scribbled on"
cfg.rows["job"].clear()
with seeded.begin() as conn:
assert HrRepository(conn).config.load().job("designer")["title"] == "Designer"
# ── a save writes only what changed, and the result is what a full save stores ─────────────
def _stored(engine, pid):
out = {}
with engine.connect() as conn:
for part in T.PERSON.parts:
rows = conn.execute(select(part.table).where(part.table.c.person_id == pid)).all()
out[part.name] = sorted(tuple((k, v) for k, v in r._mapping.items() if k != "updated_at") for r in rows)
return out
def test_a_slice_change_writes_only_the_new_and_the_delimited_slice(seeded, api):
s, h = api("POST", "/hires", HR, hire_body("Pia", "Pay", "2026-11-02", rate_amount="40000.0000",
rate_basis="year", currency="GBP"))
pid = int(h["person_id"])
s, v = api("GET", f"/people/{pid}", HR)
with counted(seeded) as seen:
s, out = api("POST", f"/people/{pid}/pay", HR, {"effective_date": "2026-12-01", "reason": "review",
"rate_amount": "41000.0000", "rate_basis": "year",
"expected_version": v["version"]})
assert s == 200, out
writes = [x for x in seen if x.lstrip().upper().startswith(("INSERT", "UPDATE", "DELETE"))]
written = {w.split("INTO ")[1].split()[0] for w in writes if "INTO " in w}
# the pay slices, the event, the root and its version claim — not the detail, engagement or assignment
assert written <= {"hr_person_versions", "hr_person", "hr_pay", "hr_event"}, written
assert {"hr_person_versions", "hr_pay", "hr_event"} <= written
def test_the_snapshot_save_stores_exactly_what_the_full_save_does(seeded, api):
"""The same changes saved twice over — once through the snapshot path, once through erp_platform's own
save — leave the same rows (apart from updated_at, a moment)."""
from brick_hr.infrastructure import batching as B
s, h = api("POST", "/hires", HR, hire_body("Sam", "Same", "2026-11-02", rate_amount="40000.0000",
rate_basis="year", currency="GBP"))
pid = int(h["person_id"])
for i, path in enumerate((f"/people/{pid}/pay", f"/people/{pid}/pay")):
api("POST", path, HR, {"effective_date": f"2026-12-0{i + 1}", "reason": "review",
"rate_amount": f"4100{i}.0000", "rate_basis": "year"})
fast = _stored(seeded, pid)
# the same aggregate written back through the platform's save: nothing may differ
with seeded.begin() as conn:
repo = R.PersonRepository(conn)
p = repo.get(pid)
repo._seen.clear() # no snapshot: erp_platform's own save
repo.save(p)
full = _stored(seeded, pid)
strip = lambda d: {k: [tuple(x for x in r if x[0] != "version") for r in v] for k, v in d.items()} # noqa: E731
assert strip(fast) == strip(full)
assert B.Snapshot is __import__("erp_platform").repository.Snapshot # the platform's (0.9.6) path
def test_a_stale_snapshot_save_is_a_conflict_and_writes_nothing(seeded, api):
s, h = api("POST", "/hires", HR, hire_body("Rae", "Race", "2026-11-02"))
pid = int(h["person_id"])
first = api("GET", f"/people/{pid}", HR)[1]["detail_history"][0]["valid_from"]
with seeded.connect() as c1:
t1 = c1.begin()
mine = R.PersonRepository(c1)
p = mine.get(pid) # loaded at version 1
s, other = api("POST", f"/people/{pid}/corrections", HR, {"record": "detail", "valid_from": first,
"values": {"preferred_name": "Other"}})
assert s == 200, other # version 2 commits meanwhile
p.detail.slices[0].data["preferred_name"] = "Mine"
with pytest.raises(Conflict):
mine.save(p)
t1.rollback()
s, v = api("GET", f"/people/{pid}", HR)
assert v["detail"]["preferred_name"] == "Other" and v["version"] == 2
# ── the tick selects what is due by a query ────────────────────────────────────────────────
def _step(conn, run_id, step_no, state, due_at=None, next_check_at=None):
conn.execute(insert(T.run_step).values(run_id=run_id, step_no=step_no, state=state, due_at=due_at,
next_check_at=next_check_at, kind="task"))
def test_the_tick_finds_a_due_step_and_never_a_done_or_not_yet_due_one(engine):
now = datetime(2026, 11, 2, 9, 0)
with engine.begin() as conn:
_step(conn, "due-pending", "1", "pending", due_at=now - timedelta(minutes=1))
_step(conn, "due-recheck", "1", "waiting", due_at=now - timedelta(days=3), next_check_at=now)
_step(conn, "done", "1", "done", due_at=now - timedelta(days=1))
_step(conn, "cancelled", "1", "cancelled", due_at=now - timedelta(days=1))
_step(conn, "not-yet", "1", "pending", due_at=now + timedelta(minutes=1))
_step(conn, "waiting-no-check", "1", "waiting", due_at=now - timedelta(days=1))
_step(conn, "recheck-later", "1", "waiting", due_at=now - timedelta(days=1),
next_check_at=now + timedelta(hours=1))
_step(conn, "no-time", "1", "pending")
with engine.begin() as conn:
uow = HrRepository(conn)
due = uow.runs.needing_attention(work.aware(now))
_events, _ids, due_at_start = uow.runs.tick_start(work.aware(now))
assert due == ["due-pending", "due-recheck"]
assert due_at_start == due
def test_the_tick_loads_only_the_people_whose_dates_fall_in_the_window(seeded, api):
"""Warnings and retention: a query picks the candidates, so a tick with nobody due loads no person."""
for i in range(3):
s, h = api("POST", "/hires", HR, hire_body(f"Quiet{i}", "Person", "2026-11-02"))
assert s == 200, h
api("POST", "/run-due-steps", headers("clock"), {}) # expand, act once
with counted(seeded) as seen:
s, out = api("POST", "/run-due-steps", headers("clock"), {})
assert s == 200, out
assert not _reads(seen, "hr_person_detail") and not _reads(seen, "hr_pay") # no Person was loaded
def test_a_contract_ending_in_the_window_is_a_candidate_and_one_beyond_it_is_not(seeded, api):
s, near = api("POST", "/hires", HR, hire_body("Near", "End", "2026-10-20", worker_type="freelancer_sole",
planned_end="2026-10-30", rate_amount="500.0000",
rate_basis="day", currency="GBP"))
assert s == 200, near
s, far = api("POST", "/hires", HR, hire_body("Far", "End", "2026-10-20", worker_type="freelancer_sole",
planned_end="2027-06-30", rate_amount="500.0000",
rate_basis="day", currency="GBP"))
assert s == 200, far
with seeded.begin() as conn:
found = HrRepository(conn).persons.warning_candidates(date(2026, 10, 19), 14, 28)
assert int(near["person_id"]) in found and int(far["person_id"]) not in found