dodil/erp-brick-timesheets
DODIL ERP brick: timesheets
tests/test_invariants.py
"""The recurring bug classes, each tried to break: one claim key per operation; side effects only after the local commit, from pending rows;
states never move backwards; nobody decides their own week; the business date; hours and money bounded (422, never 500); differences, not totals."""
from datetime import date
from decimal import Decimal
import pytest
from conftest import (
APPROVAL_DAY,
EMPLOYEE,
FINANCE,
FINANCE2,
MANAGER,
OWNER,
SHEET,
SYSTEM,
api,
approve,
approved_week,
hours,
line,
must,
on_day,
open_week,
refused,
sheet,
story_week,
submit,
)
from sqlalchemy import func, select
from brick_timesheets.application import use_cases
from brick_timesheets.infrastructure import tables as t
def claims(scope=None):
with use_cases.repository() as repo:
q = select(t.ts_claim.c.claim_scope, t.ts_claim.c.claim_key, t.ts_claim.c.n)
if scope:
q = q.where(t.ts_claim.c.claim_scope == scope)
return sorted(tuple(r) for r in repo.conn.execute(q))
def count(table):
with use_cases.repository() as repo:
return repo.conn.execute(select(func.count()).select_from(table)).scalar()
# ── one claim key per operation on a week revision ────────────────────────────────────────────────
def test_every_operation_on_a_week_takes_exactly_one_claim(nb):
approved_week()
must(api("POST", f"/sheets/{SHEET}/reopen", FINANCE, {"reason": "fix"}))
submit()
must(api("POST", f"/sheets/{SHEET}/return", MANAGER, {"reason": "again"}))
assert claims() == sorted([("open", f"{SHEET}:r1", 1), ("submit", f"{SHEET}:r1", 1), ("approval", f"{SHEET}:r1", 1), ("reopen", f"{SHEET}:r1", 1),
("submit", f"{SHEET}:r2", 1), ("close", f"{SHEET}:r2", 1)])
@pytest.mark.parametrize("scope,key,call", [
("open", f"{SHEET}:r1", lambda: api("POST", "/sheets", EMPLOYEE, {"week": "2026-W47"})),
("submit", f"{SHEET}:r1", lambda: api("POST", f"/sheets/{SHEET}/submit", EMPLOYEE)),
("approval", f"{SHEET}:r1", lambda: api("POST", f"/sheets/{SHEET}/approve", MANAGER)),
("close", f"{SHEET}:r1", lambda: api("POST", f"/sheets/{SHEET}/return", MANAGER, {"reason": "x"})),
])
def test_a_lost_race_on_a_claim_key_is_a_409_and_writes_nothing_more(nb, scope, key, call):
if scope != "open":
open_week()
story_week()
if scope in ("approval", "close"):
submit()
before = (count(t.ts_outbox), count(t.ts_billing_claim), sheet()["status"] if scope != "open" else None)
with use_cases.repository() as repo: # the other request got there first: THEIR claim row is committed
repo.claims.once(scope, key, use_cases._now(), "someone.else")
with on_day(APPROVAL_DAY):
refused(call(), 409, "Someone else did the same")
assert (count(t.ts_outbox), count(t.ts_billing_claim), sheet()["status"] if scope != "open" else None) == before
if scope == "open":
assert count(t.ts_sheet) == 0 # no sheet was written
def test_a_journal_a_delivery_and_a_billing_claim_are_each_claimed_once(nb):
approved_week()
assert count(t.ts_billing_claim) == 1 and sorted(r["ref"] for r in _outbox()) == ["ts:4410:2026-W47:r1:2", "ts:4410:2026-W47:r1:approve"]
with use_cases.repository() as repo: # a second writer of the same delivery claim is refused by the key itself
with pytest.raises(Exception, match="UNIQUE|duplicate"):
repo.billing.add_many([{"sheet_id": SHEET, "revision": 1, "line_no": 2, "purpose": "delivery", "delivery_ref": "x", "quantity": Decimal(1)}],
use_cases._now())
def _outbox():
with use_cases.repository() as repo:
return repo.outbox.of_sheet(SHEET)
# ── side effects only after the local commit, from pending rows ──────────────────────────────────
def test_a_rolled_back_approval_sends_nothing_and_writes_nothing(nb, monkeypatch):
open_week()
story_week()
submit()
monkeypatch.setattr(use_cases.sh, "journal_lines", lambda *a, **k: (_ for _ in ()).throw(use_cases.RuleBroken("the journal cannot be built")))
refused(api("POST", f"/sheets/{SHEET}/approve", MANAGER), 409, "cannot be built")
assert nb.journals == {} and nb.journal_posts == 0 and nb.delivery_posts == 0 # nothing was sent ...
assert count(t.ts_outbox) == 0 and count(t.ts_billing_claim) == 0 and claims("approval") == [] # ... nothing waits, and the claim is free again
assert sheet()["status"] == "submitted"
def test_a_crash_after_the_commit_is_sent_by_the_next_drain_from_the_pending_rows(nb, monkeypatch):
open_week()
story_week()
submit()
with monkeypatch.context() as dies: # (undo() would also drop the engine fixture's repository)
dies.setattr(use_cases, "_drain", lambda *a, **k: []) # the process dies right after the commit
assert approve()[0] == 200
assert nb.journal_posts == 0 and nb.delivery_posts == 0
assert sorted(r["status"] for r in _outbox()) == ["pending", "pending"] # the rows were committed WITH the approval
out = must(api("POST", "/outbox/drain", SYSTEM))["sent"] + must(api("POST", "/deliveries/report", SYSTEM))["sent"]
assert sorted(x["status"] for x in out) == ["sent", "sent"] and len(nb.journals) == 1 and len(nb.deliveries) == 1
assert must(api("POST", "/outbox/drain", SYSTEM))["sent"] == [] and must(api("POST", "/deliveries/report", SYSTEM))["sent"] == []
def test_a_neighbour_is_never_called_by_a_use_case_that_only_reads_or_edits(nb):
open_week()
story_week()
submit()
sheet()
sheet(who=FINANCE)
assert (nb.journal_posts, nb.delivery_posts) == (0, 0)
# ── states never move backwards ──────────────────────────────────────────────────────────────
def test_no_backward_state_is_reachable_through_a_door(nb):
approved_week()
for method, path, body in [("POST", f"/sheets/{SHEET}/submit", None), ("POST", f"/sheets/{SHEET}/approve", None),
("POST", f"/sheets/{SHEET}/return", {"reason": "x"}), ("POST", f"/sheets/{SHEET}/lines", {"activity_code": "DEV", "cost_centre": "CONS-LON"}),
("POST", f"/sheets/{SHEET}/lines/2/entries", {"entries": [{"work_date": "2026-11-16", "hours": "1"}]}),
("DELETE", f"/sheets/{SHEET}/lines/2", None), ("POST", f"/sheets/{SHEET}/refresh", None)]:
who = EMPLOYEE if "approve" not in path and "return" not in path else MANAGER
code, _ = api(method, path, who, body)
assert code in (404, 409), (path, code) # an approved week never goes back to submitted or open
assert sheet()["status"] == "approved" and sheet()["revision"] == 1
must(api("POST", f"/sheets/{SHEET}/reopen", FINANCE, {"reason": "fix"}))
for method, path, who, body in [("POST", f"/sheets/{SHEET}/reopen", FINANCE2, {"reason": "again"}), ("POST", f"/sheets/{SHEET}/approve", MANAGER, None),
("POST", f"/sheets/{SHEET}/return", MANAGER, {"reason": "x"})]:
assert api(method, path, who, body)[0] in (404, 409) # a reopened (open) revision 2 is not approved, returned or reopened again
assert [r["status"] for r in sheet()["revisions"]] == ["superseded", "open"]
def test_a_returned_revision_never_counts_for_the_next_and_its_approvals_stay_with_it(nb):
with use_cases.repository() as repo:
repo.settings.put("approval_route", "project_lead_then_manager", date(2026, 1, 1), "owner", use_cases._now())
from conftest import LEADER
open_week()
story_week()
submit()
must(api("POST", f"/sheets/{SHEET}/approve-project-lines", LEADER))
must(api("POST", f"/sheets/{SHEET}/return", MANAGER, {"reason": "again"}))
submit()
refused(api("POST", f"/sheets/{SHEET}/approve", MANAGER), 409, "wait for their project lead") # revision 1's line approvals do not count for 2
assert count(t.ts_line_approval) == 2 and sheet()["line_approvals"] == []
# ── the business date ───────────────────────────────────────────────────────────────────────
def test_the_business_date_is_the_date_in_business_timezone_never_the_utc_date(nb):
open_week()
story_week()
submit()
approve(at="2026-11-23T23:30:00+00:00") # 23:30 in London (GMT): the 23rd
assert sheet()["approved_on"] == "2026-11-23"
(j,) = nb.journals.values()
assert j["journal_date"] == "2026-11-23" and sheet(who=FINANCE)["outbox"][0]["status"] == "sent"
def test_a_summer_approval_at_00_30_bst_posts_on_the_new_day_and_a_zone_row_moves_the_date_for_everyone(nb):
wk = "2026-W28" # Monday 2026-07-06
open_week(week=wk)
n = line("4410:2026-W28")["lines"][-1]["line_no"]
hours("4410:2026-W28", n, {f"2026-07-{d}": "7.50" for d in ("06", "07", "08", "09", "10")})
submit("4410:2026-W28")
approve("4410:2026-W28", at="2026-07-13T23:30:00+00:00") # 00:30 BST on 2026-07-14
(j,) = nb.journals.values()
assert j["journal_date"] == "2026-07-14"
with use_cases.repository() as repo: # business_timezone is a dated row
repo.settings.put("business_timezone", "Pacific/Auckland", date(2026, 1, 1), "owner", use_cases._now())
open_week(week="2026-W29")
n = line("4410:2026-W29")["lines"][-1]["line_no"]
hours("4410:2026-W29", n, {f"2026-07-{d}": "7.50" for d in ("13", "14", "15", "16", "17")})
submit("4410:2026-W29")
approve("4410:2026-W29", at="2026-07-20T23:30:00+00:00") # already the 21st in Auckland
assert sheet("4410:2026-W29")["approved_on"] == "2026-07-21"
def test_a_zone_row_dated_tomorrow_is_not_yet_in_force(nb):
with use_cases.repository() as repo:
repo.settings.put("business_timezone", "Pacific/Auckland", date(2026, 11, 21), "owner", use_cases._now())
assert repo.settings.timezone() == "Europe/London" # not before the business date has reached it
# ── hours and money are bounded: 422 or 4xx, never a 500 ─────────────────────────────────────────
def test_what_a_week_cannot_hold_is_a_422_never_a_500(nb):
open_week()
n = line(SHEET)["lines"][-1]["line_no"]
bad_week = [("POST", "/sheets", {"week": w}) for w in ("2026-W54", "2026-47", "", "x" * 200, "9999-W01", "0001-W01")]
bad_entries = [("POST", f"/sheets/{SHEET}/lines/{n}/entries", {"entries": e}) for e in
([], [{"work_date": "2026-11-16", "hours": h} for h in ("99999999999999999999", "-1", "1e400")][:1],
[{"work_date": "2026-11-16", "hours": "1"}] * 8, [{"work_date": "2026-11-16", "hours": "1", "note": "n" * 501}],
[{"work_date": "2026-11-16"}], [{"work_date": "2999-01-01", "hours": "1"}], [{"work_date": "2026-11-16", "hours": "1.5.5"}])]
bad_lines = [("POST", f"/sheets/{SHEET}/lines", b) for b in
({"activity_code": "DEV"}, {"activity_code": "DEV", "project_id": "P", "cost_centre": "C"}, {"activity_code": "x" * 65, "cost_centre": "C"},
{"activity_code": "DEV", "project_id": "p" * 141}, {"activity_code": "DEV", "cost_centre": "CONS-LON", "billable": "maybe"})]
bad_rates = [("POST", "/cost-rates", {"company_code": "UK01", "rate_group": "S", "rate_amount": a, "currency": c, "valid_from": d})
for a, c, d in (("1e30", "GBP", "2027-01-01"), ("5", "GB", "2027-01-01"), ("5", "GBP", "27-01-01"), ("5", "GBP", "1999-01-01"), ("5", "£££", "2027-01-01"),
("5", "GBP", "2101-01-01"), ("0.00001", "GBP", "2027-01-01"))]
bad_ids = [("POST", "/person-rate-groups", {"person_id": p, "rate_group": "S", "valid_from": "2027-01-01"}) for p in ("x", "-1", "99999999999999999999999", "")]
bad_reads = [("GET", "/approved-hours", {"company_code": "UK01", "from": a, "to": b}) for a, b in (("2026-01-01", "2027-06-01"), ("x", "y"), ("2026-11-16", "2026-11-01"))]
for method, path, body in bad_week + bad_entries + bad_lines + bad_ids:
code, got = api(method, path, EMPLOYEE if "rate" not in path else FINANCE, body)
assert code in (409, 422) and "Traceback" not in str(got), (path, body, code, got)
for method, path, body in bad_rates + bad_ids[:0]:
assert api(method, path, FINANCE, body)[0] == 422, (body,)
for method, path, body in bad_reads:
from conftest import PAYROLL
assert api(method, path, PAYROLL, body)[0] == 422, body
for reason in ("", " ", "r" * 501):
assert api("POST", f"/sheets/{SHEET}/return", MANAGER, {"reason": reason})[0] in (404, 422)
assert api("POST", f"/sheets/{SHEET}/reopen", FINANCE, {"reason": reason})[0] in (404, 422)
assert api("POST", "/controls", OWNER, {"key": "k" * 65, "value": "v", "valid_from": "2026-12-01"})[0] in (409, 422)
def test_the_largest_week_the_bounds_allow_posts_without_overflow(nb):
with use_cases.repository() as repo:
repo.rates.put_rate("UK01", "SENIOR", Decimal("9999999.9999"), "GBP", date(2026, 11, 1), "fin", use_cases._now())
open_week()
for i in range(3):
n = line(SHEET)["lines"][-1]["line_no"]
hours(SHEET, n, {f"2026-11-{d}": ("24", "long day") for d in ("16", "17", "18", "19")} if i == 0 else {"2026-11-21": "1"}) # Friday is leave
submit()
approve()
(j,) = nb.journals.values()
total = sum(Decimal(x["debit"]) for x in j["lines"] if "debit" in x)
assert total == (Decimal("9999999.9999") * 98).quantize(Decimal("0.01")) and total < Decimal("1e16") # inside NUMERIC(18,2)
# ── differences, not totals ──────────────────────────────────────────────────────────────────
def net(nb):
out: dict = {}
for j in nb.journals.values():
for x in j["lines"]:
out[x["cost_centre"]] = out.get(x["cost_centre"], Decimal(0)) + Decimal(x.get("debit", "0")) - Decimal(x.get("credit", "0"))
return out
def test_after_any_sequence_of_revisions_cost_and_delivery_equal_the_latest_approved_revision(nb):
approved_week()
delivered = lambda: nb.orders_book["quote:Q-2026-0042@v2"][2]["delivered"] # noqa: E731
for wed in ("2.00", "2.00", "4.00", "0.50"): # a correction, a no-change, back again, and less
must(api("POST", f"/sheets/{SHEET}/reopen", FINANCE, {"reason": "fix"}))
other = Decimal("7.50") - Decimal("3.50") - Decimal(wed)
hours(SHEET, 2, {"2026-11-18": wed})
hours(SHEET, 4, {"2026-11-18": f"{other:.2f}", "2026-11-19": "7.50"}) # zero clears the day
submit()
assert approve()[0] == 200
p1 = Decimal("15.00") + Decimal(wed)
assert delivered() == Decimal(40) + p1 # the order holds the latest revision's hours
got = net(nb)
assert got["DELIV-1"] == (p1 * 52).quantize(Decimal("0.01")) # the ledger holds the latest revision's cost
assert got["DELIV-2"] == ((Decimal("7.50") + other) * 52).quantize(Decimal("0.01")) and got["CONS-LON"] == -(got["DELIV-1"] + got["DELIV-2"])
assert sheet()["revision"] == 5 and count(t.ts_billing_claim) == 4 # revisions 1, 2, 4 and 5 moved the order; revision 3 changed nothing for the order: no claim, no call
def test_a_revision_with_no_change_for_the_order_tells_it_nothing(nb):
approved_week()
must(api("POST", f"/sheets/{SHEET}/reopen", FINANCE, {"reason": "typo in a note"}))
submit()
approve()
assert nb.delivery_posts == 1 and list(nb.deliveries) == ["ts:4410:2026-W47:r1:2"] # the same hours: nothing sent
assert len(nb.journals) == 3 # the cost is reversed and reposted all the same