dodil/erp-brick-timesheets
DODIL ERP brick: timesheets
tests/test_statement_budget.py
"""Statement budgets: every statement is a round trip (100-300 ms on the live pg wire), so a request's statements must not grow with the number of
entries on a week or with the unrelated sheets around it. Counted at the engine, one request at a time (erp_platform.testing.count_statements)."""
from datetime import date
import pytest
from conftest import EMPLOYEE, EMPLOYEE2, FINANCE, MANAGER, P1, PAYROLL, SYSTEM, api, approve, hours, must, open_week, submit
from erp_platform.testing import count_statements
from brick_timesheets.application import use_cases
from brick_timesheets.domain import sheet as sh
SHEET2 = "4425:2026-W47"
@pytest.fixture
def counted(engine):
return lambda: count_statements(engine)
@pytest.fixture(autouse=True)
def warn_mode():
"""missing_hours_mode warn: a week of 2 entries and a week of 20 can both be submitted."""
with use_cases.repository() as repo:
repo.settings.put("missing_hours_mode", "warn", date(2026, 1, 1), "owner", use_cases._now())
def week_of(n, who, sheet_id):
"""A week with exactly n entries: one billable P1 line first (a delivery at approval), then internal lines, at most 7 days each."""
open_week(who)
placed, first = 0, True
days = ["16", "17", "18", "19", "20", "21", "22"]
while placed < n:
k = min(7, n - placed)
body = {"activity_code": "DEV", "project_id": P1} if first else {"activity_code": "ADMIN", "cost_centre": "CONS-LON"}
ln = must(api("POST", f"/sheets/{sheet_id}/lines", who, body), 201)["lines"][-1]["line_no"]
hours(sheet_id, ln, {f"2026-11-{d}": "1.00" for d in days[:k]}, who)
placed += k
first = False
return sheet_id
def _show(label, c):
print(f"BUDGET {label}: statements={c.statements} round_trips={c.round_trips}")
def test_reading_a_week_does_not_grow_with_its_entries(nb, counted):
two, twenty = week_of(2, EMPLOYEE, "4410:2026-W47"), week_of(20, EMPLOYEE2, SHEET2)
with counted() as few:
must(api("GET", f"/sheets/{two}", EMPLOYEE))
with counted() as many:
must(api("GET", f"/sheets/{twenty}", EMPLOYEE2))
_show("get (2 entries)", few)
_show("get (20 entries)", many)
assert many.statements == few.statements and few.statements <= 7, (few.statements, many.statements)
def test_booking_hours_does_not_grow_with_the_entries_already_there(nb, counted):
two, twenty = week_of(2, EMPLOYEE, "4410:2026-W47"), week_of(20, EMPLOYEE2, SHEET2)
with counted() as few:
hours(two, 2, {"2026-11-22": "1.00"}, EMPLOYEE)
with counted() as many:
hours(twenty, 2, {"2026-11-22": "2.00"}, EMPLOYEE2)
_show("book (2 entries there)", few)
_show("book (20 entries there)", many)
assert many.statements == few.statements and few.statements <= 9, (few.statements, many.statements)
def test_submitting_does_not_grow_with_the_entries(nb, counted):
two, twenty = week_of(2, EMPLOYEE, "4410:2026-W47"), week_of(20, EMPLOYEE2, SHEET2)
with counted() as few:
submit(two, EMPLOYEE)
with counted() as many:
submit(twenty, EMPLOYEE2)
_show("submit (2 entries)", few)
_show("submit (20 entries)", many)
assert many.statements == few.statements and few.statements <= 22, (few.statements, many.statements)
def test_approving_and_sending_the_journal_and_the_delivery_does_not_grow_with_the_entries(nb, counted):
a, b = week_of(2, EMPLOYEE, "4410:2026-W47"), week_of(20, EMPLOYEE2, SHEET2)
submit(a, EMPLOYEE)
submit(b, EMPLOYEE2)
with counted() as few:
assert approve(a, MANAGER)[0] == 200
with counted() as many:
assert approve(b, MANAGER)[0] == 200
_show("approve+journal+delivery (2 entries)", few)
_show("approve+journal+delivery (20 entries)", many)
assert many.statements == few.statements and few.statements <= 30, (few.statements, many.statements)
def seed_unrelated(repo, n, company="DE01"):
"""n sheets (lines, entries, approved) of other people in another company, written straight to the tables."""
at = use_cases._now()
for k in range(n):
s = sh.Sheet(f"{9000 + k}:2026-W47", 9000 + k, company, date(2026, 11, 16), sh.Identity(f"p{k}@de.example", f"p{k}"), status="approved",
approver=sh.Identity("boss@de.example", "boss"), approved_at=at, approved_on=date(2026, 11, 23), approved_by="boss@de.example",
expected={"2026-11-16": 480}, last_line_no=1)
s.lines = [sh.Line(1, "project", project_id="X", cost_centre="C", billable=True, entries=[sh.Entry(date(2026, 11, 16), sh.hours4(8))])]
repo.sheets.create(s, at)
def test_lists_and_the_approved_hours_do_not_grow_with_unrelated_sheets_and_do_not_change(nb, counted, engine):
mine = week_of(2, EMPLOYEE, "4410:2026-W47")
submit(mine, EMPLOYEE)
approve(mine, MANAGER)
ask = lambda: (api("GET", "/sheets", EMPLOYEE), api("GET", "/sheets", MANAGER, {"scope": "waiting"}), api("GET", "/sheets", FINANCE, {"scope": "all"}), # noqa: E731
api("GET", "/approved-hours", PAYROLL, {"company_code": "UK01", "from": "2026-11-16", "to": "2026-11-22"}))
with use_cases.repository() as repo:
seed_unrelated(repo, 2)
with counted() as few:
before = ask()
with use_cases.repository() as repo:
for k in range(2, 50):
seed_unrelated_one(repo, k)
with counted() as many:
after = ask()
_show("lists + approved-hours (2 unrelated)", few)
_show("lists + approved-hours (50 unrelated)", many)
assert many.statements == few.statements and few.statements <= 8, (few.statements, many.statements)
assert before[0] == after[0] and before[1] == after[1] and before[3][1]["rows"] == after[3][1]["rows"] # the person's, the waiting and the company's answers are unchanged
assert len(after[2][1]["timesheets"]) == len(before[2][1]["timesheets"]) + 48 # finance's `all` is by definition every sheet
def seed_unrelated_one(repo, k):
at = use_cases._now()
s = sh.Sheet(f"{9000 + k}:2026-W47", 9000 + k, "DE01", date(2026, 11, 16), sh.Identity(f"p{k}@de.example", f"p{k}"), status="approved",
approver=sh.Identity("boss@de.example", "boss"), approved_at=at, approved_on=date(2026, 11, 23), approved_by="boss@de.example",
expected={"2026-11-16": 480}, last_line_no=1)
s.lines = [sh.Line(1, "project", project_id="X", cost_centre="C", billable=True, entries=[sh.Entry(date(2026, 11, 16), sh.hours4(8))])]
repo.sheets.create(s, at)
def test_approving_and_reading_a_week_do_not_grow_with_unrelated_sheets(nb, counted):
a, b = week_of(2, EMPLOYEE, "4410:2026-W47"), week_of(2, EMPLOYEE2, SHEET2)
submit(a, EMPLOYEE)
submit(b, EMPLOYEE2)
with use_cases.repository() as repo:
for k in range(2):
seed_unrelated_one(repo, k)
with counted() as few:
assert approve(a, MANAGER)[0] == 200
must(api("GET", f"/sheets/{a}", EMPLOYEE))
with use_cases.repository() as repo:
for k in range(2, 50):
seed_unrelated_one(repo, k)
with counted() as many:
assert approve(b, MANAGER)[0] == 200
must(api("GET", f"/sheets/{b}", EMPLOYEE2))
_show("approve+get (2 unrelated)", few)
_show("approve+get (50 unrelated)", many)
assert many.statements == few.statements, (few.statements, many.statements)
assert must(api("POST", "/outbox/drain", SYSTEM))["sent"] == []
def test_a_billing_run_and_its_release_do_not_grow_with_other_projects_sheets(nb, counted, engine):
from conftest import FINANCE2, approved_week, on_day
approved_week()
at = use_cases._now()
def other_projects(repo, lo, hi):
for k in range(lo, hi):
s = sh.Sheet(f"{9000 + k}:2026-W47", 9000 + k, "UK01", date(2026, 11, 16), sh.Identity(f"p{k}@uk.example", f"p{k}"), status="approved",
approver=sh.Identity("boss@uk.example", "boss"), approved_at=at, approved_on=date(2026, 11, 23), approved_by="boss@uk.example",
expected={"2026-11-16": 480}, last_line_no=1)
s.lines = [sh.Line(1, "project", project_id=f"pj:other:{k}", cost_centre="C", billable=True, bill_via="ar", product_id="CONSULT-SR",
entries=[sh.Entry(date(2026, 11, 16), sh.hours4(8))])]
repo.sheets.create(s, at)
with use_cases.repository() as repo:
other_projects(repo, 0, 2)
with on_day("2026-11-30T10:00:00+00:00"):
with counted() as few:
run = must(api("POST", "/billing-runs", SYSTEM, {"project_id": "pj:contract:CT-2026-0009", "period_id": "2026-11"}))["run"]["run_id"]
with use_cases.repository() as repo:
other_projects(repo, 2, 50)
with counted() as many:
again = api("POST", "/billing-runs", SYSTEM, {"project_id": "pj:contract:CT-2026-0009", "period_id": "2026-11"})
with counted() as rel:
must(api("POST", f"/billing-runs/{run}/release", FINANCE2))
with counted() as get:
must(api("GET", f"/billing-runs/{run}", FINANCE))
_show("billing run create (2 unrelated)", few)
_show("billing run again (50 unrelated)", many)
_show("billing run release", rel)
assert again[0] == 200 and many.statements < few.statements and few.statements <= 22 and rel.statements <= 22 and get.statements <= 4, (few.statements, many.statements, rel.statements, get.statements)
def test_project_hours_do_not_grow_with_other_projects_sheets_and_do_not_change(nb, counted, engine):
from conftest import LEADER, approved_week
approved_week()
ask = lambda: api("GET", "/project-hours", LEADER, {"project_id": "pj:contract:CT-2026-0009", "from": "2026-11-01", "to": "2026-11-30"}) # noqa: E731
with use_cases.repository() as repo:
seed_unrelated(repo, 2, "UK01")
with counted() as few:
before = ask()
with use_cases.repository() as repo:
for k in range(2, 50):
seed_unrelated_one(repo, k)
with counted() as many:
after = ask()
_show("project hours (2 unrelated)", few)
_show("project hours (50 unrelated)", many)
assert many.statements == few.statements and few.statements <= 8 and before == after and before[0] == 200, (few.statements, many.statements)