dodil/erp-brick-expenses
DODIL ERP brick: expenses
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; the business date; money bounded (422, never 500); no row of one person changes another's answer."""
import pytest
from conftest import (
EMPLOYEE,
FINANCE,
HR_CLIENT,
MANAGER,
PAYMENTS_CLIENT,
SYSTEM,
add,
api,
approve,
claim_551,
must,
new_report,
refused,
report,
submit,
)
from sqlalchemy import func, select
from brick_expenses.application import use_cases
from brick_expenses.infrastructure import tables as t
def claims(scope=None):
with use_cases.repository() as repo:
q = select(t.exp_claim.c.claim_scope, t.exp_claim.c.claim_key, t.exp_claim.c.n)
if scope:
q = q.where(t.exp_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 money / person operation ──────────────────────────────────────────────────
def test_every_money_or_person_operation_takes_exactly_one_claim(nb):
rid = claim_551()
submit(rid)
must(approve(rid, MANAGER))
must(approve(rid, FINANCE))
must(api("POST", f"/reports/{rid}/paid", PAYMENTS_CLIENT, {"payout_ref": f"exp:{rid}:r1:pay", "payment_id": "p1", "paid_on": "2026-11-13", "amount": "551.40"}))
must(api("POST", "/employees/3301/leaver", HR_CLIENT, {"last_day": "2026-11-06", "source_ref": "hr:left:3301:1"}))
assert claims() == sorted([("series", "UK01-2026", 1), ("submit", f"{rid}:r1", 1), ("approval", f"{rid}:r1:1", 1), ("approval", f"{rid}:r1:2", 1),
("complete", f"{rid}:r1", 1), ("paid", f"exp:{rid}:r1:pay", 1), ("leaver", "hr:left:3301:1", 1)])
def test_a_lost_race_on_a_claim_key_is_a_409_and_writes_nothing_more(nb):
rid = claim_551()
submit(rid)
with use_cases.repository() as repo: # the other approver got there first: THEIR claim row is committed
repo.claims.once("approval", f"{rid}:r1:1", use_cases._now(), "someone.else")
refused(approve(rid, MANAGER), 409, "Someone else did the same")
assert count(t.exp_approval_step) == 0 and report(rid)["waiting_level"] == 1
def test_the_same_report_number_is_never_issued_twice(nb):
with use_cases.repository() as repo:
repo.claims.take("series", "UK01-2026", use_cases._now(), "racer") # another request took number 1 a moment ago
assert new_report()["report_id"] == "UK01-EXP-2026-00002"
ids = {new_report()["report_id"] for _ in range(5)}
assert len(ids) == 5
# ── side effects only after the local commit, from pending rows ──────────────────────────────────
def test_a_rolled_back_approval_sends_nothing_and_a_crash_after_commit_is_sent_on_the_next_drain(nb, monkeypatch):
rid = claim_551()
submit(rid)
must(approve(rid, MANAGER))
real = use_cases.rp.journal_lines
monkeypatch.setattr(use_cases.rp, "journal_lines", lambda *a, **k: (_ for _ in ()).throw(use_cases.RuleBroken("the journal cannot be built")))
refused(approve(rid, FINANCE), 409, "cannot be built") # the completion transaction rolled back as ONE unit ...
assert nb.journals == {} and nb.journal_posts == 0 and count(t.exp_outbox) == 0 # ... so nothing was sent and no row waits
got = report(rid, FINANCE)
assert got["status"] == "submitted" and got["waiting_level"] is None and len(got["decisions"]) == 2 # the level itself committed
monkeypatch.setattr(use_cases.rp, "journal_lines", real)
out = must(api("POST", "/outbox/drain", SYSTEM)) # the worker finds the stuck claim, completes it, posts, requests the payout
assert out["completed"] == 1 and [s["status"] for s in out["sent"]] == ["posted", "requested"]
assert report(rid)["status"] == "posted" and len(nb.journals) == 1 and len(nb.payouts) == 1
def test_the_journal_and_the_payout_are_written_as_pending_rows_in_the_transaction_of_their_state_change(nb):
rid = claim_551()
submit(rid)
nb.gl_status = 500
must(approve(rid, MANAGER))
must(approve(rid, FINANCE))
with use_cases.repository() as repo:
(row,) = repo.outbox.pending()
assert (row["target"], row["ref"], row["status"]) == ("gl", f"exp:{rid}:r1:approve", "pending")
assert nb.payouts == {} and count(t.exp_outbox) == 1 # the payout row is born only when the journal is posted
nb.gl_status = None
must(api("POST", "/outbox/drain", SYSTEM))
with use_cases.repository() as repo:
assert repo.outbox.pending() == [] and sorted((r["target"], r["status"]) for r in repo.outbox.of_report(rid)) == [("gl", "sent"), ("payments", "sent")]
def test_a_payout_and_a_journal_never_share_a_key(nb):
rid = claim_551()
submit(rid)
must(approve(rid, MANAGER))
must(approve(rid, FINANCE))
with use_cases.repository() as repo:
refs = {r["target"]: r["ref"] for r in repo.outbox.of_report(rid)}
assert refs == {"gl": f"exp:{rid}:r1:approve", "payments": f"exp:{rid}:r1:pay"}
(j,) = nb.journals.values()
assert j["source_document"] == f"exp:{rid}:r1" and next(iter(nb.payouts)) == f"exp:{rid}:r1:pay"
# ── states never go backwards ───────────────────────────────────────────────────────────────────
def test_a_submitted_claim_cannot_go_back_to_draft_and_a_posted_one_cannot_be_decided_again(nb):
rid = claim_551()
submit(rid)
refused(api("POST", f"/reports/{rid}/submit", EMPLOYEE), 409, "is submitted")
refused(api("POST", f"/reports/{rid}/cancel", EMPLOYEE), 409, "is submitted")
refused(api("POST", f"/reports/{rid}/lines", EMPLOYEE, {"category_code": "OFFICE", "expense_date": "2026-11-04", "merchant": "x", "amount": "1"}), 409, "frozen")
refused(api("DELETE", f"/reports/{rid}/lines/1", EMPLOYEE), 409, "frozen")
must(approve(rid, MANAGER))
must(approve(rid, FINANCE))
for who in (MANAGER, FINANCE):
refused(approve(rid, who), 409, "is posted")
refused(api("POST", f"/reports/{rid}/return", who, {"reason": "x"}), 409, "is posted")
refused(api("POST", f"/reports/{rid}/reject", who, {"reason": "x"}), 409, "is posted")
refused(api("POST", f"/reports/{rid}/submit", EMPLOYEE), 409, "is posted")
assert report(rid)["status"] == "posted"
def test_paid_is_terminal_and_a_cancelled_or_rejected_claim_stays_so(nb):
rid = claim_551()
submit(rid)
must(approve(rid, MANAGER))
must(approve(rid, FINANCE))
pay = {"payout_ref": f"exp:{rid}:r1:pay", "payment_id": "p1", "paid_on": "2026-11-13", "amount": "551.40"}
must(api("POST", f"/reports/{rid}/paid", PAYMENTS_CLIENT, pay))
refused(api("POST", f"/reports/{rid}/paid", PAYMENTS_CLIENT, {**pay, "amount": "1.00"}), 409) # not paid twice, not repaid another amount
assert must(api("POST", f"/reports/{rid}/post", FINANCE))["status"] == "paid" # a repeat of post changes nothing
d = new_report()["report_id"]
must(api("POST", f"/reports/{d}/cancel", EMPLOYEE))
refused(api("POST", f"/reports/{d}/cancel", EMPLOYEE), 409, "is cancelled")
refused(api("POST", f"/reports/{d}/submit", EMPLOYEE), 409, "is cancelled")
def test_a_returned_revision_is_final_and_its_decisions_stay_with_it(nb):
rid = claim_551()
submit(rid)
must(api("POST", f"/reports/{rid}/return", MANAGER, {"reason": "why?"}))
assert report(rid)["status"] == "draft" and report(rid)["revision"] == 2
refused(approve(rid, EMPLOYEE), 409, "is draft") # nobody approves a draft, and rev 1's return cannot be reopened
refused(approve(rid, MANAGER), 404)
with use_cases.repository() as repo:
assert repo.reports.revisions(rid) == [(1, "returned")]
# ── money bounded: 422, never 500 ──────────────────────────────────────────────────────────────────
@pytest.mark.parametrize("amount", ["0", "-5.00", "10.001", "1e20", "abc", "NaN", "Infinity", "10000000000000.00", "99999999999999999999", "", " "])
def test_a_bad_amount_on_a_line_is_a_422_never_a_500(nb, amount):
rid = new_report()["report_id"]
got, body = api("POST", f"/reports/{rid}/lines", EMPLOYEE, {"category_code": "OFFICE", "expense_date": "2026-11-04", "merchant": "x", "amount": amount})
assert got == 422, (amount, got, body)
got, _ = api("POST", f"/reports/{rid}/lines", EMPLOYEE, {"category_code": "OFFICE", "expense_date": "2026-11-04", "merchant": "x", "amount": "1.00",
"vat_amount": amount if amount.strip() else "-1"})
assert got != 500
assert len(report(rid)["lines"]) == (1 if amount == "0" else 0) # a zero VAT is a valid VAT; nothing else got in
def test_the_largest_lines_still_total_and_post_within_numeric_18_2(nb):
rid = new_report()["report_id"]
for _ in range(3):
add(rid, amount="9999999999999.99")
got = submit(rid)
assert got["total_amount"] == "29999999999999.97"
must(approve(rid, MANAGER))
assert must(approve(rid, FINANCE))["status"] == "posted"
def test_a_number_a_float_or_a_boolean_where_an_amount_goes_is_refused_not_rounded(nb):
rid = new_report()["report_id"]
for amount in (12.5, True, None, [1], {"a": 1}):
got, _ = api("POST", f"/reports/{rid}/lines", EMPLOYEE, {"category_code": "OFFICE", "expense_date": "2026-11-04", "merchant": "x", "amount": amount})
assert got == 422, amount
def test_dates_ids_and_text_on_every_door_are_bounded(nb):
rid = new_report()["report_id"]
base = {"category_code": "OFFICE", "merchant": "x", "amount": "1.00"}
for when in ("tomorrow", "2026-13-45", "1066-10-14", "9999-12-31", "2026-11-08", "x" * 100, ""):
got, _ = api("POST", f"/reports/{rid}/lines", EMPLOYEE, {**base, "expense_date": when})
assert got in (409, 422), when # 2026-11-08 is tomorrow: a business rule (409); the rest are malformed
for body in ({"currency": "pounds"}, {"currency": "12"}, {"currency": ""}, {"currency": "GÉP"}):
assert api("POST", "/reports", EMPLOYEE, body)[0] == 422, body
assert api("POST", "/reports", EMPLOYEE, {"currency": "GBP", "title": "t" * 10_000})[0] == 422
assert api("DELETE", f"/reports/{rid}/lines/99999999999999999999999", EMPLOYEE)[0] in (404, 422)
assert api("POST", f"/reports/{rid}/lines", EMPLOYEE, {**base, "expense_date": "2026-11-04", "receipt_key": "raw/expenses/OTHER/1/x.jpg"})[0] == 422
def test_the_paid_route_bounds_what_payments_sends(nb):
rid = claim_551()
submit(rid)
must(approve(rid, MANAGER))
must(approve(rid, FINANCE))
body = {"payout_ref": f"exp:{rid}:r1:pay", "payment_id": "p1", "paid_on": "2026-11-13", "amount": "551.40"}
for bad in ({"amount": "1e30"}, {"amount": "-1"}, {"amount": "abc"}, {"paid_on": "soon"}, {"paid_on": "0001-01-01"}):
got, out = api("POST", f"/reports/{rid}/paid", PAYMENTS_CLIENT, {**body, **bad})
assert got == 422 and out["gate"] == "bounds", (bad, got, out)
assert report(rid)["status"] == "posted"
def test_a_claim_holds_at_most_two_hundred_lines(nb):
rid = new_report()["report_id"]
with use_cases.repository() as repo:
from datetime import date
from decimal import Decimal
from brick_expenses.domain import report as rp
repo.reports.add_lines(rid, 1, [rp.Line(n, "OFFICE", date(2026, 11, 4), "x", Decimal("1.00")) for n in range(1, 201)], use_cases._now())
refused(api("POST", f"/reports/{rid}/lines", EMPLOYEE, {"category_code": "OFFICE", "expense_date": "2026-11-04", "merchant": "x", "amount": "1"}), 409, "at most 200")
# ── one person's rows never change another's answer ─────────────────────────────────────────────────
def test_an_employees_list_holds_only_their_own_claims_and_a_waiting_list_only_what_waits_for_them(nb):
mine = claim_551()
submit(mine)
nb.employee(5001, "other@uk.example", manager=nb.as_manager(2280, "mgr2280@uk.example"))
from conftest import person
other = person("other@uk.example", ["exp:report:read_own", "exp:report:write_own", "exp:report:submit"])
theirs = new_report(other)["report_id"]
add(theirs, other, amount="20.00")
submit(theirs, other)
assert [r["report_id"] for r in must(api("GET", "/reports", EMPLOYEE))["reports"]] == [mine]
assert [r["report_id"] for r in must(api("GET", "/reports", other))["reports"]] == [theirs]
assert sorted(r["report_id"] for r in must(api("GET", "/reports", MANAGER, {"scope": "waiting"}))["reports"]) == sorted([mine, theirs])
refused(api("GET", f"/reports/{theirs}", EMPLOYEE), 404)
refused(api("GET", "/reports", EMPLOYEE, {"scope": "all"}), 403, "exp:report:read_all")
assert len(must(api("GET", "/reports", FINANCE, {"scope": "all"}))["reports"]) == 2