dodil/erp-brick-payments
DODIL ERP brick: payments
tests/scenarios/catalogue.py
"""pay's scenario catalogue (module.yaml tests.scenarios, SC-PAY-01 ... SC-PAY-09), each mapped to its stories and written
ONCE over the composition's doors - routes and MCP tools - never a table. Invoicing (ar) and the ledger (gl) are stood in
by tests/fakes.py, registered as local seams; the partner register is the real business-partner brick. Each scenario makes
its own customer and documents (s.key) and sets the business clock, so none leans on another.
erp-scenarios list tests/scenarios/catalogue.py
erp-scenarios run tests/scenarios/catalogue.py
"""
import pathlib
import sys
from erp_platform import seam
from erp_platform.scenarios import Catalogue
HERE = pathlib.Path(__file__).resolve().parent
sys.path.insert(0, str(HERE.parent))
from fakes import Neighbours # noqa: E402
NB = Neighbours()
def install() -> None:
"""Stand invoicing and the ledger in (the partner register is the real brick). At import for the CLI; again from the
pytest test, because the unit tests' own fakes unregister theirs when they end."""
seam.register_local("ar", NB.ar)
seam.register_local("gl", NB.gl)
install()
CAT = Catalogue("payments", compose=HERE / "compose.yaml", roles=HERE / "roles.yaml", zone="Europe/London",
ticks=["POST /payments/collections/tick"], tick_as="system")
# Generated over the composition's doors, both HTTP and MCP: every role x every use case - 401 with no identity, 403
# without the permission, served with it - and the wildcard probes (`*:*:*` and `pay:*:*` run every person use case
# and no machine one).
CAT.permission_matrix()
TODAY = "2026-11-20T09:00:00Z"
def customer(s):
"""The partner steward adds a customer to the register; its bp_id."""
return s.as_("partner_steward").put("/business-partner/partners", {"bp_key": s.key("CUST"), "legal_name": "Northwind Ltd", "country": "GB",
"roles": ["customer"]}).ok().value("bp_id")
def camt(entries):
body = "".join(f"<Ntry><NtryRef>{r}</NtryRef><Amt Ccy='GBP'>{a}</Amt><CdtDbtInd>CRDT</CdtDbtInd><BookgDt><Dt>{d}</Dt></BookgDt></Ntry>"
for r, a, d in entries)
return f"<Document xmlns='urn:iso:std:iso:20022:tech:xsd:camt.053.001.02'><BkToCstmrStmt><Stmt>{body}</Stmt></BkToCstmrStmt></Document>"
@CAT.scenario("SC-PAY-01", "one transfer, three invoices", stories=["PAY-01", "PAY-32", "PAY-03", "PAY-30"], tags=["money"],
personas=["cashier", "partner_steward"])
def sc_pay_01(s):
cash = s.as_("cashier")
s.clock.set(TODAY)
bp = customer(s)
n = s.key("INV")
a, b, c = (NB.invoice(f"{n}-{i}", int(bp), amt, number=f"{n}-{i}", due_date="2026-11-10") for i, amt in
((41, "4000.00"), (42, "3500.00"), (43, "3000.00")))
s.given("invoices of 4,000.00, 3,500.00 and 3,000.00; a receipt of 10,000.00 with a reference naming the first two")
got = cash.post("/payments/receipts", {"bp_id": bp, "amount": "10000.00", "currency": "GBP", "received_on": "2026-11-20",
"bank_reference": f"{n}-41 {n}-42"}).ok()
pid = got.value("payment_id")
s.when("the match is suggested, confirmed, and the remainder allocated to the third")
got.field("suggestion.proposed_lines.0.invoice_id", a).field("suggestion.proposed_lines.1.invoice_id", b)
cash.post(f"/payments/receipts/{pid}/decide-match", {"decision": "accept"}).ok().field("allocated", "7500.00")
cash.post(f"/payments/receipts/{pid}/allocate", {"lines": [{"invoice_id": c, "amount": "2500.00"}]}).ok()
s.then("the first two paid, the third with 500.00 outstanding, 0.00 unallocated")
done = cash.get("/payments/settlement", invoice_ids=f"{a},{b},{c}").ok()
done.field("invoices.0.outstanding", "0.00").field("invoices.1.outstanding", "0.00").field("invoices.2.outstanding", "500.00")
cash.get(f"/payments/receipts/{pid}").ok().field("unallocated", "0.00")
@CAT.scenario("SC-PAY-02", "the statement imported twice", stories=["PAY-02"], tags=["money"], personas=["cashier", "approver", "partner_steward"])
def sc_pay_02(s):
cash = s.as_("cashier")
s.clock.set(TODAY)
acct = s.key("BANK")
s.as_("approver").post(f"/payments/bank-accounts/{acct}", {"name": "Current", "currency": "GBP", "gl_account_id": "1200"}).ok()
s.given("a camt.053 file of 42 credit lines")
file = camt([(f"{acct}-{i}", "10.00", "2026-11-18") for i in range(42)])
s.when("it is imported twice")
first = cash.post("/payments/receipts/import", {"format": "camt053", "content": file, "bank_account_id": acct}).ok().field("created", 42)
second = cash.post("/payments/receipts/import", {"format": "camt053", "content": file, "bank_account_id": acct}).ok()
s.then("42 receipts, not 84")
second.field("created", 0).field("skipped_duplicate", 42)
first.field("lines", 42)
cash.get("/payments/receipts", limit=1000).ok().matches("receipts", lambda r: len([x for x in r if x["bank_account_id"] == acct]) == 42, "42 receipts")
@CAT.scenario("SC-PAY-03", "bounced payment", stories=["PAY-07", "PAY-30"], tags=["money"], personas=["cashier", "partner_steward"])
def sc_pay_03(s):
cash = s.as_("cashier")
s.clock.set(TODAY)
bp, n = customer(s), s.key("INV")
a, b = NB.invoice(f"{n}-A", int(bp), "700.00"), NB.invoice(f"{n}-B", int(bp), "500.00")
s.given("a receipt of 1,200.00 allocated to two invoices")
pid = cash.post("/payments/receipts", {"bp_id": bp, "amount": "1200.00", "currency": "GBP", "received_on": "2026-11-18"}).ok().value("payment_id")
cash.post(f"/payments/receipts/{pid}/allocate", {"lines": [{"invoice_id": a, "amount": "700.00"}, {"invoice_id": b, "amount": "500.00"}]}).ok()
s.when("it is reversed")
cash.post(f"/payments/receipts/{pid}/reverse", {"bank_return_ref": "RTN-7", "returned_on": "2026-11-19"}).ok().field("status", "reversed")
s.then("both invoices outstanding again; the receipt reversed; allocations net to 0.00")
cash.get("/payments/settlement", invoice_ids=f"{a},{b}").ok().field("invoices.0.outstanding", "700.00").field("invoices.1.outstanding", "500.00")
cash.get(f"/payments/receipts/{pid}").ok().field("allocation_state", "reversed").field("unallocated", "0.00") \
.matches("allocations", lambda rows: sum(float(r["amount"]) for r in rows) == 0.0, "allocations net to zero")
@CAT.scenario("SC-PAY-04", "overpayment refunded", stories=["PAY-05", "PAY-08"], tags=["money", "permissions"], personas=["cashier", "approver", "approver#2", "partner_steward"])
def sc_pay_04(s):
cash, owner, owner2 = s.as_("cashier"), s.as_("approver"), s.as_("approver", 2)
s.clock.set(TODAY)
bp, n = customer(s), s.key("INV")
inv = NB.invoice(f"{n}-1", int(bp), "1000.00")
s.given("a receipt of 1,300.00 against an invoice of 1,000.00")
pid = cash.post("/payments/receipts", {"bp_id": bp, "amount": "1300.00", "currency": "GBP", "received_on": "2026-11-20"}).ok().value("payment_id")
cash.post(f"/payments/receipts/{pid}/allocate", {"lines": [{"invoice_id": inv, "amount": "1000.00"}]}).ok()
s.when("300.00 is requested back, approved by another person, and paid")
rid = cash.post("/payments/refunds", {"bp_id": bp, "amount": "300.00", "currency": "GBP"}).ok().value("refund_id")
cash.post(f"/payments/refunds/{rid}/approve", {}).forbidden("pay:refund:approve")
owner.post(f"/payments/refunds/{rid}/approve", {}).ok().field("status", "approved")
cash.post(f"/payments/refunds/{rid}/pay", {"bank_ref": "BNK-1"}).ok().field("status", "paid")
s.then("customer credit 0.00; the same person approving their own request is refused")
cash.get(f"/payments/receipts/{pid}").ok().field("unallocated", "0.00")
cash.post("/payments/refunds", {"bp_id": bp, "amount": "0.01", "currency": "GBP"}).refused(409, "unallocated credit in GBP is 0.00")
ask = owner.post("/payments/refunds", {"bp_id": bp, "amount": "5.00", "currency": "GBP"})
ask.forbidden("pay:refund:request") # an approver holds approve, not request: nobody both asks and approves here
owner2.post("/payments/refunds/none/approve", {}).not_found()
@CAT.scenario("SC-PAY-05", "reminders with a promise", stories=["PAY-31", "PAY-10"], tags=["clock"], personas=["approver", "collector", "partner_steward"])
def sc_pay_05(s):
owner, coll = s.as_("approver"), s.as_("collector")
s.clock.set(TODAY)
bp, n = customer(s), s.key("INV")
for lv, days in ((1, 7), (2, 14), (3, 30)):
owner.post("/payments/reminder-levels", {"level": {"level": lv, "days_after_due": days, "mode": "suggest"}}).ok()
s.given("an invoice 7 days overdue, levels 7/14/30, mode suggest")
inv = NB.invoice(f"{n}-1", int(bp), "100.00", due_date="2026-11-13")
s.when("the daily pass runs twice, then the customer promises payment by day 20")
s.tick()
s.tick()
mine = lambda r: [d for d in r if d["invoice_id"] == inv] # noqa: E731
coll.get("/payments/collections/reminders").ok().matches("drafts", lambda d: [x["level"] for x in mine(d)] == [1], "one level-1 draft")
coll.post("/payments/collections/promises", {"bp_id": bp, "promise_date": "2026-12-03"}).ok()
s.then("no level-2 on day 14 (suppressed by the promise); level 2 on day 21 if still unpaid")
s.clock.set("2026-11-27T09:00:00Z")
s.tick()
coll.get("/payments/collections/reminders").ok().matches("drafts", lambda d: [x["level"] for x in mine(d)] == [1], "still only level 1")
s.clock.set("2026-12-04T09:00:00Z")
s.tick()
coll.get("/payments/collections/reminders").ok().matches("drafts", lambda d: sorted(x["level"] for x in mine(d)) == [1, 2], "level 2 now")
@CAT.scenario("SC-PAY-06", "credit hold reaches sales-order", stories=["PAY-12", "PAY-40"], tags=["permissions"], personas=["sales", "approver", "partner_steward"])
def sc_pay_06(s):
sales, owner = s.as_("sales"), s.as_("approver")
s.clock.set(TODAY)
bp, n = customer(s), s.key("INV")
s.given("a hold rule of 60 days overdue")
NB.invoice(f"{n}-1", int(bp), "100.00", due_date="2026-09-20")
s.when("an invoice reaches 61 days")
s.then("credit_status says hold with the reason; sales-order refuses (or warns) on confirm")
sales.get(f"/payments/customers/{bp}/credit-status").ok().field("hold", True).field("hold_mode", "warn") \
.matches("reason", lambda r: "61 days overdue" in r, "the reason names the days")
owner.post("/payments/controls/credit_hold_mode", {"value": "block"}).ok()
sales.get(f"/payments/customers/{bp}/credit-ok").ok().field("gate", "credit_ok").field("met", False).field("hold_mode", "block")
sales.get("/payments/receipts").forbidden("pay:receipt:read") # sales sees the hold, never a receipt
@CAT.scenario("SC-PAY-07", "no back-dating, corrections are reversals", stories=["PAY-01", "PAY-06", "PAY-11", "PAY-12"], tags=["clock", "money"],
personas=["cashier", "collector", "approver", "partner_steward"])
def sc_pay_07(s):
cash, coll, owner = s.as_("cashier"), s.as_("collector"), s.as_("approver")
s.clock.set("2026-11-30T12:00:00Z")
bp, n = customer(s), s.key("INV")
a = NB.invoice(f"{n}-A", int(bp), "4500.00", issued_on="2026-11-01")
b = NB.invoice(f"{n}-B", int(bp), "300.00", issued_on="2026-12-01")
s.given("a statement sent as at 2026-11-30 with a closing balance of 4,500.00, today 2026-12-03 in Europe/London")
coll.post("/payments/statements", {"bp_id": bp, "as_at": "2026-11-30"}).ok().field("closing_balance.GBP", "4500.00")
s.clock.set("2026-12-03T09:00:00Z")
s.when("a receipt dated 2026-12-04, an allocation dated 2026-11-29 and a credit limit from today are tried; then a wrong allocation is undone")
cash.post("/payments/receipts", {"bp_id": bp, "amount": "10.00", "currency": "GBP", "received_on": "2026-12-04"}).refused(409, "which is after today (2026-12-03, Europe/London)")
owner.post(f"/payments/customers/{bp}/credit-limit", {"amount": "9000.00", "currency": "GBP", "valid_from": "2026-12-03"}).refused(409, "from tomorrow (2026-12-04) at the earliest")
pid = cash.post("/payments/receipts", {"bp_id": bp, "amount": "300.00", "currency": "GBP", "received_on": "2026-12-02"}).ok().value("payment_id")
cash.post(f"/payments/receipts/{pid}/allocate", {"lines": [{"invoice_id": b, "amount": "300.00"}]}).ok().field("allocations.0.allocated_on", "2026-12-03")
s.then("the three are refused with their readiness reasons; the undo is a reversing row dated 2026-12-03; the statement as at 2026-11-30 still re-computes to 4,500.00")
cash.post(f"/payments/receipts/{pid}/unallocate", {"invoice_id": b, "seq": 1, "reason": "wrong invoice"}).ok() \
.field("reversal.allocated_on", "2026-12-03").field("reversal.reverses_seq", 1)
coll.post("/payments/statements", {"bp_id": bp, "as_at": "2026-11-30"}).ok().field("closing_balance.GBP", "4500.00") \
.field("sent.0.matches_recorded", True)
del a
@CAT.scenario("SC-PAY-08", "the AI suggests, the cashier decides", stories=["PAY-32", "PAY-14", "PAY-13"], tags=["permissions", "money"],
personas=["cashier", "cashier_admin", "approver", "partner_steward"])
def sc_pay_08(s):
cash, clerk, owner = s.as_("cashier"), s.as_("cashier_admin"), s.as_("approver")
s.clock.set(TODAY)
bp, n = customer(s), s.key("INV")
ids = [NB.invoice(f"{n}-{i}", int(bp), amt, number=f"{n}-{i}") for i, amt in ((11, "100.00"), (12, "200.00"), (13, "300.00"))]
s.given("mode suggest; three receipts with suggestions")
pids = [cash.post("/payments/receipts", {"bp_id": bp, "amount": amt, "currency": "GBP", "received_on": "2026-11-20",
"bank_reference": f"{n}-{i}"}).ok().value("payment_id") for i, amt in ((11, "100.00"), (12, "200.00"), (13, "300.00"))]
s.when("one is accepted, one changed, one rejected; the cashier then tries to switch on auto-with-review")
cash.post(f"/payments/receipts/{pids[0]}/decide-match", {"decision": "accept"}).ok().field("allocations.0.decided_how", "accepted")
cash.post(f"/payments/receipts/{pids[1]}/decide-match", {"decision": "change", "lines": [{"invoice_id": ids[1], "amount": "150.00"}]}).ok() \
.field("allocations.0.decided_how", "changed")
cash.post(f"/payments/receipts/{pids[2]}/decide-match", {"decision": "reject"}).ok().field("decision", "rejected")
s.then("two allocations with decided_how accepted / changed citing their suggestions, one receipt unallocated; the switch is refused, the owner's is accepted and recorded")
cash.get(f"/payments/receipts/{pids[2]}").ok().field("unallocated", "300.00")
clerk.post("/payments/settings/ai_suggest_match", {"value": "auto-with-review"}).forbidden("pay:settings:controls")
owner.post("/payments/controls/ai_suggest_match", {"value": "auto-with-review"}).ok()
owner.get("/payments/settings-history", setting_key="ai_suggest_match").ok().field("history.0.to", "auto-with-review")
@CAT.scenario("SC-PAY-09", "Odoo migration reconciles to zero unexplained", stories=["PAY-44", "PAY-15"], tags=["money", "permissions"],
personas=["system", "approver", "cashier"])
def sc_pay_09(s):
from odoo_fake import CUTOVER, DB, SUSPENSE, FakeOdoo, FakePartners, load_ar
from brick_payments.infrastructure import odoo
system, owner, cash = s.as_("system"), s.as_("approver"), s.as_("cashier")
s.clock.set(TODAY)
s.given("an Odoo database with posted customer payments, partial and full reconciliations, two unreconciled bank lines, one customer refund and one cross-currency payment; ar migrated first")
load_ar(NB)
odoo.use(odoo.Ports(FakeOdoo(), FakePartners()))
body = {"db": DB, "cutover_date": CUTOVER, "suspense_bp_id": SUSPENSE}
cash.post("/payments/migration/odoo/run", body).forbidden("pay:migration:import")
s.when("the migration runs twice into a test copy and the reconciliation is opened")
system.post("/payments/migration/odoo/run", body).ok().field("receipts", 7).field("allocations", 6)
system.post("/payments/migration/odoo/run", body).ok().field("receipts", 0).field("allocations", 0).field("refunds", 0)
s.then("every 'equal' check in migration.reconcile is equal per customer and currency; the cross-currency payment is the one explained difference; the second run added no rows; the owner signs")
rec = owner.get("/payments/migration/reconcile").ok().field("blocking", 0).field("can_sign", True).field("explained_differences", 1)
rec.matches("checks", lambda rows: [r["check"] for r in rows if not r["equal"]] == ["cross-currency payments"], "only the cross-currency payment differs")
owner.post("/payments/migration/sign", {}).ok().field("signed.by", "approver.1@scenario.test")
system.post("/payments/migration/odoo/run", body).refused(409, "read-only")
odoo.use(None)