"""Statement budgets: on DataK3 every statement is a round trip (~100 ms on the pg wire), so the heaviest requests —
hire, record a leaver, complete a run step, the tick, headcount — are counted at the engine
(erp_platform.testing.count_statements), and none of them may grow with the number of UNRELATED people or runs.
Run with -s to see the numbers (BUDGET lines); the README's table is these numbers."""
import pytest
from erp_platform import work
from erp_platform.testing import count_statements
from conftest import headers
from seed import hire_body, seed
HR = headers("hr_admin")
TICK = headers("clock")
@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))
api("GET", "/my-tasks", HR) # the configuration cache is warm
return engine
def _show(label, few, many):
print(f"BUDGET {label}: statements {few.statements} -> {many.statements} "
f"(round trips {few.round_trips} -> {many.round_trips})")
_n = [0]
def _hire(api, start="2026-11-02"):
_n[0] += 1
s, out = api("POST", "/hires", HR, hire_body("Pat", f"Other{_n[0]}", start))
assert s == 200, out
return out
def _others(api, n):
for _ in range(n):
_hire(api)
def _measure(seeded, api, fn):
"""fn() once with 2 unrelated people, once with 10: (few, many)."""
_others(api, 2)
arg = fn("prepare")
with count_statements(seeded) as few:
fn(arg)
_others(api, 8)
arg = fn("prepare")
with count_statements(seeded) as many:
fn(arg)
return few, many
def test_a_hire_does_not_grow_with_the_people_already_there(seeded, api):
def hire(arg):
if arg == "prepare":
return None
_hire(api)
few, many = _measure(seeded, api, hire)
_show("hire", few, many)
assert many.statements <= few.statements + 2, (few.statements, many.statements)
assert few.statements <= 17, few.statements # was 15 on 0.5.4: +2 version checks (0.9.6's diff save)
def test_recording_a_leaver_does_not_grow_with_the_people_already_there(seeded, api):
def leave(arg):
if arg == "prepare":
return _hire(api, "2026-10-01")["person_id"]
s, out = api("POST", f"/people/{arg}/leave", HR, {"last_day": "2026-12-31", "reason": "resignation"})
assert s == 200, out
few, many = _measure(seeded, api, leave)
_show("record leaver", few, many)
assert many.statements <= few.statements + 2, (few.statements, many.statements)
assert few.statements <= 24, few.statements # was 31: the run is expanded from the committed person
def test_completing_a_run_step_does_not_grow_with_the_runs_already_there(seeded, api):
def complete(arg):
if arg == "prepare":
run_id = _hire(api)["run_id"]
s, run = api("GET", f"/runs/{run_id}", HR)
step = next(x for x in run["steps"] if x["kind"] == "task" and x["responsible_group"] == "it")
return run_id, step["step_no"]
s, out = api("POST", f"/runs/{arg[0]}/steps/{arg[1]}/complete", HR, {})
assert s == 200, out
few, many = _measure(seeded, api, complete)
_show("complete step", few, many)
assert many.statements <= few.statements + 2, (few.statements, many.statements)
assert few.statements <= 19, few.statements # was 22: no re-read when nothing followed
def test_a_steady_tick_does_not_grow_with_the_open_runs(seeded, api):
def tick(arg):
if arg == "prepare":
api("POST", "/run-due-steps", TICK, {}) # whatever was due is done: the next is steady
return None
s, out = api("POST", "/run-due-steps", TICK, {})
assert s == 200, out
few, many = _measure(seeded, api, tick)
_show("steady tick", few, many)
assert many.statements <= few.statements + 2, (few.statements, many.statements)
assert few.statements <= 7, few.statements # was 7
def test_headcount_is_one_read_per_table_whatever_the_number_of_people(seeded, api):
def report(arg):
if arg == "prepare":
return None
s, out = api("GET", "/reports/headcount", HR, params={"on": "2026-11-30"})
assert s == 200, out
few, many = _measure(seeded, api, report)
_show("headcount", few, many)
assert many.statements == few.statements, (few.statements, many.statements)
assert few.statements <= 10, few.statements # was 10
def test_the_onboarding_monitor_is_one_read_per_table_whatever_the_number_of_runs(seeded, api):
def report(arg):
if arg == "prepare":
return None
s, out = api("GET", "/reports/onboarding", HR)
assert s == 200, out
few, many = _measure(seeded, api, report)
_show("onboarding monitor", few, many)
assert many.statements == few.statements, (few.statements, many.statements)
assert few.statements <= 8, few.statements