dodil/erp-brick-payments
DODIL ERP brick: payments
tests/test_v4_slice_b.py
"""Payments v4, slice B - PAY-24 (charges and interest), PAY-25 (transfers between our accounts), PAY-26 (petty cash), PAY-27 (bank accounts and
the unmasked read purchasing needs), PAY-28 (deposit batches of cheques), PAY-29 (the cash forecast). Through the real app on SQLite; the neighbours
(ar, bp, gl, tax, purchasing) are tests/fakes.py, each answer checked against its callee's signed spec."""
from decimal import Decimal
from unittest import mock
import pytest
from conftest import (
APPROVER,
CASHIER,
COLLECTOR,
NOBODY,
ROLES,
SALES,
SYSTEM,
allocate,
bank,
must,
on_day,
person,
receipt,
refused,
settle,
)
from conftest import api as _api
from erp_platform.seam import derived_id
from sqlalchemy import select
from test_v4_directdebit import PREP, active_mandates, ident
from test_v4_directdebit import run as dd_run
from test_v4_directdebit import setup as dd_setup
from test_v4_statements import camt, e, imp, lines
from brick_payments.application import use_cases
from brick_payments.infrastructure import tables as t
api = _api
D = Decimal
def journals(nb, event):
return [j for j in nb.journals.values() if j["event"] == event]
def clearing(nb):
"""What the ledger's transfer clearing account shows: debits less credits over every transfer journal."""
net = D(0)
for j in journals(nb, "transfer") + journals(nb, "bank_line_reversal"):
for ln in j["lines"]:
if ln.get("purpose") == "transfer_clearing":
net += D(ln.get("debit", "0")) - D(ln.get("credit", "0"))
return net
def table(tbl):
with use_cases.repository() as repo:
return [dict(m) for m in repo.conn.execute(select(tbl)).mappings()]
class Snapshot:
"""A transaction that began before another committed does not see its claim rows: it counts the claims as they WERE."""
def __init__(self):
self.counts = {}
for c in table(t.pay_claim):
self.counts[(c["claim_scope"], c["claim_key"])] = self.counts.get((c["claim_scope"], c["claim_key"]), 0) + 1
def __enter__(self):
self.patch = mock.patch.object(t.ClaimRepository, "count", lambda me, scope, key: self.counts.get((scope, key), 0))
self.patch.start()
return self
def __exit__(self, *a):
self.patch.stop()
# ── PAY-24: charges, fees and interest ────────────────────────────────────────────────────────────────
def charges_statement():
bank("BANK1", gl="1200", account_number="12345678")
with on_day("2026-11-18T09:00:00+00:00"):
must(api("POST", "/reconciliation-rules/INT", APPROVER, {"direction": "credit", "text_contains": "INTEREST", "account_id": "7000", "tax_code": "E", "mode": "suggest"}))
imp(camt([e("D1", "DBIT", "12.50", text="SERVICE CHARGE", name="Bank"), e("D2", "DBIT", "38.20", day="2026-11-17", text="CARD TERMINAL FEES", name="Elavon"),
e("I1", "CRDT", "3.21", text="CREDIT INTEREST", name="Bank")], stmt_id="NOV-1"))
def post(line, account, tax="E", who=None, **more):
return api("POST", f"/statement-lines/bank:BANK1:{line}/post", who or CASHIER, {"account_id": account, "tax_code": tax, **more})
def test_PAY_24_each_line_posts_one_journal_dated_by_the_banks_booking_date(nb):
charges_statement()
for line, acct in (("D1", "6900"), ("D2", "6950"), ("I1", "7000")):
must(post(line, acct))
by = {j["source_document"]: j for j in journals(nb, "bank_line")}
d1, d2, i1 = (by[f"bank:BANK1:{x}"] for x in ("D1", "D2", "I1"))
assert (d1["journal_date"], d2["journal_date"], i1["journal_date"]) == ("2026-11-18", "2026-11-17", "2026-11-18") # the bank's date, not today's
assert d1["lines"] == [{"account_id": "6900", "debit": "12.50"}, {"account_id": "1200", "credit": "12.50"}]
assert d2["lines"] == [{"account_id": "6950", "debit": "38.20"}, {"account_id": "1200", "credit": "38.20"}]
assert i1["lines"] == [{"account_id": "1200", "debit": "3.21"}, {"account_id": "7000", "credit": "3.21"}] # no VAT on interest
assert d1["journal_id"] == str(derived_id("bank:BANK1:D1", "bank_line")) # derived from the line
assert not [c for c in nb.tax_calls if c["document_ref"] == "pay_line:bank:BANK1:D1"] # E: tax has nothing to compute
def test_PAY_24_posting_twice_changes_nothing_and_the_account_is_never_a_default(nb):
charges_statement()
must(post("D1", "6900"))
n = len(nb.journals)
refused(post("D1", "6950"), 409, "already posted")
assert len(nb.journals) == n
status, _ = api("POST", "/statement-lines/bank:BANK1:D2/post", CASHIER, {"tax_code": "E"}) # no account named: not defaulted
assert status == 422
refused(post("D2", " "), 409, "ledger account")
def test_PAY_24_a_posted_line_is_undone_only_by_a_reversing_journal_with_a_reason(nb):
charges_statement()
must(post("D1", "6900"))
assert api("POST", "/statement-lines/bank:BANK1:D1/unmatch", CASHIER, {})[0] == 422 # a reason is required
must(api("POST", "/statement-lines/bank:BANK1:D1/unmatch", CASHIER, {"reason": "wrong account"}))
rev = journals(nb, "bank_line_reversal")[0]
assert rev["lines"] == [{"account_id": "1200", "debit": "12.50"}, {"account_id": "6900", "credit": "12.50"}]
assert len(journals(nb, "bank_line")) == 1 # the original stays; nothing was deleted
# ── PAY-25: a transfer between our own accounts ───────────────────────────────────────────────────────
def two_accounts():
bank("BANK1", gl="1200", account_number="12345678")
bank("BANK2", gl="1210", account_number="87654321")
OUT = ("T1", "DBIT", "20000.00", "GBP", "2026-11-14", "TRANSFER TO BARC-02", "Dodil", "GB11TEST87654321")
def side_out(day="2026-11-14", ccy="GBP", amount="20000.00", ref="T1"):
return imp(camt([(ref, "DBIT", amount, ccy, day, "TRANSFER", "Dodil", "GB11TEST87654321")], stmt_id=f"B1-{ref}"), account="BANK1")
def side_in(day="2026-11-15", ccy="GBP", amount="20000.00", ref="T2"):
return imp(camt([(ref, "CRDT", amount, ccy, day, "TRANSFER", "Dodil", "GB00TEST12345678")], stmt_id=f"B2-{ref}"), account="BANK2")
def test_PAY_25_the_two_sides_post_on_their_own_dates_through_clearing(nb):
two_accounts()
got = side_out()
assert got["transfers"] == {"in_transit": 1, "paired": 0, "failed": []}
out = journals(nb, "transfer")[0]
assert out["journal_date"] == "2026-11-14" and out["lines"] == [{"purpose": "transfer_clearing", "debit": "20000.00"}, {"account_id": "1200", "credit": "20000.00"}]
assert clearing(nb) == D("20000.00") # in transit: clearing shows it
transit = must(api("GET", "/transfers/in-transit", CASHIER))
assert [(i["line_id"], i["to_bank_account_id"], i["amount"]) for i in transit["in_transit"]] == [("bank:BANK1:T1", "BANK2", "20000.00")]
assert transit["totals"] == {"GBP": "20000.00"}
got = side_in()
assert got["transfers"]["paired"] == 1 and got["created"] == 0 # not a customer's receipt
inn = [j for j in journals(nb, "transfer") if j["journal_date"] == "2026-11-15"][0]
assert inn["lines"] == [{"account_id": "1210", "debit": "20000.00"}, {"purpose": "transfer_clearing", "credit": "20000.00"}]
assert clearing(nb) == D(0) # zero once both have arrived
ls = lines()
assert (ls["bank:BANK1:T1"]["matched_kind"], ls["bank:BANK1:T1"]["matched_ref"]) == ("transfer", "bank:BANK2:T2")
assert (ls["bank:BANK2:T2"]["matched_kind"], ls["bank:BANK2:T2"]["matched_ref"]) == ("transfer", "bank:BANK1:T1")
assert must(api("GET", "/transfers/in-transit", CASHIER))["in_transit"] == []
def test_PAY_25_a_debit_with_no_credit_yet_stays_in_transit_and_a_rerun_changes_nothing(nb):
two_accounts()
side_out()
n = len(nb.journals)
assert must(api("POST", "/transfers/reconcile", CASHIER, {})) == {"in_transit": 0, "paired": 0, "failed": []}
assert len(nb.journals) == n
def test_PAY_25_different_currencies_or_dates_outside_the_window_are_never_one_transfer(nb):
two_accounts()
side_out(day="2026-11-04")
assert side_in(day="2026-11-05", ccy="EUR", ref="E1")["transfers"]["paired"] == 0 # same number, other currency
assert side_in(day="2026-11-15", ref="L1")["transfers"]["paired"] == 0 # 11 days after the debit: window is 5
assert clearing(nb) == D("20000.00")
must(api("POST", "/settings/transfer_window_days", APPROVER, {"value": 12}))
assert must(api("POST", "/transfers/reconcile", CASHIER, {}))["paired"] == 1 # wider window, same GBP credit
assert lines()["bank:BANK2:E1"]["matched_kind"] != "transfer" and lines()["bank:BANK2:L1"]["matched_ref"] == "bank:BANK1:T1"
def test_PAY_25_a_credit_pairs_with_one_debit_only(nb):
two_accounts()
side_out(ref="T1")
imp(camt([("T3", "DBIT", "20000.00", "GBP", "2026-11-14", "TRANSFER", "Dodil", "GB11TEST87654321")], stmt_id="B1-T3"), account="BANK1")
side_in()
ls = lines()
paired = [ls[f"bank:BANK1:{x}"]["matched_ref"] for x in ("T1", "T3")]
assert sorted(paired) == ["", "bank:BANK2:T2"] # the other debit is still in transit
assert len(must(api("GET", "/transfers/in-transit", CASHIER))["in_transit"]) == 1
def test_PAY_25_unmatching_the_credit_puts_the_debit_back_in_transit_and_reverses_its_journal(nb):
two_accounts()
side_out()
side_in()
assert api("POST", "/statement-lines/bank:BANK1:T1/unmatch", CASHIER, {"reason": "x"})[0] == 409 # the arriving side first
must(api("POST", "/statement-lines/bank:BANK2:T2/unmatch", CASHIER, {"reason": "wrong account"}))
assert clearing(nb) == D("20000.00")
assert len(must(api("GET", "/transfers/in-transit", CASHIER))["in_transit"]) == 1
# ── PAY-26: petty cash ────────────────────────────────────────────────────────────────────────────────
def journal(opening="200.00"):
return must(api("POST", "/cash-journals/PETTY-UK01", CASHIER, {"name": "Petty cash UK01", "currency": "GBP", "gl_account_id": "1010",
"company_code": "UK01", "opening_balance": opening}))
def entry(kind, amount, who=None, **more):
return api("POST", "/cash-journals/PETTY-UK01/entries", who or CASHIER, {"kind": kind, "amount": amount, **more})
def day_of_cash(nb):
journal()
must(api("POST", "/cash-journals/PETTY-UK01/open-day", CASHIER))
nb.invoice("INV-C1", 7311, "50.00", due_date="2026-11-25")
def test_PAY_26_the_days_entries_balance_to_the_count_and_the_day_closes(nb):
day_of_cash(nb)
paid = must(entry("receipt", "50.00", bp_id=7311, description="Customer paid cash"))
assert paid["receipt"]["method"] == "cash" and paid["receipt"]["status"] == "received"
must(allocate(paid["receipt"]["payment_id"], [("INV-C1", "50.00")])) # allocated like any other
must(entry("expense", "12.00", tax_code="E", account_id="6100", description="Stamps"))
clean = must(entry("expense", "24.00", tax_code="S", account_id="6200", description="Cleaning materials", receipt_image="img/clean-1.jpg"))
assert (clean["vat"], clean["vat_reclaimed"]) == ("4.00", True)
tax_call = [c for c in nb.tax_calls if c["document_ref"].startswith("pay_cash:PETTY-UK01:2026-11-20")][0]
assert tax_call["direction"] == "purchase" and tax_call["prices_include_tax"] is True
counted = must(api("POST", "/cash-journals/PETTY-UK01/count", CASHIER, {"denominations": [{"value": "20.00", "count": 10}, {"value": "10.00", "count": 1},
{"value": "2.00", "count": 2}]}))
assert (counted["expected_balance"], counted["counted_amount"], counted["difference"]) == ("214.00", "214.00", "0.00")
closed = must(api("POST", "/cash-journals/PETTY-UK01/close-day", CASHIER))
assert closed["lifecycle"] == "closed" and closed["difference_journal_id"] == "" and not journals(nb, "cash_difference")
reclaimed = [j for j in journals(nb, "cash_entry") if j["source_document"] == "PETTY-UK01:2026-11-20:3"][0]
assert reclaimed["lines"] == [{"account_id": "6200", "tax_code": "S", "debit": "24.00"}, {"account_id": "1010", "credit": "24.00"}]
refused(entry("expense", "1.00", tax_code="E", account_id="6100"), 409, "closed") # a closed day refuses entries
def test_PAY_26_a_cash_expense_needs_an_account_and_a_tax_code_and_is_reclaimed_only_with_the_receipt(nb):
day_of_cash(nb)
refused(entry("expense", "5.00", tax_code="E"), 409, "account")
refused(entry("expense", "5.00", account_id="6100"), 409, "tax code")
no_img = must(entry("expense", "24.00", tax_code="S", account_id="6200"))
assert (no_img["vat"], no_img["vat_reclaimed"]) == ("4.00", False)
assert [j for j in journals(nb, "cash_entry") if j["source_document"].endswith(":1")][0]["lines"][0] == {"account_id": "6200", "debit": "24.00"}
assert entry("expense", "5.001", tax_code="E", account_id="6100")[0] == 422
def test_PAY_26_a_small_difference_posts_to_over_short_and_a_large_one_needs_another_person(nb):
day_of_cash(nb)
must(api("POST", "/cash-journals/PETTY-UK01/count", CASHIER, {"counted_amount": "198.50"}))
must(api("POST", "/cash-journals/PETTY-UK01/close-day", CASHIER))
short = journals(nb, "cash_difference")[0]
assert short["lines"] == [{"purpose": "cash_over_short", "debit": "1.50"}, {"account_id": "1010", "credit": "1.50"}]
with on_day("2026-11-21T09:00:00+00:00"):
assert must(api("POST", "/cash-journals/PETTY-UK01/open-day", CASHIER))["opening_balance"] == "198.50" # the last counted balance
must(api("POST", "/cash-journals/PETTY-UK01/count", CASHIER, {"counted_amount": "188.50"})) # 10.00 short: above 5.00
refused(api("POST", "/cash-journals/PETTY-UK01/close-day", CASHIER), 409, "approver")
BOTH = ident("u-ann", "ann@shop.example", [*ROLES["cashier"], *ROLES["approver"]])
BOTH_SUB = ident("u-ann", "ann2@shop.example", [*ROLES["cashier"], *ROLES["approver"]])
must(api("POST", "/cash-journals/PETTY-UK01/count", BOTH, {"counted_amount": "188.50"}))
refused(api("POST", "/cash-journals/PETTY-UK01/approve-difference", BOTH, {"reason": "r"}), 409, "counted") # same email
refused(api("POST", "/cash-journals/PETTY-UK01/approve-difference", BOTH_SUB, {"reason": "r"}), 409, "counted") # same sub, other email
must(api("POST", "/cash-journals/PETTY-UK01/approve-difference", APPROVER, {"reason": "till checked twice"}))
must(api("POST", "/cash-journals/PETTY-UK01/close-day", CASHIER))
assert sorted(j["lines"][0]["debit"] for j in journals(nb, "cash_difference")) == ["1.50", "10.00"]
def test_PAY_26_a_new_count_withdraws_the_approval_and_a_day_opens_only_after_the_last_closed(nb):
day_of_cash(nb)
must(api("POST", "/cash-journals/PETTY-UK01/count", CASHIER, {"counted_amount": "100.00"}))
must(api("POST", "/cash-journals/PETTY-UK01/approve-difference", APPROVER, {"reason": "ok"}))
must(api("POST", "/cash-journals/PETTY-UK01/count", CASHIER, {"counted_amount": "90.00"}))
refused(api("POST", "/cash-journals/PETTY-UK01/close-day", CASHIER), 409, "approver")
with on_day("2026-11-21T09:00:00+00:00"):
refused(api("POST", "/cash-journals/PETTY-UK01/open-day", CASHIER), 409, "still open")
refused(entry("expense", "1.00", tax_code="E", account_id="6100"), 409, "not open") # entries are dated today only
def test_PAY_26_an_entry_and_the_close_racing_for_one_day_collide_on_its_claim(nb):
day_of_cash(nb)
must(api("POST", "/cash-journals/PETTY-UK01/count", CASHIER, {"counted_amount": "200.00"}))
with Snapshot():
must(api("POST", "/cash-journals/PETTY-UK01/close-day", CASHIER))
with mock.patch.object(t.CashRepository, "day", lambda self, j, d: {**table(t.pay_cash_day)[0], "lifecycle": "open"}):
status, body = entry("expense", "1.00", tax_code="E", account_id="6100")
assert status == 409 and "Someone else wrote" in str(body)
def test_PAY_26_a_deposit_leaves_through_clearing_and_is_matched_on_the_statement(nb):
day_of_cash(nb)
bank("BANK1", gl="1200", account_number="12345678")
dep = must(entry("deposit", "100.00"))
assert journals(nb, "transfer")[0]["lines"] == [{"purpose": "transfer_clearing", "debit": "100.00"}, {"account_id": "1010", "credit": "100.00"}]
with on_day("2026-11-18T09:00:00+00:00"):
must(api("POST", "/reconciliation-rules/DEP", APPROVER, {"direction": "credit", "text_contains": "CASH PAID IN", "account_id": "1010", "tax_code": "E", "mode": "suggest"}))
imp(camt([e("K1", "CRDT", "100.00", text="CASH PAID IN", name="Counter")], stmt_id="K-1"))
ref = dep["entry_id"]
refused(api("POST", "/statement-lines/bank:BANK1:K1/match", CASHIER, {"kind": "cash_deposit", "ref": "PETTY-UK01:2026-11-20:9"}), 404)
must(api("POST", "/statement-lines/bank:BANK1:K1/match", CASHIER, {"kind": "cash_deposit", "ref": ref}))
assert clearing(nb) == D(0)
assert must(api("GET", "/cash-journals/PETTY-UK01/days/2026-11-20", CASHIER))["entries"][0]["statement_line_id"] == "bank:BANK1:K1"
# ── PAY-27: bank accounts, defaults, and the unmasked read ────────────────────────────────────────────
def barclays():
must(api("POST", "/house-banks/BARC", APPROVER, {"name": "Barclays", "sort_code": "20-00-00", "sun": "123456", "bic": "BARCGB22"}))
bank("BARC-01", gl="1200", account_number="31926819", purposes="receive,pay", house_bank_id="BARC", name="Dodil Trading Ltd")
DETAILS = "/bank-accounts/BARC-01/account-details"
def test_PAY_27_accounts_are_per_company_and_shown_masked_everywhere(nb):
barclays()
bank("BARC-DEP", gl="1201", account_number="55667788", iban="GB29NWBK60161331926819", purposes="receive", house_bank_id="BARC")
bank("LLOY-01", gl="1300", account_number="11223344", company_code="UK02")
listed = {b["bank_account_id"]: b for b in must(api("GET", "/bank-accounts", CASHIER))["bank_accounts"]}
assert listed["BARC-01"]["iban_or_sort_code_masked"] == "****6819" and listed["BARC-DEP"]["iban_or_sort_code_masked"] == "******************6819"
assert "31926819" not in str(listed) and "GB29NWBK" not in str(listed)
refused(api("POST", "/bank-accounts/BARC-01", APPROVER, {"name": "x", "currency": "GBP", "company_code": "UK02"}), 409, "cannot move")
refused(api("POST", "/bank-accounts/NEW-1", APPROVER, {"name": "x", "currency": "GBP", "company_code": "ZZ99"}), 422, "no company ZZ99")
refused(api("POST", "/bank-accounts/NEW-2", APPROVER, {"name": "x", "currency": "GBP", "purposes": "receive,spend"}), 422, "purposes")
assert api("POST", "/bank-accounts/NEW-3", CASHIER, {"name": "x", "currency": "GBP"})[0] == 403 # pay:bank_account:write
def test_PAY_27_a_statement_for_an_account_posts_only_in_its_own_company(nb):
bank("LLOY-01", gl="1300", account_number="11223344", company_code="UK02")
imp(camt([e("D1", "DBIT", "9.00", text="CHARGE", name="Bank")], stmt_id="L-1"), account="LLOY-01")
must(api("POST", "/statement-lines/bank:LLOY-01:D1/post", CASHIER, {"account_id": "6900", "tax_code": "E"}))
assert journals(nb, "bank_line")[0]["company_code"] == "UK02"
def test_PAY_27_the_default_receiving_account_is_a_dated_row_from_tomorrow(nb):
barclays()
bank("BARC-EUR", currency="EUR", gl="1202", account_number="99887766", company_code="UK01")
body = {"company_code": "UK01", "currency": "GBP", "bank_account_id": "BARC-01"}
refused(api("POST", "/default-bank-accounts", APPROVER, {**body, "valid_from": "2026-11-20"}), 409, "tomorrow")
refused(api("POST", "/default-bank-accounts", APPROVER, {**body, "bank_account_id": "BARC-EUR"}), 409, "EUR")
got = must(api("POST", "/default-bank-accounts", APPROVER, body))
assert got["valid_from"] == "2026-11-21"
assert must(api("GET", "/default-bank-accounts", CASHIER))["defaults"] == [] and must(api("GET", "/default-bank-accounts", CASHIER))["upcoming"][0]["bank_account_id"] == "BARC-01"
assert must(api("GET", "/default-bank-accounts", CASHIER, {"on": "2026-11-21"}))["defaults"][0]["bank_account_id"] == "BARC-01"
refused(api("POST", "/default-bank-accounts", APPROVER, body), 409, "Someone else wrote") # one default per company, currency and date
def test_PAY_27_an_account_closes_from_a_date_after_today_and_keeps_its_history(nb):
barclays()
imp(camt([e("D1", "DBIT", "9.00", text="CHARGE", name="Bank")], stmt_id="B-1"), account="BARC-01")
refused(api("POST", "/bank-accounts/BARC-01/close", APPROVER, {"closed_from": "2026-11-20"}), 409, "after today")
must(api("POST", "/bank-accounts/BARC-01/close", APPROVER, {"closed_from": "2026-11-25"}))
assert must(api("GET", "/statement-lines", CASHIER))["lines"][0]["line_id"] == "bank:BARC-01:D1" # history kept
refused(api("POST", "/default-bank-accounts", APPROVER, {"company_code": "UK01", "currency": "GBP", "bank_account_id": "BARC-01", "valid_from": "2026-11-26"}), 409, "closed")
with on_day("2026-11-26T09:00:00+00:00"):
refused(api("POST", "/statements/import", CASHIER, {"format": "camt053", "content": camt([e("D2", "DBIT", "1.00", text="X")], stmt_id="B-2"),
"bank_account_id": "BARC-01"}), 409, "closed from")
def test_PAY_27_only_a_machine_identity_holding_the_permission_reads_the_account_unmasked(nb):
barclays()
for who in (CASHIER, APPROVER, COLLECTOR, SALES, NOBODY, person("w@x", ["pay:*:*"]), person("w@x", ["*:*:*"]), person("w@x", ["pay:bank_account:*"])):
assert api("GET", DETAILS, who)[0] == 403, who
assert not table(t.pay_bank_account_read) # a refusal leaves no row
got = must(api("GET", DETAILS, SYSTEM))
assert got == {"bank_account_id": "BARC-01", "company_code": "UK01", "account_name": "Dodil Trading Ltd", "sort_code": "20-00-00",
"account_number": "31926819", "iban": "", "bic": "BARCGB22", "currency": "GBP"}
rows = table(t.pay_bank_account_read)
assert len(rows) == 1 and rows[0]["bank_account_id"] == "BARC-01" and rows[0]["caller"] == "system@shop.example" and rows[0]["requested_at"]
must(api("GET", DETAILS, SYSTEM, {"purpose": "Bacs run 2026-12-01"}))
reads = must(api("GET", "/bank-accounts/BARC-01/reads", APPROVER))["reads"]
assert [r["purpose"] for r in reads] == ["bank file", "Bacs run 2026-12-01"] and "31926819" not in str(reads) # two reads at one instant: two rows
def test_PAY_27_a_read_that_cannot_be_logged_is_refused_and_nothing_is_returned(nb):
barclays()
with mock.patch.object(t.AccountRepository, "log_read", side_effect=RuntimeError("audit store down")):
status, body = api("GET", DETAILS, SYSTEM)
assert status == 503 and "31926819" not in str(body) and "logged" in str(body)
assert not table(t.pay_bank_account_read) and not [c for c in table(t.pay_claim) if c["claim_scope"] == "bank_account_read"] # rolled back whole
def test_PAY_27_an_unknown_account_or_one_that_does_not_pay_is_404(nb):
barclays()
bank("DEP-ONLY", gl="1201", account_number="55667788", purposes="receive")
refused(api("GET", "/bank-accounts/NOPE/account-details", SYSTEM), 404)
refused(api("GET", "/bank-accounts/DEP-ONLY/account-details", SYSTEM), 404, "purpose pay")
assert not table(t.pay_bank_account_read)
def test_PAY_27_an_iban_account_is_served_with_its_iban_and_bic(nb):
must(api("POST", "/house-banks/BARC", APPROVER, {"name": "Barclays", "bic": "BARCGB22", "sort_code": "20-00-00"}))
bank("BARC-EUR", currency="EUR", gl="1202", iban="IE29AIBK93115212345678", purposes="pay", house_bank_id="BARC", company_code="IE01")
got = must(api("GET", "/bank-accounts/BARC-EUR/account-details", SYSTEM))
assert (got["iban"], got["bic"], got["account_number"], got["currency"], got["company_code"]) == ("IE29AIBK93115212345678", "BARCGB22", "", "EUR", "IE01")
# ── PAY-28: a deposit batch of cheques ────────────────────────────────────────────────────────────────
CHEQUES = (("1200.00", "INV-D1"), ("800.00", "INV-D2"), ("450.00", "INV-D3"), ("2050.00", "INV-D4"))
def cheques(nb):
bank("BANK1", gl="1200", account_number="12345678", purposes="receive,pay")
ids = []
for n, (amount, inv) in enumerate(CHEQUES, start=1):
nb.invoice(inv, 7311, amount, due_date="2026-11-25")
r = receipt(amount=amount, received_on="2026-11-18", reference=f"CHQ-{n}", method="cheque", bank_account_id="BANK1")
must(allocate(r["payment_id"], [(inv, amount)]))
ids.append(r["payment_id"])
return ids
def batch(ids, slip="004512"):
b = must(api("POST", "/deposit-batches", CASHIER, {"bank_account_id": "BANK1"}))
for pid in ids:
must(api("POST", f"/deposit-batches/{b['batch_id']}/cheques", CASHIER, {"payment_id": pid}))
return b["batch_id"], must(api("POST", f"/deposit-batches/{b['batch_id']}/close", CASHIER, {"slip_no": slip}))
def test_PAY_28_four_cheques_are_one_batch_keyed_by_the_slip_and_one_statement_credit_matches_the_batch(nb):
ids = cheques(nb)
_, closed = batch(ids)
assert (closed["batch_id"], closed["total"], closed["cheque_count"], closed["lifecycle"]) == ("dep:BANK1:004512", "4500.00", 4, "deposited")
got = imp(camt([e("P1", "CRDT", "4500.00", day="2026-11-19", text="PAYING IN 004512", name="Cheques")], stmt_id="C-1"))
assert got["created"] == 0 and got["receipt_ids"] == [] and got["matched_to_deposit_batch"] == 1 # the line is not a receipt again
ln = lines()["bank:BANK1:P1"]
assert (ln["lifecycle"], ln["matched_kind"], ln["matched_ref"]) == ("matched", "deposit_batch", "dep:BANK1:004512")
assert must(api("GET", "/deposit-batches/dep:BANK1:004512", CASHIER))["lifecycle"] == "matched"
assert len(table(t.pay_receipt)) == 4 and all(r["method"] == "cheque" for r in table(t.pay_receipt)) # the four cheques, nothing else
assert all(settle(i)[i]["outstanding"] == "0.00" for _, i in CHEQUES) # each cheque settled its invoice
def test_PAY_28_a_bounced_cheque_reverses_only_its_own_receipt_and_leaves_the_batch_matched(nb):
ids = cheques(nb)
batch(ids)
imp(camt([e("P1", "CRDT", "4500.00", text="PAYING IN", name="Cheques")], stmt_id="C-1"))
must(api("POST", f"/receipts/{ids[1]}/reverse", CASHIER, {"bank_return_ref": "RTN-9", "returned_on": "2026-11-20"}))
assert must(api("GET", "/deposit-batches/dep:BANK1:004512", CASHIER))["lifecycle"] == "matched"
assert settle("INV-D2")["INV-D2"]["outstanding"] == "800.00" and settle("INV-D1")["INV-D1"]["outstanding"] == "0.00"
assert [r["status"] for r in table(t.pay_receipt)].count("reversed") == 1
def test_PAY_28_a_batch_whose_total_differs_from_the_line_is_never_matched(nb):
with on_day("2026-11-18T09:00:00+00:00"):
must(api("POST", "/reconciliation-rules/PAYIN", APPROVER, {"direction": "credit", "text_contains": "PAYING IN", "account_id": "1100", "tax_code": "E", "mode": "suggest"}))
ids = cheques(nb)
batch(ids)
got = imp(camt([e("P1", "CRDT", "4400.00", text="PAYING IN", name="Cheques")], stmt_id="C-1"))
assert "matched_to_deposit_batch" not in got and lines()["bank:BANK1:P1"]["lifecycle"] == "open" # the batch is not guessed
assert lines()["bank:BANK1:P1"]["matched_kind"] != "deposit_batch"
refused(api("POST", "/deposit-batches/dep:BANK1:004512/match", CASHIER, {"line_id": "bank:BANK1:P1"}), 409, "The batch is 4,500.00; the line 4,400.00")
assert must(api("GET", "/deposit-batches/dep:BANK1:004512", CASHIER))["lifecycle"] == "deposited"
def test_PAY_28_a_cheque_is_in_one_batch_and_only_a_live_cheque(nb):
ids = cheques(nb)
bank("BANK2", gl="1210", account_number="87654321")
b1 = must(api("POST", "/deposit-batches", CASHIER, {"bank_account_id": "BANK1"}))["batch_id"]
b2 = must(api("POST", "/deposit-batches", CASHIER, {"bank_account_id": "BANK1"}))["batch_id"]
must(api("POST", f"/deposit-batches/{b1}/cheques", CASHIER, {"payment_id": ids[0]}))
refused(api("POST", f"/deposit-batches/{b2}/cheques", CASHIER, {"payment_id": ids[0]}), 409, "already in batch")
transfer = receipt(reference="TRF-1", bank_account_id="BANK1")
refused(api("POST", f"/deposit-batches/{b1}/cheques", CASHIER, {"payment_id": transfer["payment_id"]}), 409, "not by cheque")
must(api("POST", f"/receipts/{ids[1]}/reverse", CASHIER, {"bank_return_ref": "R1", "returned_on": "2026-11-20"}))
refused(api("POST", f"/deposit-batches/{b1}/cheques", CASHIER, {"payment_id": ids[1]}), 409, "returned")
refused(api("POST", f"/deposit-batches/{b2}/close", CASHIER, {"slip_no": "1"}), 409, "no cheques")
must(api("POST", f"/deposit-batches/{b1}/close", CASHIER, {"slip_no": "001"}))
b3 = must(api("POST", "/deposit-batches", CASHIER, {"bank_account_id": "BANK1"}))["batch_id"]
must(api("POST", f"/deposit-batches/{b3}/cheques", CASHIER, {"payment_id": ids[2]}))
refused(api("POST", f"/deposit-batches/{b3}/close", CASHIER, {"slip_no": "001"}), 409, "already used") # a slip number once per account
def test_PAY_28_a_cheque_added_while_the_batch_closes_collides_on_the_batch_claim(nb):
ids = cheques(nb)
b = must(api("POST", "/deposit-batches", CASHIER, {"bank_account_id": "BANK1"}))["batch_id"]
must(api("POST", f"/deposit-batches/{b}/cheques", CASHIER, {"payment_id": ids[0]}))
with Snapshot():
must(api("POST", f"/deposit-batches/{b}/close", CASHIER, {"slip_no": "7"}))
with mock.patch.object(t.DepositRepository, "get", lambda self, bid: {**table(t.pay_deposit_batch)[0], "lifecycle": "open"} if bid == b else None):
status, body = api("POST", f"/deposit-batches/{b}/cheques", CASHIER, {"payment_id": ids[1]})
assert status == 409 and "Someone else wrote" in str(body)
# ── PAY-29: the cash position and forecast ────────────────────────────────────────────────────────────
def forecast(**q):
return must(api("GET", "/cash-forecast", APPROVER, q or None))
def cur(body, code):
return next(c for c in body["currencies"] if c["currency"] == code)
def test_PAY_29_position_and_weeks_per_currency_with_each_figure_drilling_to_its_documents(nb):
bank("BANK1", gl="1200", account_number="12345678")
bank("BANKE", currency="EUR", gl="1205", account_number="22334455")
imp(camt([e("D1", "DBIT", "10.00", text="CHARGE", name="Bank")], stmt_id="F-1", opening="5010.00", closing="5000.00"), account="BANK1")
imp(camt([e("E1", "DBIT", "1.00", text="CHARGE", name="Bank", ccy="EUR")], stmt_id="F-2", opening="301.00", closing="300.00"), account="BANKE")
receipt(amount="75.00", reference="TRF-9", bank_account_id="BANK1") # an unreconciled entry
nb.invoice("H1", 7311, "500.00", due_date="2026-10-01") # 7311 paid it 50 days late: that is its habit
nb.invoice("INV-A", 7311, "1000.00", due_date="2026-11-25")
nb.invoice("INV-B", 7312, "200.00", currency="EUR", due_date="2026-11-30")
must(allocate(receipt(amount="500.00", reference="TRF-H")["payment_id"], [("H1", "500.00")]))
nb.bills.append({"bill_id": "B1", "bill_no": "BILL-1", "bp_id": 9001, "currency": "GBP", "total": "800.00", "outstanding": "800.00", "due_date": "2026-11-27"})
nb.bills.append({"bill_id": "B2", "bill_no": "BILL-2", "bp_id": 9002, "currency": "EUR", "total": "90.00", "outstanding": "90.00", "due_date": "2026-11-01"})
nb.liabilities.append({"period_key": "26AD", "charge_type": "vat", "outstanding_amount": "2500.00", "due_date": "2027-02-07"})
nb.cis_due.append({"tax_month": "2026-11", "amount": "300.00", "due_date": "2026-12-22"})
f = forecast()
gbp, eur = cur(f, "GBP"), cur(f, "EUR")
assert gbp["position"]["total"] == "5075.00" and eur["position"]["total"] == "300.00" # never summed across currencies
acct = next(i for i in gbp["position"]["items"] if i["ref"] == "BANK1")
assert (acct["closing_balance"], acct["unreconciled_entries"], len(acct["entries"])) == ("5000.00", "75.00", 1)
inv = next(i for w in gbp["weeks"] for i in w["expected_receipts"]["items"] if i["ref"] == "INV-A")
assert inv["expected_on"] == "2027-01-14" and inv["days_late_habit"] == 50 # due 11-25 plus 50 days
assert [i["ref"] for i in cur(f, "EUR")["weeks"][1]["expected_receipts"]["items"]] == ["INV-B"] and eur["weeks"][1]["expected_receipts"]["total"] == "200.00"
assert [i["ref"] for i in gbp["weeks"][1]["expected_payments"]["items"]] == ["B1"]
assert [i["ref"] for i in eur["weeks"][0]["expected_payments"]["items"]] == ["B2"] # overdue: due today at the earliest
kinds = {i["ref"]: i["kind"] for w in gbp["weeks"] for i in w["expected_payments"]["items"]}
assert kinds == {"B1": "bill", "26AD:vat": "vat_liability", "2026-11": "cis_due"}
running = [D(w["running_balance"]) for w in gbp["weeks"]]
assert running[0] == D("5075.00") and running[1] == D("4275.00") and running[-1] == D("5075.00") + D("1000.00") - D("800.00") - D("2800.00")
assert f["unavailable"] == []
def test_PAY_29_direct_debits_count_on_their_collection_dates_and_not_twice_as_invoices(nb):
dd_setup(nb)
with on_day(PREP):
active_mandates()
dd_run("2026-12-01")
gbp = cur(forecast(), "GBP")
items = [i for w in gbp["weeks"] for i in w["expected_receipts"]["items"]]
assert [(i["kind"], i["ref"], i["amount"], i["expected_on"]) for i in items] == [("direct_debit_run", "dd:BARC-01:2026-12-01", "600.00", "2026-12-01")]
def test_PAY_29_a_neighbour_that_cannot_answer_leaves_its_side_out_and_says_so(nb):
bank("BANK1", gl="1200", account_number="12345678")
nb.bills_status = 500
f = forecast()
assert [n["side"] for n in f["unavailable"]] == ["bills"]
def test_PAY_29_it_is_derived_on_read_and_a_snapshot_is_kept_only_when_saved(nb):
bank("BANK1", gl="1200", account_number="12345678")
forecast()
assert not table(t.pay_forecast_snapshot)
saved = must(api("POST", "/cash-forecast/snapshots", APPROVER, {"weeks": 4}))
assert len(table(t.pay_forecast_snapshot)) == 1 and saved["weeks"] == 4
got = must(api("GET", f"/cash-forecast/snapshots/{saved['snapshot_id']}", APPROVER))
assert got["forecast"]["weeks"] == 4 and len(must(api("GET", "/cash-forecast/snapshots", APPROVER))["snapshots"]) == 1
assert api("GET", "/cash-forecast", CASHIER)[0] == 403 # the approver's view (pay:forecast:read)
@pytest.mark.parametrize("weeks", [0, 53, -1])
def test_PAY_29_weeks_outside_one_to_fifty_two_are_refused_422(nb, weeks):
assert api("GET", "/cash-forecast", APPROVER, {"weeks": weeks})[0] == 422
def test_v4_slice_b_amounts_outside_numeric_18_2_are_refused_422(nb):
day_of_cash(nb)
assert entry("expense", "1e15", tax_code="E", account_id="6100")[0] == 422
assert entry("receipt", "10000000000000000", bp_id=7311)[0] == 422
assert api("POST", "/cash-journals/X", CASHIER, {"name": "x", "currency": "GBP", "gl_account_id": "1", "company_code": "UK01", "opening_balance": "1.234"})[0] == 422
assert api("POST", "/cash-journals/PETTY-UK01/count", CASHIER, {"counted_amount": "-1.00"})[0] == 422
def test_PAY_28_an_import_never_guesses_a_batch_for_a_credit_of_another_total(nb):
batch(cheques(nb))
got = imp(camt([e("P1", "CRDT", "4499.99", text="SOMEONE ELSE", name="Other Ltd")], stmt_id="C-1"))
assert got["created"] == 1 and "matched_to_deposit_batch" not in got # an ordinary customer credit
assert must(api("GET", "/deposit-batches/dep:BANK1:004512", CASHIER))["lifecycle"] == "deposited"
def test_v4_slice_b_the_new_use_cases_are_agent_tools_and_the_machine_read_is_not(nb):
from conftest import tool, tools
names = tools(APPROVER)
for want in ("put_bank_account", "close_bank_account", "set_default_bank_account", "reconcile_transfers", "transfers_in_transit", "put_cash_journal",
"open_cash_day", "record_cash_entry", "count_cash", "approve_cash_difference", "close_cash_day", "start_deposit_batch",
"add_cheque_to_batch", "close_deposit_batch", "match_deposit_batch", "cash_forecast", "save_cash_forecast"):
assert want in names, want
assert "get_account_details" not in names # a machine seam, not an agent tool
status, body = tool(APPROVER, "cash_forecast", {"weeks": 2})
assert status == "ok" and body["weeks"] == 2