dodil/erp-brick-ar
DODIL ERP brick: accounts receivable
tests/test_audit1.py
"""Audit 1: each test is the proof of one finding, written before its fix."""
import base64
import json
from decimal import Decimal
from conftest import (
APPROVER,
BILLER,
BP,
ROLES,
SYSTEM,
api,
billing,
control,
hand_draft,
issue,
issued,
line,
must,
on_day,
)
def ident(sub, email, perms):
"""A caller whose sub and email differ (a real token): the permissions of the given roles."""
allowed = sorted({p for r in perms for p in ROLES[r]})
payload = base64.urlsafe_b64encode(json.dumps({"permissions": allowed}).encode()).decode()
user = {"sub": sub, "app_roles": []}
if email:
user["email"] = email
return {"X-Dodil-User": json.dumps(user), "X-Dodil-User-JWT": f"h.{payload.rstrip('=')}.s"}
# ── 1, 2: self-approval under either identity ────────────────────────────────────────────────
def test_1_a_writeoff_proposer_cannot_approve_it_under_their_sub(world):
must(control("second_approver", True))
must(control("writeoff_threshold", 25))
inv = issued(world, lines=[line("Small", "1", "10.33")])
dana = ident("u-42", "dana@x.example", ["biller", "approver"])
same_person_no_email = ident("u-42", None, ["approver"])
w = must(api("POST", "/writeoffs", dana, {"invoice_id": inv["invoice_id"], "amount": "12.40", "reason": "closed"}))
s, b = api("POST", f"/writeoffs/{w['writeoff_id']}/approve", same_person_no_email, {})
assert s == 409 and "second_person" in b["detail"]
other = ident("u-43", "eve@x.example", ["approver"])
assert must(api("POST", f"/writeoffs/{w['writeoff_id']}/approve", other, {}))["status"] == "approved"
def test_2_a_credit_note_drafter_cannot_approve_it_under_their_sub(world):
must(control("second_approver", True))
inv = issued(world)
dana = ident("u-42", "dana@x.example", ["biller", "approver"])
same_person_no_email = ident("u-42", None, ["approver"])
d = must(api("POST", "/credit-notes", dana, {"invoice_id": inv["invoice_id"], "reason": "r", "full": True}))
world.payments.credit_faces[d["credit_id"]] = Decimal(d["total"])
s, b = api("POST", f"/credit-notes/{d['credit_id']}/approve", same_person_no_email, {})
assert s == 409 and "second_person" in b["detail"]
other = ident("u-43", "eve@x.example", ["approver"])
assert must(api("POST", f"/credit-notes/{d['credit_id']}/approve", other, {}))["status"] == "approved"
# ── 3: a credit note against a void invoice ──────────────────────────────────────────────────
def test_3_a_credit_note_cannot_be_approved_against_a_void_invoice(world):
inv = issued(world)
d = must(api("POST", "/credit-notes", BILLER, {"invoice_id": inv["invoice_id"], "reason": "r", "full": True}))
world.payments.credit_faces[d["credit_id"]] = Decimal(d["total"])
s, b = api("POST", f"/invoices/{inv['invoice_id']}/void", APPROVER, {"reason": "oops"})
assert s == 409 and "draft credit note" in b["detail"] # void refuses while a draft names it
must(api("POST", f"/credit-notes/{d['credit_id']}/cancel", BILLER, {}))
must(api("POST", f"/invoices/{inv['invoice_id']}/void", APPROVER, {"reason": "oops"}))
def test_3_approve_credit_requires_an_issued_invoice(engine, world):
"""The approve path itself: a void invoice (however it got there) refuses the credit."""
from brick_ar.infrastructure.tables import ar_invoices
inv = issued(world)
d = must(api("POST", "/credit-notes", BILLER, {"invoice_id": inv["invoice_id"], "reason": "r", "full": True}))
with engine.begin() as c:
c.execute(ar_invoices.update().where(ar_invoices.c.invoice_id == inv["invoice_id"]).values(lifecycle="void"))
s, b = api("POST", f"/credit-notes/{d['credit_id']}/approve", APPROVER, {})
assert s == 409 and "issued" in b["detail"]
# ── 4: the time zone in force is chosen by the business date ─────────────────────────────────
def test_4_the_zone_in_force_is_chosen_by_the_business_date_not_utc(world):
must(control("business_timezone", "Pacific/Pago_Pago"))
must(control("business_timezone", "Pacific/Kiritimati", valid_from="2026-11-21"))
from brick_ar.application import use_cases
with on_day("2026-11-21", hour=5): # 05:00Z: 11-20 18:00 in Pago_Pago; Kiritimati (+14) would say 11-21
with use_cases.repository() as repo:
assert str(use_cases._today(repo)) == "2026-11-20"
with on_day("2026-11-21", hour=11): # 11-21 00:00 Pago_Pago: the new row takes over, Kiritimati 11-22 01:00
with use_cases.repository() as repo:
assert str(use_cases._today(repo)) == "2026-11-22"
# ── 5: numbers never collide after a cut-over ────────────────────────────────────────────────
from conftest import ( # noqa: E402
migrate_payments, # noqa: E402
)
from test_migration import RUN, explain_and_sign, load, reconcile # noqa: E402
def _signed_odoo(odoo, world, **doors):
must(load(**({"doors": doors} if doors else {})))
migrate_payments(world, odoo)
reconcile()
explain_and_sign(world)
def test_5_a_yyyy_format_cannot_be_combined_with_numbers_that_do_not_restart(odoo, world):
s, b = control("number_restarts_yearly", False) # the default format has {yyyy}
assert s == 409 and "{yyyy}" in b["detail"]
must(control("invoice_number_format", "INV/{seq:5}"))
assert control("number_restarts_yearly", False)[0] == 200
s, b = control("invoice_number_format", "INV/{yyyy}/{seq:5}", valid_from="2026-11-21")
assert s == 409 and "{yyyy}" in b["detail"]
s, b = control("invoice_number_format", "INV", valid_from="2026-11-21")
assert s == 422 and "{seq" in b["detail"]
def test_5_no_two_documents_share_a_number_migrated_ones_included(odoo, world, engine):
"""New-prefix door: Odoo's numbers do not raise the mark, and the owner's format renders the very same text."""
_signed_odoo(odoo, world, numbering="new_prefix")
must(control("invoice_number_format", "INV/{yyyy}/{seq:5}"))
migrated = {"INV/2026/00001", "INV/2026/00002", "INV/2026/00004", "INV/2026/00005"}
got = []
from conftest import on_day
with on_day("2026-11-21"):
billing()
for _ in range(4):
got.append(issue(hand_draft()["invoice_id"], world=world)["invoice_no"])
assert not (set(got) & migrated) and len(set(got)) == 4
from sqlalchemy import select
from brick_ar.infrastructure.tables import ar_invoice_no_claims, ar_invoices
with engine.begin() as c:
numbers = [r[0] for r in c.execute(select(ar_invoices.c.invoice_no).where(ar_invoices.c.invoice_no != ""))]
claimed = {r[0] for r in c.execute(select(ar_invoice_no_claims.c.invoice_no))}
assert len(numbers) == len(set(numbers)) and set(numbers) <= claimed
def test_5_the_mark_is_raised_on_the_series_the_format_resolves_to(odoo, world, engine):
must(control("invoice_number_format", "INV/{yyyy}/{seq:5}"))
must(load(doors={"numbering": "new_prefix"}))
from sqlalchemy import select
from brick_ar.infrastructure.tables import ar_number_series
with engine.begin() as c:
marks = dict(c.execute(select(ar_number_series.c.series_id, ar_number_series.c.high_water_mark)).all())
assert marks.get("INV/2026") == 5
# ── 6: signed history is never rewritten by a later run ──────────────────────────────────────
def test_6_a_later_run_never_touches_a_document_of_a_signed_run(odoo, world):
_signed_odoo(odoo, world)
odoo.move_rows[0]["ref"] = "PO-CHANGED" # Odoo changed move 100 after the signature
out = must(load(cutover_date="2026-11-05"))
assert out["run_id"] == "odoo:odooprod:2026-11-05" and out["counts"]["differs"] == ["odoo:odooprod:100"]
assert out["counts"]["updated"] == 0
inv = must(api("GET", "/invoices/odoo:odooprod:100", BILLER))
assert inv["customer_po_ref"] == "PO-100" and inv["migration_run_id"] == RUN and inv["lifecycle"] == "issued"
again = must(load(cutover_date="2026-11-05")) # and it stays that way
assert again["counts"]["differs"] == ["odoo:odooprod:100"]
def test_6_a_reload_that_changes_nothing_leaves_a_reconciled_run_reconciled(odoo, world):
must(load())
migrate_payments(world, odoo)
reconcile()
assert must(load())["status"] == "reconciled"
odoo.move_rows[0]["ref"] = "PO-CHANGED"
changed = must(load())
assert changed["status"] == "loaded" and changed["counts"]["updated"] == 1
# ── 7: concurrent credit notes on one invoice ────────────────────────────────────────────────
def test_7_two_snapshot_transactions_cannot_together_credit_more_than_the_invoice(world, engine, monkeypatch):
"""Each approval reads 'nothing credited yet' (its snapshot), as two overlapping transactions would; the number series is
not what stops the second: the per-invoice claim is."""
from brick_ar.infrastructure import tables
inv = issued(world)
ids = []
for _ in range(2):
d = must(api("POST", "/credit-notes", BILLER, {"invoice_id": inv["invoice_id"], "reason": "r", "full": True}))
world.payments.credit_faces[d["credit_id"]] = Decimal(d["total"])
ids.append(d["credit_id"])
must(api("POST", f"/credit-notes/{ids[0]}/approve", APPROVER, {}))
real_total, real_count = tables.CreditNoteRepository.credited_total, tables.CreditNoteRepository.committed_count
stale = {"left": 1}
def snapshot_total(self, invoice_id):
return Decimal(0) if stale["left"] else real_total(self, invoice_id)
def snapshot_count(self, invoice_id):
if stale["left"]:
stale["left"] -= 1
return 0
return real_count(self, invoice_id)
monkeypatch.setattr(tables.CreditNoteRepository, "credited_total", snapshot_total)
monkeypatch.setattr(tables.CreditNoteRepository, "committed_count", snapshot_count)
s, b = api("POST", f"/credit-notes/{ids[1]}/approve", APPROVER, {})
assert s == 409 and "within_invoice" in b["detail"]
monkeypatch.setattr(tables.CreditNoteRepository, "credited_total", real_total)
monkeypatch.setattr(tables.CreditNoteRepository, "committed_count", real_count)
statuses = [must(api("GET", f"/credit-notes/{i}", APPROVER))["status"] for i in ids]
assert statuses == ["approved", "draft"]
# ── 8: raise_due pages through every schedule ────────────────────────────────────────────────
def test_8_raise_due_pages_through_all_schedules(world):
with on_day("2026-10-31"):
billing()
for i in range(5):
must(api("POST", "/schedules", SYSTEM, {"source_kind": "contract", "source_id": f"ct:C-{i}", "line_id": "L1", "bp_id": BP,
"starts_on": "2026-10-01", "periods": 12, "unit_price": "2.50", "currency": "GBP",
"billing_kind": "recurring", "billing_plan": "monthly", "in_advance": False,
"description": "Seat", "unit_code": "C62", "tax_category": "standard"}))
out = must(api("POST", "/schedules/raise-due", SYSTEM, {"limit": 2}))
assert len(out["raised"]) == 5 and not out["failed"]
# ── 9: a foreign-currency invoice needs its rate ─────────────────────────────────────────────
def test_9_raise_takes_a_bounded_rate_and_issue_needs_one_in_a_foreign_currency(world):
billing()
body = {"source_kind": "contract", "source_id": "ct:FX-1", "bp_id": BP, "currency": "EUR", "lines": [line(qty="1", price="1000.00")]}
d = must(api("POST", "/invoices/raise", SYSTEM, body))
r = must(api("GET", f"/invoices/{d['invoice_id']}/readiness", BILLER))
gate = next(g for g in r["gates"] if g["gate"] == "fx_rate")
assert gate["met"] is False and gate["reason"] and gate["fix"]
s, b = api("POST", f"/invoices/{d['invoice_id']}/issue", BILLER, {})
assert s == 409 and "fx_rate" in b["detail"]
ok = must(api("POST", "/invoices/raise", SYSTEM, {**body, "source_id": "ct:FX-2", "fx_rate": "0.8600"}))
assert ok["fx_rate"] == "0.8600" and ok["total_functional"] == "1032.00"
assert must(api("POST", f"/invoices/{ok['invoice_id']}/issue", BILLER, {}))["total_functional"] == "1032.00"
for bad in ("0", "-1", "0.86001", "1000000000000000", "nan"):
s, _b = api("POST", "/invoices/raise", SYSTEM, {**body, "source_id": f"ct:FX-{bad}", "fx_rate": bad})
assert s == 422, bad
home = must(api("POST", "/invoices/raise", SYSTEM, {**body, "source_id": "ct:FX-3", "currency": "GBP"})) # no rate needed at home
assert must(api("POST", f"/invoices/{home['invoice_id']}/issue", BILLER, {}))["lifecycle"] == "issued"