dodil/erp-brick-expenses
DODIL ERP brick: expenses
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 lines on a claim or of unrelated rows. Counted at the engine, one request at a time (erp_platform.testing.count_statements)."""
import pytest
from conftest import EMPLOYEE, MANAGER, add, api, approve, must, new_report, submit
from erp_platform.testing import count_statements
@pytest.fixture
def counted(engine):
return lambda: count_statements(engine)
CATEGORIES = ["TRAVEL_RAIL", "TRAVEL_AIR", "ACCOMMODATION", "MEALS_CLIENT", "MEALS_STAFF", "OFFICE"] # many (category, date) pairs on a big claim
def claim_of(n, who=None):
rid = new_report(who)["report_id"]
for k in range(n):
add(rid, who, merchant=f"Merchant {k}", amount="10.00", receipt_key="", category_code=CATEGORIES[k % 6], expense_date=f"2026-11-0{1 + k % 4}")
return rid
def _show(label, c):
print(f"BUDGET {label}: statements={c.statements} round_trips={c.round_trips}")
def test_reading_a_claim_does_not_grow_with_its_lines(nb, counted):
two, twenty = claim_of(2), claim_of(20)
with counted() as few:
must(api("GET", f"/reports/{two}", EMPLOYEE))
with counted() as many:
must(api("GET", f"/reports/{twenty}", EMPLOYEE))
_show("get (2 lines)", few)
_show("get (20 lines)", many)
assert many.statements == few.statements and few.statements <= 8, (few.statements, many.statements)
def test_submitting_does_not_grow_with_the_lines(nb, counted):
two, twenty = claim_of(2), claim_of(20)
with counted() as few:
submit(two)
with counted() as many:
submit(twenty)
_show("submit (2 lines)", few)
_show("submit (20 lines)", many)
assert many.statements == few.statements and few.statements <= 13, (few.statements, many.statements)
def test_approving_completing_posting_and_asking_for_the_payout_does_not_grow_with_the_lines(nb, counted):
a, b = claim_of(2), claim_of(20) # 20.00 and 200.00: under the threshold, so the manager's is the last level
for rid in (a, b):
submit(rid)
with counted() as few:
must(approve(a, MANAGER))
with counted() as many:
must(approve(b, MANAGER))
_show("approve+complete+post+payout (2 lines)", few)
_show("approve+complete+post+payout (20 lines)", many)
assert many.statements == few.statements and few.statements <= 36, (few.statements, many.statements)
def test_adding_a_line_does_not_grow_with_the_lines_already_there(nb, counted):
two, twenty = claim_of(2), claim_of(20)
with counted() as few:
add(two, merchant="One more", amount="10.00", receipt_key="")
with counted() as many:
add(twenty, merchant="One more", amount="10.00", receipt_key="")
_show("add line (2 there)", few)
_show("add line (20 there)", many)
assert many.statements == few.statements and few.statements <= 8, (few.statements, many.statements)
def test_lists_do_not_grow_with_other_peoples_claims(nb, counted):
from conftest import person
for k in range(2):
nb.employee(6000 + k, f"p{k}@uk.example", manager=nb.as_manager(2280, "mgr2280@uk.example"))
def other(k):
return person(f"p{k}@uk.example", ["exp:report:read_own", "exp:report:write_own", "exp:report:submit"])
mine = claim_of(2)
submit(mine)
with counted() as few:
must(api("GET", "/reports", EMPLOYEE))
must(api("GET", "/reports", MANAGER, {"scope": "waiting"}))
for k in range(2):
for _ in range(5):
rid = claim_of(2, other(k))
submit(rid, other(k))
with counted() as many:
must(api("GET", "/reports", EMPLOYEE))
must(api("GET", "/reports", MANAGER, {"scope": "waiting"}))
_show("lists (1 claim around)", few)
_show("lists (11 claims around)", many)
assert many.statements == few.statements and few.statements <= 4, (few.statements, many.statements)
def test_submitting_does_not_grow_with_policy_rows_or_other_peoples_card_transactions(nb, counted):
from decimal import Decimal
from conftest import FINANCE, policy, seed_card
policy(FINANCE, rule_code="CAP_OFFICE", category_code="OFFICE", limit_amount="500.00")
a, b = claim_of(2), claim_of(2)
seed_card("1000", 2280, [(f"T{k}", "2026-11-05", "x", Decimal("5.00")) for k in range(2)])
with counted() as few:
submit(a)
for k in range(3):
policy(FINANCE, rule_code=f"RULE_{k}", category_code="OFFICE", limit_amount="500.00")
seed_card("2000", 1904, [(f"T{k}", "2026-11-05", "x", Decimal("5.00")) for k in range(40)])
with counted() as many:
submit(b)
_show("submit (2 unrelated txns, 1 rule)", few)
_show("submit (42 unrelated txns, 4 rules)", many)
assert many.statements == few.statements, (few.statements, many.statements)
def test_a_delegates_waiting_list_costs_one_more_query_however_many_claims_they_cover(nb, counted):
from conftest import MANAGER, person
cover = person("cover@uk.example", ["exp:report:read_own", "exp:report:read_team", "exp:report:approve", "exp:delegation:write"])
must(api("POST", "/delegations", MANAGER, {"delegate_email": "cover@uk.example", "valid_from": "2026-11-07", "valid_to": "2026-11-14"}), 201)
submit(claim_of(2))
with counted() as few:
must(api("GET", "/reports", cover, {"scope": "waiting"}))
for _ in range(6):
submit(claim_of(2))
with counted() as many:
must(api("GET", "/reports", cover, {"scope": "waiting"}))
assert many.statements == few.statements and few.statements <= 4, (few.statements, many.statements)
def test_importing_a_card_feed_and_listing_a_holders_transactions_do_not_grow_with_unrelated_cards(nb, counted):
from decimal import Decimal
from conftest import EMPLOYEE, FINANCE, seed_card
def content(card, n):
rows = [f"{card},T{k},2026-11-01,Shop,1.00,GBP" for k in range(n)]
return "\n".join(["card_id,ref,posted_on,merchant,amount,currency", *rows])
seed_card("4417", 3301, [])
seed_card("5000", 2280, [])
with counted() as few:
must(api("POST", "/card-feeds", FINANCE, {"statement_ref": "a", "total": "10.00", "content": content("4417", 10)}), 201)
with counted() as many:
must(api("POST", "/card-feeds", FINANCE, {"statement_ref": "b", "total": "400.00", "content": content("5000", 400)}), 201)
_show("import feed (10 txns)", few)
_show("import feed (400 txns)", many)
assert many.statements == few.statements, (few.statements, many.statements)
with counted() as mine_few:
must(api("GET", "/card-txns", EMPLOYEE))
seed_card("6000", 1904, [(f"U{k}", "2026-11-01", "x", Decimal("1.00")) for k in range(60)])
with counted() as mine_many:
must(api("GET", "/card-txns", EMPLOYEE))
assert mine_many.statements == mine_few.statements, (mine_few.statements, mine_many.statements)
def test_submitting_with_tax_and_foreign_lines_does_not_grow_with_the_lines(nb, counted):
from erp_platform import seam
seam.register_local("tax", nb.tax)
nb.fx_rates["EUR"] = [("2026-01-01", "0.870000")]
def foreign_claim(n):
rid = new_report()["report_id"]
for k in range(n):
add(rid, merchant=f"Merchant {k}", amount="10.00", currency="EUR", vat_amount="1.00", receipt_key="", category_code=CATEGORIES[k % 6],
expense_date=f"2026-11-0{1 + k % 4}")
return rid
two, twenty = foreign_claim(2), foreign_claim(20)
with counted() as few:
submit(two)
with counted() as many:
submit(twenty)
_show("submit with tax, EUR (2 lines)", few)
_show("submit with tax, EUR (20 lines)", many)
assert many.statements == few.statements, (few.statements, many.statements)
assert len(nb.tax_calls) == 22 # one HTTP call a receipt: the seam, not the engine, grows
def test_storing_a_receipt_does_not_grow_with_the_lines_or_other_claims_receipts(nb, counted):
import base64
from conftest import EMPLOYEE, JPEG
two, twenty = claim_of(2), claim_of(20)
def upload(rid, body):
return api("POST", f"/reports/{rid}/receipts", EMPLOYEE, {"content_type": "image/jpeg", "content_base64": base64.b64encode(body).decode(), "line_no": 1})
with counted() as few:
must(upload(two, JPEG + b"1"), 201)
for k in range(8): # other claims' receipts around (distinct images)
must(upload(claim_of(1), JPEG + bytes([k])), 201)
with counted() as many:
must(upload(twenty, JPEG + b"2"), 201)
_show("add receipt (2 lines)", few)
_show("add receipt (20 lines, 9 other receipts)", many)
assert many.statements == few.statements and few.statements <= 11, (few.statements, many.statements)
def test_adding_mileage_and_a_trip_do_not_grow_with_other_claims_or_other_peoples_lines(nb, counted):
from conftest import person
def mileage(rid):
return api("POST", f"/reports/{rid}/mileage", EMPLOYEE, {"expense_date": "2026-11-04", "miles": "10", "route": "x"})
def trip(rid, who, day):
return api("POST", f"/reports/{rid}/per-diem", who, {"trip_from": f"2026-11-{day}T07:00", "trip_to": f"2026-11-{day}T19:00", "country": "GB"})
nb.employee(7100, "p7100@uk.example", manager=nb.as_manager(2280, "mgr2280@uk.example"))
a, b = claim_of(2), claim_of(20)
with counted() as few:
must(mileage(a), 201)
for k in range(2): # other people's mileage around, and other claims of the same person
other = person("p7100@uk.example", ["exp:report:read_own", "exp:report:write_own", "exp:report:submit"])
rid = must(api("POST", "/reports", other, {"currency": "GBP"}), 201)["report_id"]
for _ in range(5):
must(api("POST", f"/reports/{rid}/mileage", other, {"expense_date": "2026-11-04", "miles": "10", "route": "x"}), 201)
with counted() as many:
must(mileage(b), 201)
_show("add mileage (2 lines)", few)
_show("add mileage (20 lines, 10 unrelated mileage lines)", many)
assert many.statements == few.statements and few.statements <= 11, (few.statements, many.statements)