dodil/erp-brick-payments
DODIL ERP brick: payments
tests/test_audit1.py
"""Audit 1 (2026-10-22): each PROVEN finding as the failing test its proof describes, then the fix that makes it pass."""
import base64
import json
from contextlib import contextmanager
from decimal import Decimal
from unittest import mock
import pytest
from conftest import (
APPROVER,
CASHIER,
ROLES,
allocate,
api,
bank,
must,
on_day,
receipt,
refused,
set_control,
settle,
)
from odoo_fake import FakeOdoo, FakePartners, load_ar
from sqlalchemy import select
from test_migration import run
from test_rules import camt
from brick_payments.application import use_cases
from brick_payments.domain import matching
from brick_payments.infrastructure import odoo
from brick_payments.infrastructure import tables as t
def ident(sub, email, permissions):
"""A caller with a sub AND an email, which `person()` never makes (it uses the email for both)."""
payload = base64.urlsafe_b64encode(json.dumps({"permissions": list(permissions)}).encode()).decode()
user = {"sub": sub, **({"email": email} if email else {})}
return {"X-Dodil-User": json.dumps(user), "X-Dodil-User-JWT": f"h.{payload.rstrip('=')}.s"}
@contextmanager
def snapshot_before_the_other_commit():
"""DataK3 reads a snapshot: a transaction that began before another committed does not see its rows, and detects a
conflict only on the SAME key. Emulated by making every allocation read come back empty (the committed state when the
snapshot was taken), while the writes land for real."""
with mock.patch.object(t.AllocationRepository, "_where", lambda self, cond, limit=None: []):
yield
def rows(table):
with use_cases.repository() as repo:
return [dict(m) for m in repo.conn.execute(select(table)).mappings()]
# ── 1. a refund cannot be approved by its requester under another identity ──────────────────────
def test_1_the_requester_cannot_approve_their_own_refund_with_a_token_that_has_only_a_sub(nb):
receipt(amount="300.00", reference="c")
both = [*ROLES["cashier"], "pay:refund:approve", "pay:refund:request"]
ann_by_email = ident("u-ann", "ann@shop.example", both)
ann_by_sub = ident("u-ann", None, both)
rid = must(api("POST", "/refunds", ann_by_email, {"bp_id": 7311, "amount": "100.00", "currency": "GBP"}))["refund_id"]
refused(api("POST", f"/refunds/{rid}/approve", ann_by_sub, {}), 409, "You requested this refund")
# and the other way round: requested with a sub only, approved with sub + email
rid2 = must(api("POST", "/refunds", ann_by_sub, {"bp_id": 7311, "amount": "100.00", "currency": "GBP"}))["refund_id"]
refused(api("POST", f"/refunds/{rid2}/approve", ann_by_email, {}), 409, "You requested this refund")
assert must(api("POST", f"/refunds/{rid}/approve", APPROVER, {}))["status"] == "approved" # a second person still can
# ── 2. two writers cannot both spend the same money ─────────────────────────────────────────────
def test_2_two_cashiers_cannot_both_allocate_to_one_invoice(nb):
nb.invoice("INV-1", 7311, "4000.00")
r1, r2 = receipt(amount="4000.00", reference="a"), receipt(amount="4000.00", reference="b")
must(allocate(r1["payment_id"], [("INV-1", "4000.00")]))
with snapshot_before_the_other_commit():
refused(allocate(r2["payment_id"], [("INV-1", "4000.00")]), 409, "Someone else wrote")
assert settle("INV-1")["INV-1"]["outstanding"] == "0.00"
def test_2_one_receipt_cannot_be_spent_twice_over_two_invoices(nb):
nb.invoice("INV-1", 7311, "4000.00")
nb.invoice("INV-2", 7311, "4000.00")
r = receipt(amount="4000.00", reference="a")
must(allocate(r["payment_id"], [("INV-1", "4000.00")]))
with snapshot_before_the_other_commit():
refused(allocate(r["payment_id"], [("INV-2", "4000.00")]), 409, "Someone else wrote")
assert settle("INV-2")["INV-2"]["outstanding"] == "4000.00"
def test_2_a_refund_and_an_allocation_cannot_both_spend_one_credit(nb):
nb.invoice("INV-1", 7311, "300.00")
r = receipt(amount="300.00", reference="c")
rid = must(api("POST", "/refunds", CASHIER, {"bp_id": 7311, "amount": "300.00", "currency": "GBP"}))["refund_id"]
must(api("POST", f"/refunds/{rid}/approve", APPROVER, {}))
must(allocate(r["payment_id"], [("INV-1", "300.00")])) # the other writer got there first
with snapshot_before_the_other_commit():
refused(api("POST", f"/refunds/{rid}/pay", CASHIER, {"bank_ref": "B"}), 409, "Someone else wrote")
def test_2_the_claims_do_not_get_in_the_way_of_an_undo_or_a_second_allocation(nb):
nb.invoice("INV-1", 7311, "1000.00")
nb.invoice("INV-2", 7311, "1000.00")
r = receipt(amount="1500.00", reference="a")
pid = r["payment_id"]
must(allocate(pid, [("INV-1", "1000.00")]))
must(api("POST", f"/receipts/{pid}/unallocate", CASHIER, {"invoice_id": "INV-1", "seq": 1, "reason": "wrong invoice"}))
assert settle("INV-1")["INV-1"]["outstanding"] == "1000.00"
must(allocate(pid, [("INV-2", "1000.00")]))
must(allocate(pid, [("INV-1", "500.00")]))
assert (settle("INV-1")["INV-1"]["outstanding"], settle("INV-2")["INV-2"]["outstanding"]) == ("500.00", "0.00")
got = must(api("POST", f"/receipts/{pid}/reverse", CASHIER, {"bank_return_ref": "RET-1"}))
assert got["status"] == "reversed"
assert settle("INV-1")["INV-1"]["outstanding"] == "1000.00"
# ── 3. a statement already sent does not change when a back-dated receipt is recorded ───────────
def test_3_a_back_dated_receipt_does_not_change_a_past_statement(nb):
nb.invoice("INV-1", 7311, "1000.00", issued_on="2026-11-01")
sent = must(api("GET", "/customers/7311/open-items", CASHIER, {"as_at": "2026-11-19"}))
assert sent["closing_balance"] == {"GBP": "1000.00"}
receipt(amount="300.00", reference="late", received_on="2026-11-18") # recorded on 11-20 (today)
again = must(api("GET", "/customers/7311/open-items", CASHIER, {"as_at": "2026-11-19"}))
assert again["closing_balance"] == {"GBP": "1000.00"} and again["credits"] == []
assert must(api("GET", "/customers/7311/open-items", CASHIER, {"as_at": "2026-11-20"}))["closing_balance"] == {"GBP": "700.00"}
# ── 4. a change of time zone never dates anything into the past ─────────────────────────────────
def test_4_a_new_time_zone_takes_effect_from_the_next_business_date(nb):
nb.invoice("INV-1", 7311, "100.00")
pid = receipt(amount="100.00", reference="a")["payment_id"]
set_control("business_timezone", "Pacific/Pago_Pago") # London 2026-11-20 09:00; Pago_Pago is 2026-11-19 22:00
got = must(allocate(pid, [("INV-1", "10.00")]))
assert got["allocations"][0]["allocated_on"] == "2026-11-20"
with on_day("2026-11-21T09:00:00+00:00"): # London 11-21: the new zone is in force
got = must(allocate(pid, [("INV-1", "10.00")]))
assert got["allocations"][0]["allocated_on"] == "2026-11-20" # Pago_Pago's date, never before the last one used
def test_4_today_never_goes_below_the_last_business_date_already_used(nb):
nb.invoice("INV-1", 7311, "100.00")
pid = receipt(amount="100.00", reference="a")["payment_id"]
set_control("business_timezone", "Pacific/Kiritimati") # UTC+14, from London's 11-21
with on_day("2026-11-21T01:00:00+00:00"): # Kiritimati 11-21 15:00
assert must(allocate(pid, [("INV-1", "10.00")]))["allocations"][0]["allocated_on"] == "2026-11-21"
set_control("business_timezone", "Pacific/Pago_Pago") # from Kiritimati's 11-22
with on_day("2026-11-21T10:30:00+00:00"): # Kiritimati 11-22 00:30; Pago_Pago is 11-20 23:30
assert must(allocate(pid, [("INV-1", "10.00")]))["allocations"][0]["allocated_on"] == "2026-11-21"
# ── 5. a statement file with an impossible amount is refused line by line, never a 500 ──────────
def test_5_a_csv_with_a_non_finite_or_huge_or_too_precise_amount_is_refused_with_its_row(nb):
bank()
for amount in ("Infinity", "-Infinity", "NaN", "1e30", "1000000000000000.00", "1.005"):
csv = f"booking_date,amount,currency,entry_ref\n2026-11-19,10.00,GBP,OK-1\n2026-11-19,{amount},GBP,BAD-1\n"
body = refused(api("POST", "/receipts/import", CASHIER, {"format": "csv", "content": csv, "bank_account_id": "BANK1"}), 409, "Row 3")
assert "BAD" not in json.dumps(body) or "Row 3" in json.dumps(body)
assert must(api("GET", "/receipts", CASHIER))["receipts"] == [] # nothing of a refused file landed
def test_5_a_camt_with_a_non_finite_or_huge_amount_is_refused_and_a_csv_bp_id_is_bounded(nb):
bank()
for amount in ("Infinity", "NaN", "1E+30", "12.345"):
refused(api("POST", "/receipts/import", CASHIER, {"format": "camt053", "bank_account_id": "BANK1",
"content": camt([("E1", "CRDT", amount, "GBP", "2026-11-19", "x")])}), 409, "E1")
csv = "booking_date,amount,currency,entry_ref,bp_id\n2026-11-19,10.00,GBP,E2,99999999999999999999999\n"
refused(api("POST", "/receipts/import", CASHIER, {"format": "csv", "content": csv, "bank_account_id": "BANK1"}), 409, "Row 2")
# ── 6. an exchange rate is bounded ──────────────────────────────────────────────────────────────
def test_6_an_fx_rate_must_be_finite_positive_and_bounded(nb):
body = {"bp_id": 7311, "amount": "5.00", "currency": "EUR", "received_on": "2026-11-20"}
for bad in ("1e30", "Infinity", "NaN", "0", "-1.5", "1.23456", "1" + "0" * 15, 1.5):
assert api("POST", "/receipts", CASHIER, {**body, "fx_rate": bad})[0] == 422, bad
got = must(api("POST", "/receipts", CASHIER, {**body, "fx_rate": "1.2345"}))
assert got["fx_rate"] == "1.2345"
# ── 7. only the whole invoice number is an EXACT match ──────────────────────────────────────────
def test_7_a_year_or_a_po_number_is_a_suggestion_never_an_exact_match():
inv = matching.Open("A", "INV-2026-0042", Decimal("500.00"))
for reference in ("PO 42 shipped", "paid 0042", "INV42", "order 2026-0042"):
got = matching.suggest(amount=Decimal("500.00"), remainder=Decimal("500.00"), reference=reference, open_invoices=[inv])
assert got is not None and got.exact is False, reference
for reference in ("INV-2026-0042", "inv 2026 0042 thanks", "INV20260042"):
got = matching.suggest(amount=Decimal("500.00"), remainder=Decimal("500.00"), reference=reference, open_invoices=[inv])
assert got is not None and got.exact is True, reference
def test_7_auto_with_review_does_not_allocate_on_a_year(nb):
nb.invoice("INV-2026-0042", 7311, "500.00", number="INV-2026-0042")
set_control("ai_suggest_match", "auto-with-review")
got = receipt(amount="500.00", reference="PO 42 refund of 2026")
assert got["suggestion_result"]["suggested"] is True and got["suggestion_result"].get("auto_allocated") is None
assert got["allocated"] == "0.00"
exact = receipt(amount="500.00", reference="INV-2026-0042")
assert exact["allocated"] == "500.00"
# ── 8. a transfer typed without a bank account is still recognised when the bank statement brings it ─
def test_8_a_typed_receipt_without_an_account_blocks_the_same_reference_on_import_and_back(nb):
bank()
receipt(amount="100.00", reference="TRF-77") # typed: no bank account
got = must(api("POST", "/receipts/import", CASHIER, {"format": "camt053", "bank_account_id": "BANK1",
"content": camt([("TRF-77", "CRDT", "100.00", "GBP", "2026-11-19", "x")])}))
assert (got["created"], got["skipped_duplicate"]) == (0, 1)
must(api("POST", "/receipts/import", CASHIER, {"format": "camt053", "bank_account_id": "BANK1",
"content": camt([("TRF-88", "CRDT", "50.00", "GBP", "2026-11-19", "x")])}))
refused(api("POST", "/receipts", CASHIER, {"bp_id": 7311, "amount": "50.00", "currency": "GBP", "received_on": "2026-11-19",
"bank_reference": "TRF-88"}), 409, "already on receipt")
# ── 9. a second migration run leaves the first run's record, and the cut-over, alone ────────────
@pytest.fixture
def odoo_fakes(nb):
load_ar(nb)
odoo.use(odoo.Ports(FakeOdoo(), FakePartners()))
yield
odoo.use(None)
def test_9_a_second_run_keeps_the_first_record_and_the_cutover(odoo_fakes):
must(run())
first = [r for r in rows(t.pay_migration_run) if r["run_id"] == "odoo"][0]
set_control("statement_import_from", "2026-11-18") # the owner moved it
with on_day("2026-11-25T09:00:00+00:00"):
again = must(run())
assert again["already_loaded"] >= 15
refused(run(cutover_date="2026-11-10"), 409, "cut-over")
after = {r["run_id"]: r for r in rows(t.pay_migration_run)}
assert {k: after["odoo"][k] for k in ("counts", "ran_at", "ran_by", "params", "odoo_figures")} == {
k: first[k] for k in ("counts", "ran_at", "ran_by", "params", "odoo_figures")}
assert len(after) == 2 and any(k.startswith("odoo#") for k in after) # the re-run is its own record
settings = {s["setting_key"]: s["value"] for s in must(api("GET", "/settings", CASHIER))["settings"]}
assert settings["statement_import_from"] == "2026-11-18"