dodil/erp-brick-payments
DODIL ERP brick: payments
tests/test_v4_statements.py
"""Payments v4, slice A - the statement, line by line: PAY-02 (changed), PAY-21 (reconcile), PAY-22 (the AI suggests), PAY-23 (rules).
Through the real app on SQLite, the neighbours in tests/fakes.py (ar, bp, gl, tax - each answer checked against its callee's spec)."""
from decimal import Decimal
from unittest import mock
from conftest import APPROVER, APPROVER2, CASHIER, SYSTEM, bank, must, on_day, person, receipt, refused
from conftest import api as _api
from sqlalchemy import select
from brick_payments.application import use_cases
from brick_payments.infrastructure import tables as t
def api(method, path, who, body=None):
return _api(method, path, who, body)
def camt(entries, stmt_id=None, opening=None, closing=None, account="GB00BANK"):
"""camt.053: entries are (ref, 'CRDT'|'DBIT', amount, ccy, booked_on, text, counterparty_name, counterparty_account)."""
body = "".join(
f"<Ntry><NtryRef>{r}</NtryRef><Amt Ccy='{c}'>{a}</Amt><CdtDbtInd>{d}</CdtDbtInd><BookgDt><Dt>{day}</Dt></BookgDt>"
f"<NtryDtls><TxDtls><RltdPties><Dbtr><Nm>{nm}</Nm></Dbtr><DbtrAcct><Id><IBAN>{ac}</IBAN></Id></DbtrAcct>"
f"<Cdtr><Nm>{nm}</Nm></Cdtr><CdtrAcct><Id><IBAN>{ac}</IBAN></Id></CdtrAcct></RltdPties><RmtInf><Ustrd>{u}</Ustrd></RmtInf></TxDtls></NtryDtls></Ntry>"
for r, d, a, c, day, u, nm, ac in entries)
bal = lambda code, v: (f"<Bal><Tp><CdOrPrtry><Cd>{code}</Cd></CdOrPrtry></Tp><Amt Ccy='GBP'>{v}</Amt><CdtDbtInd>CRDT</CdtDbtInd></Bal>" # noqa: E731
if v else "")
return ("<?xml version='1.0'?><Document xmlns='urn:iso:std:iso:20022:tech:xsd:camt.053.001.02'><BkToCstmrStmt><Stmt>"
f"{f'<Id>{stmt_id}</Id>' if stmt_id else ''}<Acct><Id><IBAN>{account}</IBAN></Id></Acct>{bal('OPBD', opening)}{bal('CLBD', closing)}{body}"
"</Stmt></BkToCstmrStmt></Document>")
def e(ref, d, amount, day="2026-11-18", text="", name="Acme Ltd", acct="GB00ACME12345678", ccy="GBP"):
return (ref, d, amount, ccy, day, text, name, acct)
def imp(content, fmt="camt053", who=None, account="BANK1"):
return must(api("POST", "/statements/import", who or CASHIER, {"format": fmt, "content": content, "bank_account_id": account}))
def lines(**kw):
return {x["line_id"]: x for x in must(api("GET", "/statement-lines", CASHIER, kw))["lines"]}
def table(tbl):
with use_cases.repository() as repo:
return [dict(m) for m in repo.conn.execute(select(tbl)).mappings()]
def two_banks():
bank("BANK1", gl="1200", account_number="12345678", purposes="receive,pay")
bank("DEP1", gl="1210", account_number="87654321", purposes="receive")
def vat_control():
must(api("POST", "/controls/vat_control_account_id", APPROVER, {"value": "2200"}))
# ── PAY-02 (v4): every line is kept; only a customer's credit is a receipt ───────────────────────────
def test_PAY_02_v4_every_line_is_kept_and_only_a_customers_credit_is_a_receipt(nb):
two_banks()
with on_day("2026-11-18T09:00:00+00:00"): # a rule starts tomorrow: saved on the 18th, in force on the 19th
must(api("POST", "/reconciliation-rules/INT", APPROVER, {"direction": "credit", "text_contains": "INTEREST", "account_id": "7000",
"tax_code": "E", "mode": "suggest"}))
content = camt([e("C1", "CRDT", "100.00", text="INV-1"), e("C2", "CRDT", "250.00", text="INV-2", name="Beta Ltd"),
e("T1", "CRDT", "20000.00", text="TRANSFER", name="Us", acct="GB11TEST87654321"),
e("I1", "CRDT", "3.21", text="CREDIT INTEREST", name="Bank"), e("D1", "DBIT", "12.50", text="SERVICE CHARGE", name="Bank")],
stmt_id="NOV-1", opening="1000.00", closing="21340.71")
got = imp(content)
assert (got["created"], got["own_transfers"], got["named_by_a_rule"], got["debits_kept"], got["lines_created"]) == (2, 1, 1, 1, 5)
ls = lines()
assert set(ls) == {f"bank:BANK1:{r}" for r in ("C1", "C2", "T1", "I1", "D1")} # every line, credit or debit
assert [ls[f"bank:BANK1:{r}"]["lifecycle"] for r in ("C1", "C2")] == ["matched", "matched"]
assert ls["bank:BANK1:C1"]["matched_kind"] == "receipt" and ls["bank:BANK1:C1"]["matched_ref"] == "bank:BANK1:C1"
assert [ls[f"bank:BANK1:{r}"]["lifecycle"] for r in ("T1", "I1", "D1")] == ["open", "open", "open"]
assert ls["bank:BANK1:D1"]["amount"] == "-12.50" and ls["bank:BANK1:D1"]["direction"] == "debit"
assert sorted(r["payment_id"] for r in must(api("GET", "/receipts", CASHIER))["receipts"]) == ["bank:BANK1:C1", "bank:BANK1:C2"]
st = table(t.pay_bank_statement)[0]
assert (st["statement_id"], str(st["opening_balance"]), str(st["closing_balance"]), st["lifecycle"]) == ("BANK1:NOV-1", "1000.00", "21340.71", "balanced")
# the rule (suggest mode) proposed itself on the interest line; the debit has no history, so nothing is suggested
assert [s["model"] for s in ls["bank:BANK1:I1"]["suggestions"]] == ["rule:INT"] and ls["bank:BANK1:D1"]["suggestions"] == []
def test_PAY_02_v4_importing_the_same_statement_twice_adds_nothing(nb):
two_banks()
content = camt([e("C1", "CRDT", "100.00"), e("D1", "DBIT", "12.50", name="Bank")], stmt_id="NOV-1")
imp(content)
again = imp(content)
assert (again["created"], again["lines_created"], again["skipped_duplicate"]) == (0, 0, 2)
assert len(table(t.pay_statement_line)) == 2 and len(table(t.pay_receipt)) == 1 and len(table(t.pay_bank_statement)) == 1
def test_PAY_02_v4_lines_before_the_cutover_are_skipped_entirely_and_a_later_line_of_the_same_file_is_not(nb):
two_banks()
must(api("POST", "/controls/statement_import_from", APPROVER, {"value": "2026-11-10"}))
got = imp(camt([e("OLD", "CRDT", "5.00", day="2026-11-09"), e("OLDD", "DBIT", "5.00", day="2026-11-09", name="Bank"),
e("NEW", "CRDT", "6.00", day="2026-11-12")]))
assert (got["skipped_before_cutover"], got["created"], got["lines_created"]) == (2, 1, 1)
assert set(lines()) == {"bank:BANK1:NEW"}
def test_PAY_02_v4_mt940_is_read_like_camt(nb):
two_banks()
mt = (":20:STMT1\n:25:20000012345678\n:28C:00042/001\n:60F:C261101GBP1000,00\n:61:2611021102C250,00NTRFNONREF//BNK123\n"
":86:/NAME/ACME LTD GB29NWBK60161331926819 INV-1\n:61:2611031103D12,50NCHGCHG1//BNK124\n:86:SERVICE CHARGE\n:62F:C261103GBP1237,50\n-}")
got = imp(mt, "mt940")
assert (got["created"], got["debits_kept"]) == (1, 1)
ls = lines()
assert ls["bank:BANK1:BNK123"]["booked_on"] == "2026-11-02" and ls["bank:BANK1:BNK124"]["amount"] == "-12.50"
st = table(t.pay_bank_statement)[0]
assert (st["statement_id"], str(st["opening_balance"]), str(st["closing_balance"])) == ("BANK1:00042-001", "1000.00", "1237.50")
assert imp(mt, "mt940")["lines_created"] == 0
def test_PAY_02_v4_a_credit_already_a_receipt_from_v2_gets_its_line_and_no_second_receipt(nb):
two_banks()
pid = receipt(amount="100.00", reference="C1", bank_account_id="BANK1")["payment_id"]
got = imp(camt([e("C1", "CRDT", "100.00")]))
assert (got["created"], got["skipped_duplicate"], got["lines_created"]) == (0, 1, 1)
assert lines()["bank:BANK1:C1"]["matched_ref"] == pid and len(table(t.pay_receipt)) == 1
# ── PAY-21: reconcile ────────────────────────────────────────────────────────────────────────────────
def test_PAY_21_the_hmrc_debit_is_matched_to_taxs_liability_and_posts_bank_against_vat_control(nb):
two_banks()
vat_control()
nb.liabilities = [{"period_key": "26AD", "charge_type": "vat", "outstanding_amount": "31350.00", "due_date": "2026-11-07"}]
imp(camt([e("H1", "DBIT", "31350.00", text="HMRC VAT", name="HMRC")]))
got = must(api("POST", "/statement-lines/bank:BANK1:H1/match", CASHIER, {"kind": "tax_liability", "ref": "26AD:vat"}))
assert got["lifecycle"] == "matched" and got["journal_id"].isdigit()
j = next(iter(nb.journals.values()))
assert (j["event"], j["source_document"], j["company_code"], j["source_module"]) == ("bank_line", "bank:BANK1:H1", "UK01", "payments")
assert [(ln.get("account_id"), ln.get("debit"), ln.get("credit")) for ln in j["lines"]] == [("2200", "31350.00", None), ("1200", None, "31350.00")]
refused(api("POST", "/statement-lines/bank:BANK1:H1/match", CASHIER, {"kind": "tax_liability", "ref": "26AD:vat"}), 409, "already matched")
def test_PAY_21_a_match_needs_the_same_amount_and_a_counterpart_belongs_to_one_line_only(nb):
two_banks()
pid = receipt(amount="100.00", reference="typed", bank_account_id="BANK1")["payment_id"]
imp(camt([e("X1", "CRDT", "100.00", acct="GB00ACME12345678"), e("X2", "CRDT", "99.00")]))
# the imported credits made their own receipts; match a manual receipt to a line made by an OPEN credit instead
with on_day("2026-11-18T09:00:00+00:00"):
must(api("POST", "/reconciliation-rules/ANY", APPROVER, {"direction": "credit", "text_contains": "MANUALMATCH", "account_id": "7000",
"tax_code": "E", "mode": "suggest"}))
imp(camt([e("M1", "CRDT", "100.00", text="MANUALMATCH"), e("M2", "CRDT", "100.00", text="MANUALMATCH"), e("M3", "CRDT", "90.00", text="MANUALMATCH")]))
refused(api("POST", "/statement-lines/bank:BANK1:M3/match", CASHIER, {"kind": "receipt", "ref": pid}), 409, "same amount and currency")
must(api("POST", "/statement-lines/bank:BANK1:M1/match", CASHIER, {"kind": "receipt", "ref": pid}))
refused(api("POST", "/statement-lines/bank:BANK1:M2/match", CASHIER, {"kind": "receipt", "ref": pid}), 409, "already matched to line bank:BANK1:M1")
def test_PAY_21_two_lines_racing_for_one_counterpart_collide_on_its_claim(nb):
two_banks()
pid = receipt(amount="100.00", reference="typed", bank_account_id="BANK1")["payment_id"]
with on_day("2026-11-18T09:00:00+00:00"):
must(api("POST", "/reconciliation-rules/ANY", APPROVER, {"direction": "credit", "text_contains": "MANUALMATCH", "account_id": "7000",
"tax_code": "E", "mode": "suggest"}))
imp(camt([e("M1", "CRDT", "100.00", text="MANUALMATCH"), e("M2", "CRDT", "100.00", text="MANUALMATCH")]))
must(api("POST", "/statement-lines/bank:BANK1:M1/match", CASHIER, {"kind": "receipt", "ref": pid}))
real_count = t.ClaimRepository.count
with mock.patch.object(t.BankRepository, "holder_of", lambda self, kind, ref: None), \
mock.patch.object(t.ClaimRepository, "count", lambda self, scope, key: 0 if scope == "counterpart" else real_count(self, scope, key)):
# the second writer began before the first committed: it sees the counterpart free, takes the same claim and is refused
status, body = api("POST", "/statement-lines/bank:BANK1:M2/match", CASHIER, {"kind": "receipt", "ref": pid})
assert status == 409 and "Someone else wrote" in str(body)
def test_PAY_21_posting_a_line_and_matching_it_at_the_same_moment_collide_on_the_lines_claim(nb):
two_banks()
nb.liabilities = [{"period_key": "26AD", "charge_type": "vat", "outstanding_amount": "50.00", "due_date": "2026-11-07"}]
vat_control()
imp(camt([e("D1", "DBIT", "50.00", name="HMRC", text="HMRC")]))
before = lines()["bank:BANK1:D1"]
must(api("POST", "/statement-lines/bank:BANK1:D1/post", CASHIER, {"account_id": "7500", "tax_code": "E"}))
real_count = t.ClaimRepository.count
with use_cases.repository() as repo:
stale_row = repo.bank.line("bank:BANK1:D1")
stale_row = {**stale_row, "lifecycle": "open", "journal_id": "", "account_id": "", "decided_how": ""}
with mock.patch.object(t.BankRepository, "line", lambda self, i: dict(stale_row)), \
mock.patch.object(t.ClaimRepository, "count", lambda self, scope, key: 1 if scope == "line" else real_count(self, scope, key)):
status, body = api("POST", "/statement-lines/bank:BANK1:D1/match", CASHIER, {"kind": "tax_liability", "ref": "26AD:vat"})
assert status == 409 and "Someone else wrote" in str(body)
assert before["lifecycle"] == "open" and lines()["bank:BANK1:D1"]["lifecycle"] == "posted" # the first operation stands
def test_PAY_21_undoing_a_posting_is_a_recorded_reversal_with_a_reason_and_the_line_is_open_again(nb):
two_banks()
imp(camt([e("D1", "DBIT", "12.50", name="Bank", text="SERVICE CHARGE")]))
must(api("POST", "/statement-lines/bank:BANK1:D1/post", CASHIER, {"account_id": "7900", "tax_code": "E"}))
refused(api("POST", "/statement-lines/bank:BANK1:D1/unmatch", CASHIER, {"reason": ""}), 422)
got = must(api("POST", "/statement-lines/bank:BANK1:D1/unmatch", CASHIER, {"reason": "wrong account"}))
assert got["lifecycle"] == "open" and got["released"]["kind"] == "manual"
assert {k.split(":")[0] for k in nb.journals} == {"bank_line", "bank_line_reversal"}
rev = next(j for k, j in nb.journals.items() if k.startswith("bank_line_reversal:bank:BANK1:D1@")) # keyed on the line AND the unmatch's claim
assert [(ln.get("account_id"), ln.get("debit"), ln.get("credit")) for ln in rev["lines"]] == [("1200", "12.50", None), ("7900", None, "12.50")]
assert lines()["bank:BANK1:D1"]["reversal_reason"] == "wrong account"
refused(api("POST", "/statement-lines/bank:BANK1:D1/unmatch", CASHIER, {"reason": "again"}), 409, "nothing to undo")
must(api("POST", "/statement-lines/bank:BANK1:D1/post", CASHIER, {"account_id": "7500", "tax_code": "E"})) # and it can be posted again
def test_PAY_21_a_reconciled_statement_refuses_an_undo_and_a_credit_that_made_its_receipt_is_reversed_not_unmatched(nb):
two_banks()
imp(camt([e("C1", "CRDT", "100.00"), e("D1", "DBIT", "12.50", name="Bank")], stmt_id="S1"))
must(api("POST", "/statement-lines/bank:BANK1:D1/post", CASHIER, {"account_id": "7900", "tax_code": "E"}))
refused(api("POST", "/statement-lines/bank:BANK1:C1/unmatch", CASHIER, {"reason": "x"}), 409, "reverse the receipt")
with use_cases.repository() as repo:
repo.conn.execute(t.pay_bank_statement.update().values(lifecycle="reconciled"))
refused(api("POST", "/statement-lines/bank:BANK1:D1/unmatch", CASHIER, {"reason": "x"}), 409, "is reconciled")
def test_PAY_21_a_line_is_posted_only_with_an_account_and_a_tax_code_and_the_ledger_has_to_take_it(nb):
two_banks()
imp(camt([e("D1", "DBIT", "12.50", name="Bank")]))
refused(api("POST", "/statement-lines/bank:BANK1:D1/post", CASHIER, {"account_id": "7900", "tax_code": " "}), 409, "needs a tax code")
nb.gl.router.routes[:] = [r for r in nb.gl.router.routes if getattr(r, "path", "") != "/journal-entry/journals"]
refused(api("POST", "/statement-lines/bank:BANK1:D1/post", CASHIER, {"account_id": "7900", "tax_code": "E"}), 409, "Nothing was posted")
assert lines()["bank:BANK1:D1"]["lifecycle"] == "open"
# ── PAY-22: the AI suggests, a person decides ────────────────────────────────────────────────────────
def posted_amazon(day="2026-11-10"):
imp(camt([e("A0", "DBIT", "20.00", day=day, text="AMZN MKTP UK 20.00", name="AMZN MKTP UK")], stmt_id="OLD"))
must(api("POST", "/statement-lines/bank:BANK1:A0/post", CASHIER, {"account_id": "7500", "tax_code": "S", "receipt_image": "receipts/a0.jpg"}))
def test_PAY_22_the_suggestion_is_a_row_with_evidence_and_the_ai_never_posts(nb):
two_banks()
posted_amazon()
imp(camt([e("A1", "DBIT", "45.99", day="2026-11-19", text="AMZN MKTP UK 45.99", name="AMZN MKTP UK")], stmt_id="NOV"))
line = lines()["bank:BANK1:A1"]
assert line["lifecycle"] == "open" # nothing posted by the suggestion
(s,) = line["suggestions"]
assert (s["suggestion_id"], s["account_id"], s["tax_code"], s["receipt_needed"], s["model"], s["status"]) == (
"bank:BANK1:A1#1", "7500", "S", "Y", "history-payee-v1", "open")
assert s["evidence"]["payee"] == "amzn mktp" and s["evidence"]["same_payee_postings"] == 1 and Decimal(s["confidence"]) > 0
assert not any(k.endswith("bank:BANK1:A1") for k in nb.journals)
def test_PAY_22_accepting_posts_with_the_vat_from_tax_reclaimed_only_with_a_receipt_image(nb):
two_banks()
posted_amazon()
imp(camt([e("A1", "DBIT", "45.99", day="2026-11-19", text="AMZN MKTP UK 45.99", name="AMZN MKTP UK")], stmt_id="NOV"))
got = must(api("POST", "/statement-lines/bank:BANK1:A1/decide-suggestion", CASHIER, {"decision": "accept"}))
assert (got["lifecycle"], got["decided_how"], got["vat"], got["vat_reclaimed"]) == ("posted", "accepted", "7.66", False) # no image: not reclaimed
j = nb.journals["bank_line:bank:BANK1:A1"]
assert j["lines"][0] == {"account_id": "7500", "debit": "45.99"} and "tax_code" not in j["lines"][0]
assert nb.tax_calls[-1]["direction"] == "purchase" and nb.tax_calls[-1]["document_ref"] == "pay_line:bank:BANK1:A1"
line = lines()["bank:BANK1:A1"]
assert (line["decided_by"], line["decided_how"], line["lifecycle"]) == ("cashier@shop.example", "accepted", "posted")
j0 = nb.journals["bank_line:bank:BANK1:A0"] # the earlier one had its image: VAT reclaimed
assert j0["lines"][0]["tax_code"] == "S"
def test_PAY_22_changing_the_account_records_changed_and_rejecting_leaves_the_line_open(nb):
two_banks()
posted_amazon()
imp(camt([e("A1", "DBIT", "45.99", day="2026-11-19", text="AMZN MKTP UK", name="AMZN MKTP UK"),
e("A2", "DBIT", "10.00", day="2026-11-19", text="AMZN MKTP UK", name="AMZN MKTP UK")], stmt_id="NOV"))
refused(api("POST", "/statement-lines/bank:BANK1:A1/decide-suggestion", CASHIER, {"decision": "change"}), 409, "account you choose")
got = must(api("POST", "/statement-lines/bank:BANK1:A1/decide-suggestion", CASHIER, {"decision": "change", "account_id": "7600", "tax_code": "E"}))
assert got["decided_how"] == "changed" and lines()["bank:BANK1:A1"]["account_id"] == "7600"
rej = must(api("POST", "/statement-lines/bank:BANK1:A2/decide-suggestion", CASHIER, {"decision": "reject"}))
assert rej["lifecycle"] == "open" and lines()["bank:BANK1:A2"]["suggestions"] == []
assert "bank_line:bank:BANK1:A2" not in nb.journals # the rejected suggestion left no posting
assert [s["status"] for s in table(t.pay_line_suggestion) if s["line_id"] == "bank:BANK1:A2"] == ["rejected"]
def test_PAY_22_a_decided_suggestion_is_final_and_a_late_reject_after_an_accept_is_refused(nb):
two_banks()
posted_amazon()
imp(camt([e("A1", "DBIT", "45.99", day="2026-11-19", text="AMZN MKTP UK", name="AMZN MKTP UK")], stmt_id="NOV"))
must(api("POST", "/statement-lines/bank:BANK1:A1/decide-suggestion", CASHIER, {"decision": "accept"}))
for d in ("reject", "accept"):
refused(api("POST", "/statement-lines/bank:BANK1:A1/decide-suggestion", CASHIER, {"decision": d, "suggestion_id": "bank:BANK1:A1#1"}), 409, "already accepted")
assert lines()["bank:BANK1:A1"]["lifecycle"] == "posted"
def test_PAY_22_an_accept_and_a_reject_racing_for_one_line_collide_on_its_claim(nb):
two_banks()
posted_amazon()
imp(camt([e("A1", "DBIT", "45.99", day="2026-11-19", text="AMZN MKTP UK", name="AMZN MKTP UK")], stmt_id="NOV"))
with use_cases.repository() as repo:
sug_before = repo.bank.suggestion("bank:BANK1:A1#1")
must(api("POST", "/statement-lines/bank:BANK1:A1/decide-suggestion", CASHIER, {"decision": "accept"}))
real_count, row = t.ClaimRepository.count, {**table(t.pay_statement_line)[-1], "lifecycle": "open"}
with mock.patch.object(t.BankRepository, "suggestions_of", lambda self, i: [sug_before]), \
mock.patch.object(t.ClaimRepository, "count", lambda self, scope, key: 1 if scope == "line" else real_count(self, scope, key)), \
mock.patch.object(t.BankRepository, "line", lambda self, i: dict(row)):
status, body = api("POST", "/statement-lines/bank:BANK1:A1/decide-suggestion", CASHIER, {"decision": "reject"})
assert status == 409 and "Someone else wrote" in str(body)
assert [s["status"] for s in table(t.pay_line_suggestion) if s["line_id"] == "bank:BANK1:A1"] == ["accepted"]
def test_PAY_22_running_the_suggestion_again_supersedes_an_open_one_and_never_a_decided_one(nb):
two_banks()
posted_amazon()
imp(camt([e("A1", "DBIT", "45.99", day="2026-11-19", text="AMZN MKTP UK", name="AMZN MKTP UK")], stmt_id="NOV"))
must(api("POST", "/statement-lines/bank:BANK1:A1/suggest", SYSTEM))
sugs = {s["suggestion_id"]: s["status"] for s in table(t.pay_line_suggestion) if s["line_id"] == "bank:BANK1:A1"}
assert sugs == {"bank:BANK1:A1#1": "superseded", "bank:BANK1:A1#2": "open"}
must(api("POST", "/statement-lines/bank:BANK1:A1/decide-suggestion", CASHIER, {"decision": "accept"}))
must(api("POST", "/statement-lines/bank:BANK1:A1/suggest", SYSTEM))
assert {s["suggestion_id"]: s["status"] for s in table(t.pay_line_suggestion) if s["line_id"] == "bank:BANK1:A1"}["bank:BANK1:A1#2"] == "accepted"
def test_PAY_22_a_cashier_cannot_run_the_system_step_and_a_step_turned_off_suggests_nothing(nb):
two_banks()
posted_amazon()
refused(api("POST", "/statement-lines/bank:BANK1:A0/suggest", CASHIER), 403, "pay:statement_line:suggest")
must(api("POST", "/settings/ai_suggest_statement_line", APPROVER, {"value": "off"}))
imp(camt([e("A1", "DBIT", "45.99", day="2026-11-19", text="AMZN MKTP UK", name="AMZN MKTP UK")], stmt_id="NOV"))
assert lines()["bank:BANK1:A1"]["suggestions"] == []
# ── PAY-23: reconciliation rules ─────────────────────────────────────────────────────────────────────
RULE = {"direction": "debit", "text_contains": "SERVICE CHARGE", "bank_account_id": "BANK1", "account_id": "7900", "tax_code": "E"}
def test_PAY_23_a_rule_is_a_dated_row_from_tomorrow_and_a_change_is_a_new_row(nb):
two_banks()
got = must(api("POST", "/reconciliation-rules/CHG", APPROVER, {**RULE, "mode": "suggest"}))
assert got["valid_from"] == "2026-11-21"
imp(camt([e("D1", "DBIT", "12.50", text="SERVICE CHARGE", name="Bank")]))
assert lines()["bank:BANK1:D1"]["suggestions"] == [] # not in force today
refused(api("POST", "/reconciliation-rules/CHG", APPROVER, {**RULE, "mode": "suggest"}), 409, "already has a change that starts 2026-11-21")
with on_day("2026-11-21T09:00:00+00:00"):
must(api("POST", "/reconciliation-rules/CHG", APPROVER, {**RULE, "account_id": "7901", "mode": "suggest"}))
rules = must(api("GET", "/reconciliation-rules", CASHIER))["rules"]
assert [(r["valid_from"], r["account_id"], r["in_force"]) for r in rules] == [("2026-11-21", "7900", True), ("2026-11-22", "7901", False)]
def test_PAY_23_in_suggest_mode_a_matching_line_gets_the_proposal_for_the_cashier_to_accept(nb):
two_banks()
with on_day("2026-11-18T09:00:00+00:00"):
must(api("POST", "/reconciliation-rules/SC", APPROVER, {**RULE, "mode": "suggest"}))
imp(camt([e("D1", "DBIT", "12.50", text="SERVICE CHARGE NOV", name="Bank")]))
(s,) = lines()["bank:BANK1:D1"]["suggestions"]
assert (s["model"], s["account_id"], s["tax_code"]) == ("rule:SC", "7900", "E") and lines()["bank:BANK1:D1"]["lifecycle"] == "open"
got = must(api("POST", "/statement-lines/bank:BANK1:D1/decide-suggestion", CASHIER, {"decision": "accept"}))
assert got["decided_how"] == "rule" and got["lifecycle"] == "posted"
def test_PAY_23_auto_mode_needs_the_owners_control_posts_with_decided_how_auto_and_is_listed_for_review(nb):
two_banks()
refused(api("POST", "/reconciliation-rules/SC", person("clerk@shop.example", ["pay:reconciliation_rule:write"]), {**RULE, "mode": "auto"}), 403,
"pay:settings:controls")
with on_day("2026-11-18T09:00:00+00:00"):
must(api("POST", "/reconciliation-rules/SC", APPROVER, {**RULE, "mode": "auto"}))
imp(camt([e("D1", "DBIT", "12.50", text="SERVICE CHARGE NOV", name="Bank")]))
line = lines()["bank:BANK1:D1"]
assert (line["lifecycle"], line["decided_how"], line["decided_by"], line["account_id"]) == ("posted", "auto", "rule:SC", "7900")
listed = must(api("GET", "/statement-lines/auto-posted", CASHIER))["lines"]
assert [x["line_id"] for x in listed] == ["bank:BANK1:D1"]
must(api("POST", "/statement-lines/bank:BANK1:D1/unmatch", CASHIER, {"reason": "not a bank charge"})) # the cashier undoes it
assert lines()["bank:BANK1:D1"]["lifecycle"] == "open"
assert imp(camt([e("D1", "DBIT", "12.50", text="SERVICE CHARGE NOV", name="Bank")]))["lines_created"] == 0 # and a re-import does not redo it
def test_PAY_23_two_rules_matching_one_line_are_both_shown_and_neither_is_applied(nb):
two_banks()
with on_day("2026-11-18T09:00:00+00:00"):
must(api("POST", "/reconciliation-rules/A", APPROVER, {**RULE, "mode": "auto"}))
must(api("POST", "/reconciliation-rules/B", APPROVER, {**RULE, "text_contains": "CHARGE", "account_id": "7901", "mode": "auto"}))
imp(camt([e("D1", "DBIT", "12.50", text="SERVICE CHARGE", name="Bank")]))
line = lines()["bank:BANK1:D1"]
assert line["lifecycle"] == "open" and sorted(s["model"] for s in line["suggestions"]) == ["rule:A", "rule:B"]
assert not nb.journals
def test_PAY_23_a_rule_never_matches_a_line_to_a_customers_invoice(nb):
two_banks()
nb.invoice("INV-1", 7311, "100.00")
with on_day("2026-11-18T09:00:00+00:00"):
must(api("POST", "/reconciliation-rules/ANY", APPROVER, {"direction": "credit", "text_contains": "INV-1", "account_id": "7000", "tax_code": "E", "mode": "suggest"}))
imp(camt([e("C1", "CRDT", "100.00", text="INV-1")]))
assert table(t.pay_allocation) == [] # a rule allocates nothing, ever
assert [x["account_id"] for x in must(api("GET", "/reconciliation-rules", CASHIER))["rules"]] == ["7000"]
def test_PAY_23_a_rule_needs_something_to_recognise_a_line_by_and_a_bounded_band(nb):
two_banks()
refused(api("POST", "/reconciliation-rules/X", APPROVER, {"direction": "debit", "account_id": "7900", "tax_code": "E"}), 409, "something to recognise")
refused(api("POST", "/reconciliation-rules/X", APPROVER, {**RULE, "amount_min": "10.00", "amount_max": "5.00"}), 422, "amount_min")
assert _api("POST", "/reconciliation-rules/X", APPROVER, {**RULE, "amount_max": "1e30"})[0] == 422
assert APPROVER2 and SYSTEM
def test_PAY_23_a_rule_starts_the_business_day_after_today_in_the_business_timezone_not_utc_days(nb):
two_banks()
must(api("POST", "/controls/business_timezone", APPROVER, {"value": "Pacific/Auckland"})) # a dated control: in force from the next business date
with on_day("2026-11-21T12:30:00+00:00"): # 2026-11-21 in London, already 2026-11-22 in Auckland
got = must(api("POST", "/reconciliation-rules/TZ", APPROVER, {**RULE, "mode": "suggest"}))
assert got["valid_from"] == "2026-11-23"
def test_every_v4_slice_a_use_case_is_a_tool_for_whoever_may_call_it_and_the_system_steps_are_routes_only(nb):
from conftest import tools
names = set(tools(BOTH_ALL))
assert {"import_bank_statement", "post_statement_line", "decide_line_suggestion", "match_statement_line", "unmatch_statement_line",
"put_reconciliation_rule", "apply_settlement_discount", "import_remittance_file", "apply_remittance", "record_dd_mandate",
"cancel_dd_mandate", "lodge_dd_mandates", "prepare_dd_run", "approve_dd_run", "send_dd_notices", "submit_dd_run",
"import_dd_reports", "put_house_bank"} <= names
assert not {"suggest_statement_match", "collect_dd_run", "mark_mandate_dormant"} & names # a machine's step is a route, never an agent's tool
BOTH_ALL = person("all@shop.example", [p for perms in __import__("conftest").ROLES.values() for p in perms])
def test_PAY_21_a_second_unmatch_of_the_same_line_is_a_new_reversal_and_a_posting_after_an_undo_is_a_new_journal(nb):
two_banks()
imp(camt([e("D1", "DBIT", "12.50", name="Bank", text="SERVICE CHARGE")]))
line = "/statement-lines/bank:BANK1:D1"
must(api("POST", f"{line}/post", CASHIER, {"account_id": "7900", "tax_code": "E"}))
must(api("POST", f"{line}/unmatch", CASHIER, {"reason": "first"}))
must(api("POST", f"{line}/post", CASHIER, {"account_id": "7500", "tax_code": "E"}))
must(api("POST", f"{line}/unmatch", CASHIER, {"reason": "second"}))
revs = [k for k in nb.journals if k.startswith("bank_line_reversal:")]
posts = [k for k in nb.journals if k.startswith("bank_line:")]
assert len(revs) == 2 and len(set(revs)) == 2 # the second unmatch is NOT a replay of the first
assert posts[0] == "bank_line:bank:BANK1:D1" and len(posts) == 2 and posts[1].startswith("bank_line:bank:BANK1:D1@")
assert nb.journals[posts[1]]["lines"][0]["account_id"] == "7500"