dodil/erp-brick-payments
DODIL ERP brick: payments
tests/test_v4_directdebit.py
"""Payments v4, slice A - PAY-18 (a Direct Debit Instruction), PAY-19 (a run), PAY-20 (the bank's ARUDD / ADDACS / DDICA reports)."""
import base64
import json
from unittest import mock
from conftest import APPROVER, CASHIER, ROLES, SYSTEM, api, bank, must, on_day, refused, settle
from sqlalchemy import select
from brick_payments.application import use_cases
from brick_payments.infrastructure import tables as t
PREP = "2026-11-10T09:00:00+00:00" # 15 working days before the collection: the notices are in time
COLLECT = "2026-12-01T08:00:00+00:00"
def table(tbl):
with use_cases.repository() as repo:
return [dict(m) for m in repo.conn.execute(select(tbl)).mappings()]
def ident(sub, email, permissions):
payload = base64.urlsafe_b64encode(json.dumps({"permissions": list(permissions)}).encode()).decode()
user = {"sub": sub, "app_roles": [], **({"email": email} if email else {})}
return {"X-Dodil-User": json.dumps(user), "X-Dodil-User-JWT": f"h.{payload.rstrip('=')}.s"}
def setup(nb, customers=(7311, 7312, 7313)):
must(api("POST", "/house-banks/BARC", APPROVER, {"name": "Dodil Trading Ltd", "sort_code": "20-00-00", "sun": "123456", "bic": "BARCGB22"}))
bank("BARC-01", gl="1200", account_number="12345678", purposes="receive,dd_collect", house_bank_id="BARC")
for n, bp in enumerate(customers, start=1):
nb.invoice(f"INV-{bp}", bp, f"{n}00.00", number=f"INV-{bp}", due_date="2026-11-25")
def mandate(bp=7311, ref=None, **more):
body = {"bp_id": bp, "mandate_ref": ref or f"DOD-{bp}-01", "bank_account_id": "BARC-01", "payer_sort_code": "20-00-00",
"payer_account_number": "12345678", "payer_account_name": f"Customer {bp}", "payer_confirmed": True, **more}
return api("POST", "/dd-mandates", CASHIER, body)
def active_mandates(bps=(7311, 7312, 7313)):
for bp in bps:
must(mandate(bp))
return must(api("POST", "/dd-mandates/lodge", CASHIER, {"bank_account_id": "BARC-01", "submission_ref": "AUD-1"}))
BOTH = ident("u-ann", "ann@shop.example", [*ROLES["cashier"], *ROLES["approver"]])
BOTH_BY_SUB = ident("u-ann", None, [*ROLES["cashier"], *ROLES["approver"]])
def run(date="2026-12-01"):
return must(api("POST", "/dd-runs", CASHIER, {"bank_account_id": "BARC-01", "collection_date": date}))
def to_submitted(rid="dd:BARC-01:2026-12-01"):
must(api("POST", f"/dd-runs/{rid}/approve", APPROVER))
must(api("POST", f"/dd-runs/{rid}/notices", CASHIER))
return must(api("POST", f"/dd-runs/{rid}/submit", CASHIER, {"submission_ref": "BACS-77"}))
def collect(rid="dd:BARC-01:2026-12-01"):
with on_day(COLLECT):
return must(api("POST", f"/dd-runs/{rid}/collect", SYSTEM))
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):
with use_cases.repository() as repo:
self.counts = {}
for c in repo.conn.execute(select(t.pay_claim)).mappings():
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-18: a mandate ────────────────────────────────────────────────────────────────────────────────
def test_PAY_18_a_mandate_is_pending_with_its_bank_details_masked(nb):
setup(nb)
got = must(mandate(7311))
assert (got["mandate_ref"], got["sun"], got["lifecycle"], got["payer_account_masked"]) == ("DOD-7311-01", "123456", "pending", "****5678")
assert "12345678" not in json.dumps(got) # the account number is never in a view
assert "12345678" not in json.dumps(must(api("GET", "/dd-mandates", CASHIER)))
assert [(r["sort_code"], r["account_number"]) for r in table(t.pay_dd_mandate_details)] == [("200000", "12345678")]
def test_PAY_18_the_reference_is_unique_per_sun_6_to_18_characters_and_needs_the_payers_confirmation(nb):
setup(nb)
must(mandate(7311))
refused(mandate(7312, ref="DOD-7311-01"), 409, "already used for SUN 123456")
assert mandate(7312, ref="DOD1")[0] == 422
refused(mandate(7312, payer_confirmed=False), 409, "Direct Debit Guarantee")
refused(mandate(7312, payer_sort_code="20-00"), 409, "six digits")
assert mandate(7312, notice_days=0)[0] == 422
def test_PAY_18_a_bank_account_without_a_sun_or_not_collecting_refuses_the_mandate(nb):
bank("BARC-01", gl="1200", purposes="receive")
refused(mandate(7311), 409, "does not collect direct debits")
must(api("POST", "/house-banks/NOSUN", APPROVER, {"name": "Bank"}))
bank("BARC-02", gl="1201", purposes="dd_collect", house_bank_id="NOSUN")
refused(mandate(7311, bank_account_id="BARC-02"), 409, "no Service User Number")
def test_PAY_18_lodging_makes_it_active_and_a_rejection_records_the_reason_and_leaves_it_unusable(nb):
setup(nb)
must(mandate(7311))
must(mandate(7312))
got = must(api("POST", "/dd-mandates/lodge", CASHIER, {"bank_account_id": "BARC-01", "submission_ref": "AUD-1"}))
assert sorted(got["lodged_0N"]) == ["DOD-7311-01", "DOD-7312-01"]
file = [o for o in table(t.pay_object)][0]["body"]
assert file.count("0N") >= 2 and "UHL1" in file and "12345678" in file # the full details are in the Bacs file, and only there
assert {m["mandate_ref"]: m["lifecycle"] for m in must(api("GET", "/dd-mandates", CASHIER))["mandates"]} == {"DOD-7311-01": "active", "DOD-7312-01": "active"}
rej = must(api("POST", "/dd-mandates/DOD-7312-01/reject", CASHIER, {"reason": "no account"}))
assert (rej["lifecycle"], rej["auddis_result"]) == ("cancelled", "no account")
with on_day(PREP):
assert [c["mandate_ref"] for c in run()["collections"]] == ["DOD-7311-01"] # the rejected one is never collected
def test_PAY_18_cancelling_sends_0c_and_a_cancelled_mandate_is_never_collected(nb):
setup(nb, (7311, 7312))
active_mandates((7311, 7312))
got = must(api("POST", "/dd-mandates/DOD-7312-01/cancel", CASHIER, {"reason": "customer asked"}))
assert got["lifecycle"] == "cancelled" and "0C" in got["next"]
again = must(api("POST", "/dd-mandates/DOD-7312-01/cancel", CASHIER, {"reason": "again"}))
assert again["unchanged"] is True
with on_day(PREP):
assert [c["mandate_ref"] for c in run()["collections"]] == ["DOD-7311-01"]
sub = to_submitted()
assert sub["cancelled_0C"] == ["DOD-7312-01"]
recs = must(api("GET", "/dd-runs/dd:BARC-01:2026-12-01/file", CASHIER))["content"].split("\n")
assert sorted(r[15:17] for r in recs if r and not r.startswith(("UHL1", "UTL1"))) == ["01", "0C"]
def test_PAY_18_a_never_lodged_mandate_cancels_with_nothing_to_send(nb):
setup(nb)
must(mandate(7311))
got = must(api("POST", "/dd-mandates/DOD-7311-01/cancel", CASHIER, {"reason": "changed mind"}))
assert "never lodged" in got["next"]
assert must(api("POST", "/dd-mandates/lodge", CASHIER, {"bank_account_id": "BARC-01", "submission_ref": "A"}))["unchanged"] is True
def test_PAY_18_a_mandate_with_no_collection_for_13_months_is_dormant_and_needs_a_new_instruction(nb):
setup(nb)
active_mandates((7311,))
with use_cases.repository() as repo:
m = repo.dd.mandate("123456", "DOD-7311-01")
repo.dd.save_mandate({**m, "lodged_on": __import__("datetime").date(2025, 9, 1), "recorded_on": __import__("datetime").date(2025, 9, 1)})
assert must(api("POST", "/dd-mandates/mark-dormant", SYSTEM))["dormant"] == ["DOD-7311-01"]
assert must(api("GET", "/dd-mandates", CASHIER))["mandates"][0]["lifecycle"] == "dormant"
assert must(api("POST", "/dd-mandates/mark-dormant", SYSTEM))["dormant"] == [] # final: nothing moves again
refused(api("POST", "/dd-mandates/DOD-7311-01/cancel", CASHIER, {"reason": "x"}), 409, "dormant")
# ── PAY-19: a run ────────────────────────────────────────────────────────────────────────────────────
def test_PAY_19_a_run_names_each_invoice_and_is_keyed_by_account_and_collection_date(nb):
setup(nb)
active_mandates()
with on_day(PREP):
got = run()
assert got["run_id"] == "dd:BARC-01:2026-12-01" and got["lifecycle"] == "draft" and got["input_day"] == "2026-11-27"
assert [(c["mandate_ref"], c["amount"], c["transaction_code"], c["invoice_ids"]) for c in got["collections"]] == [
("DOD-7311-01", "100.00", "01", ["INV-7311"]), ("DOD-7312-01", "200.00", "01", ["INV-7312"]), ("DOD-7313-01", "300.00", "01", ["INV-7313"])]
assert (got["total"], got["count"]) == ("600.00", 3)
def test_PAY_19_the_collection_date_must_be_a_future_working_day(nb):
setup(nb)
active_mandates()
with on_day(PREP):
for d in ("2026-11-28", "2026-12-25", "2026-11-10"): # a Saturday, Christmas Day, today
refused(api("POST", "/dd-runs", CASHIER, {"bank_account_id": "BARC-01", "collection_date": d}), 409, "")
assert api("POST", "/dd-runs", CASHIER, {"bank_account_id": "BARC-01", "collection_date": "tomorrow"})[0] == 422
def test_PAY_19_the_approver_may_not_be_the_preparer_under_either_identity(nb):
setup(nb)
active_mandates()
with on_day(PREP):
must(api("POST", "/dd-runs", BOTH, {"bank_account_id": "BARC-01", "collection_date": "2026-12-01"}))
refused(api("POST", "/dd-runs/dd:BARC-01:2026-12-01/approve", BOTH, {}), 409, "You prepared this run")
refused(api("POST", "/dd-runs/dd:BARC-01:2026-12-01/approve", BOTH_BY_SUB, {}), 409, "You prepared this run") # the same person by sub only
assert api("POST", "/dd-runs/dd:BARC-01:2026-12-01/approve", CASHIER, {})[0] == 403 # needs the approve permission
assert must(api("POST", "/dd-runs/dd:BARC-01:2026-12-01/approve", APPROVER, {}))["lifecycle"] == "approved"
def test_PAY_19_a_notice_sent_late_moves_the_invoice_to_the_next_run(nb):
setup(nb)
active_mandates()
must(api("POST", "/dd-mandates/DOD-7313-01/cancel", CASHIER, {"reason": "x"}))
with on_day("2026-11-20T09:00:00+00:00"): # 7 working days before: late for 10, in time for the 3 agreed in a DDI
with use_cases.repository() as repo:
m = repo.dd.mandate("123456", "DOD-7312-01")
repo.dd.save_mandate({**m, "notice_days": 3})
got = run()
must(api("POST", f"/dd-runs/{got['run_id']}/approve", APPROVER))
notices = must(api("POST", f"/dd-runs/{got['run_id']}/notices", CASHIER))
assert notices["notices_sent"] == ["DOD-7312-01"] and [d["mandate_ref"] for d in notices["deferred_to_next_run"]] == ["DOD-7311-01"]
assert notices["deferred_to_next_run"][0]["invoice_ids"] == ["INV-7311"] and (notices["total"], notices["count"]) == ("200.00", 1)
again = must(api("POST", f"/dd-runs/{got['run_id']}/notices", CASHIER))
assert again["notices_sent"] == [] and again["deferred_to_next_run"] == [] # idempotent
nxt = must(api("POST", "/dd-runs", CASHIER, {"bank_account_id": "BARC-01", "collection_date": "2026-12-15"}))
assert [c["invoice_ids"] for c in nxt["collections"]] == [["INV-7311"]] # it waits for the next run
def test_PAY_19_an_invoice_cannot_be_in_two_runs(nb):
setup(nb)
active_mandates((7311,))
with on_day(PREP):
run()
refused(api("POST", "/dd-runs", CASHIER, {"bank_account_id": "BARC-01", "collection_date": "2026-12-15"}), 409, "Nothing to collect")
assert [(r["claim_key"], r["n"]) for r in table(t.pay_claim) if r["claim_scope"] == "ddinv"] == [("INV-7311", 1)]
def test_PAY_19_two_runs_racing_for_one_invoice_collide_on_its_claim(nb):
setup(nb)
active_mandates((7311,))
with on_day(PREP):
with Snapshot(): # both preparers began before either committed: neither saw the invoice taken
run("2026-12-01")
with mock.patch.object(t.DirectDebitRepository, "invoices_in_flight", lambda self: {}):
status, body = api("POST", "/dd-runs", CASHIER, {"bank_account_id": "BARC-01", "collection_date": "2026-12-15"})
assert status == 409 and "Someone else wrote" in str(body)
def test_PAY_19_a_run_prepared_or_submitted_is_never_rebuilt(nb):
setup(nb)
active_mandates()
with on_day(PREP):
run()
refused(api("POST", "/dd-runs", CASHIER, {"bank_account_id": "BARC-01", "collection_date": "2026-12-01"}), 409, "is already draft")
to_submitted()
refused(api("POST", "/dd-runs", CASHIER, {"bank_account_id": "BARC-01", "collection_date": "2026-12-01"}), 409, "is already submitted")
refused(api("POST", "/dd-runs/dd:BARC-01:2026-12-01/submit", CASHIER, {"submission_ref": "again"}), 409, "never rebuilt")
assert len(table(t.pay_object)) == 2 # the AUDDIS file and the run's file, no third
def test_PAY_19_submit_needs_the_notices_and_the_input_day_and_freezes_the_run_with_a_standard_18_file(nb):
setup(nb)
active_mandates()
with on_day(PREP):
run()
must(api("POST", "/dd-runs/dd:BARC-01:2026-12-01/approve", APPROVER))
refused(api("POST", "/dd-runs/dd:BARC-01:2026-12-01/submit", CASHIER, {"submission_ref": "S"}), 409, "Advance notice has not gone")
must(api("POST", "/dd-runs/dd:BARC-01:2026-12-01/notices", CASHIER))
with on_day("2026-11-30T09:00:00+00:00"):
refused(api("POST", "/dd-runs/dd:BARC-01:2026-12-01/submit", CASHIER, {"submission_ref": "S"}), 409, "input day")
with on_day("2026-11-27T09:00:00+00:00"):
got = must(api("POST", "/dd-runs/dd:BARC-01:2026-12-01/submit", CASHIER, {"submission_ref": "BACS-77"}))
assert (got["file_format"], got["collections"], got["total"], got["file_key"]) == ("Bacs Standard 18", 3, "600.00", "dd/dd:BARC-01:2026-12-01.bacs")
body = must(api("GET", "/dd-runs/dd:BARC-01:2026-12-01/file", CASHIER))["content"]
recs = [ln for ln in body.split("\n") if ln and not ln.startswith(("UHL1", "UTL1"))]
assert len(recs) == 3 and all(len(r) == 100 for r in recs)
assert [r[15:17] for r in recs] == ["01", "01", "01"] and [int(r[35:46]) for r in recs] == [10000, 20000, 30000]
assert recs[0][:14] == "20000012345678" and "UTL1 0000000060000 0000003 0000003" in body
assert must(api("GET", "/dd-runs/dd:BARC-01:2026-12-01", CASHIER))["file_key"] == "dd/dd:BARC-01:2026-12-01.bacs"
def test_PAY_19_a_euro_account_submits_sepa_pain_008(nb):
must(api("POST", "/house-banks/DEU", APPROVER, {"name": "Dodil GmbH", "sun": "", "creditor_id": "DE98ZZZ09999999999", "bic": "DEUTDEFF"}))
bank("EUR-01", gl="1300", currency="EUR", account_number="DE89370400440532013000", purposes="receive,dd_collect", house_bank_id="DEU")
nb.invoice("INV-E1", 7311, "100.00", currency="EUR", due_date="2026-11-25")
must(api("POST", "/dd-mandates", CASHIER, {"bp_id": 7311, "mandate_ref": "SEPA-7311-01", "bank_account_id": "EUR-01", "payer_sort_code": "-",
"payer_account_number": "DE02120300000000202051", "payer_account_name": "Kunde", "payer_confirmed": True,
"scheme": "sepa_core"}))
must(api("POST", "/dd-mandates/lodge", CASHIER, {"bank_account_id": "EUR-01", "submission_ref": "x"})) # SEPA: no scheme lodging, active at once
with on_day(PREP):
must(api("POST", "/dd-runs", CASHIER, {"bank_account_id": "EUR-01", "collection_date": "2026-12-01"}))
must(api("POST", "/dd-runs/dd:EUR-01:2026-12-01/approve", APPROVER))
must(api("POST", "/dd-runs/dd:EUR-01:2026-12-01/notices", CASHIER))
got = must(api("POST", "/dd-runs/dd:EUR-01:2026-12-01/submit", CASHIER, {"submission_ref": "SEPA-1"}))
assert got["file_format"] == "SEPA pain.008"
body = must(api("GET", "/dd-runs/dd:EUR-01:2026-12-01/file", CASHIER))["content"]
assert "pain.008" in body and "<MndtId>SEPA-7311-01</MndtId>" in body and "DE98ZZZ09999999999" in body and "<CtrlSum>100.00</CtrlSum>" in body
def test_PAY_19_on_the_entry_day_each_collection_is_a_receipt_allocated_with_decided_how_direct_debit(nb):
setup(nb)
active_mandates()
with on_day(PREP):
run()
to_submitted()
with on_day("2026-11-30T08:00:00+00:00"):
refused(api("POST", "/dd-runs/dd:BARC-01:2026-12-01/collect", SYSTEM), 409, "entry day")
got = collect()
assert got["collected"] == 3 and got["lifecycle"] == "collected"
r = must(api("GET", "/receipts/dd:dd:BARC-01:2026-12-01:DOD-7311-01", CASHIER))
assert (r["amount"], r["method"], r["received_on"], r["status"]) == ("100.00", "direct_debit", "2026-12-01", "received")
assert {(a["invoice_id"], a["decided_how"]) for a in table(t.pay_allocation)} == {("INV-7311", "direct_debit"), ("INV-7312", "direct_debit"), ("INV-7313", "direct_debit")}
assert all(v["outstanding"] == "0.00" for v in settle("INV-7311", "INV-7312", "INV-7313").values())
assert {j["event"] for j in nb.journals.values()} == {"receipt"} # three receipts posted to the ledger, GL-30 shape
again = collect()
assert again["unchanged"] is True and len(table(t.pay_receipt)) == 3 # collecting twice adds nothing
def test_PAY_19_the_collect_step_is_the_systems_alone(nb):
setup(nb)
refused(api("POST", "/dd-runs/dd:BARC-01:2026-12-01/collect", CASHIER), 403, "pay:dd:collect")
refused(api("POST", "/dd-runs/dd:BARC-01:2026-12-01/collect", BOTH), 403, "pay:dd:collect")
def test_PAY_19_a_mandate_cancelled_after_the_file_went_is_not_collected(nb):
setup(nb, (7311, 7312))
active_mandates((7311, 7312))
with on_day(PREP):
run()
to_submitted()
must(api("POST", "/dd-mandates/DOD-7312-01/cancel", CASHIER, {"reason": "customer asked"}))
got = collect()
assert got["collected"] == 1 and got["skipped"] == [{"mandate_ref": "DOD-7312-01", "reason": "the mandate is cancelled"}]
assert [r["payment_id"] for r in table(t.pay_receipt)] == ["dd:dd:BARC-01:2026-12-01:DOD-7311-01"]
def test_PAY_19_a_cancel_and_a_collection_racing_for_one_mandate_collide_on_its_claim(nb):
setup(nb, (7311,))
active_mandates((7311,))
with on_day(PREP):
run()
to_submitted()
snap = Snapshot() # the cancelling cashier loaded the mandate before the collection committed
collect()
with snap:
status, body = api("POST", "/dd-mandates/DOD-7311-01/cancel", CASHIER, {"reason": "late"})
assert status == 409 and "Someone else wrote" in str(body)
assert must(api("GET", "/dd-mandates", CASHIER))["mandates"][0]["lifecycle"] == "active"
# ── PAY-20: the bank's reports ───────────────────────────────────────────────────────────────────────
ARUDD = "mandate_ref,reason_code,returned_on\nDOD-7311-01,0,2026-12-03\nDOD-7312-01,1,2026-12-03\n"
def collected_run():
setup(nb_holder[0])
active_mandates()
with on_day(PREP):
run()
to_submitted()
collect()
nb_holder = [None]
def report(kind, ref, content, who=None):
with on_day("2026-12-04T09:00:00+00:00"):
return must(api("POST", "/dd-reports/import", who or CASHIER, {"kind": kind, "report_ref": ref, "content": content}))
def test_PAY_20_a_returned_collection_reverses_its_receipt_with_the_reason_and_the_invoice_is_owed_again(nb):
nb_holder[0] = nb
collected_run()
got = report("arudd", "ARUDD-1", ARUDD)
a, b = got["results"]
assert (a["result"], a["reason"], a["mandate"]) == ("returned", "0 refer to payer", "active") and "code 18" in a["represent"]
assert (b["result"], b["mandate"]) == ("returned", "cancelled")
r = must(api("GET", "/receipts/dd:dd:BARC-01:2026-12-01:DOD-7311-01", CASHIER))
assert (r["status"], r["returned_on"], r["return_reason_code"], r["bank_return_ref"]) == ("reversed", "2026-12-03", "0", "arudd:ARUDD-1:1")
assert settle("INV-7311", "INV-7312", "INV-7313")["INV-7311"]["outstanding"] == "100.00" # owed again
assert settle("INV-7313")["INV-7313"]["outstanding"] == "0.00" # the third was not returned
assert {m["mandate_ref"]: m["lifecycle"] for m in must(api("GET", "/dd-mandates", CASHIER))["mandates"]}["DOD-7312-01"] == "cancelled"
assert {j["event"] for j in nb.journals.values()} == {"receipt", "receipt_reversal"}
def test_PAY_20_every_report_line_is_applied_once(nb):
nb_holder[0] = nb
collected_run()
report("arudd", "ARUDD-1", ARUDD)
again = report("arudd", "ARUDD-1", ARUDD)
assert [x["result"] for x in again["results"]] == ["already applied", "already applied"]
n = len(table(t.pay_allocation))
other = report("arudd", "ARUDD-2", "mandate_ref,reason_code,returned_on\nDOD-7311-01,0,2026-12-03\n") # a different report naming a returned collection
assert other["results"][0]["result"].startswith("not applied: that collection was already returned") and len(table(t.pay_allocation)) == n
def test_PAY_20_a_return_is_never_dated_before_the_collection_or_after_today(nb):
nb_holder[0] = nb
collected_run()
refused(api("POST", "/dd-reports/import", CASHIER, {"kind": "arudd", "report_ref": "X", "content": "mandate_ref,reason_code,returned_on\nDOD-7311-01,0,2026-12-20\n"}),
409, "after today or before the receipt")
refused(api("POST", "/dd-reports/import", CASHIER, {"kind": "arudd", "report_ref": "Y", "content": "mandate_ref,reason_code,returned_on\nDOD-7311-01,0,not-a-date\n"}),
409, "date that cannot be read")
def test_PAY_20_reason_0_may_be_represented_once_with_code_18(nb):
nb_holder[0] = nb
collected_run()
report("arudd", "ARUDD-1", "mandate_ref,reason_code,returned_on\nDOD-7311-01,0,2026-12-03\n")
with on_day("2026-12-04T09:00:00+00:00"):
got = must(api("POST", "/dd-runs", CASHIER, {"bank_account_id": "BARC-01", "collection_date": "2026-12-15"}))
assert [(c["mandate_ref"], c["transaction_code"], c["invoice_ids"]) for c in got["collections"]] == [("DOD-7311-01", "18", ["INV-7311"])]
assert [c["represented"] for c in table(t.pay_dd_collection) if c["run_id"] == "dd:BARC-01:2026-12-01" and c["mandate_ref"] == "DOD-7311-01"] == ["Y"]
def test_PAY_20_addacs_c_changes_the_details_from_its_date_and_other_codes_cancel(nb):
nb_holder[0] = nb
collected_run()
got = report("addacs", "ADD-1", "mandate_ref,reason_code,effective_on,new_sort_code,new_account_number\nDOD-7311-01,C,2026-12-10,30-00-00,87654321\n"
"DOD-7313-01,B,2026-12-04,,\n")
assert [x["result"] for x in got["results"]] == ["bank details changed", "mandate cancelled"]
rows = [(r["valid_from"].isoformat(), r["sort_code"], r["account_number"]) for r in table(t.pay_dd_mandate_details) if r["mandate_ref"] == "DOD-7311-01"]
assert rows == [("2026-11-20", "200000", "12345678"), ("2026-12-10", "300000", "87654321")] # the old row is kept: history
ms = {m["mandate_ref"]: m for m in must(api("GET", "/dd-mandates", CASHIER))["mandates"]}
assert ms["DOD-7311-01"]["payer_account_masked"] == "****5678" and ms["DOD-7313-01"]["lifecycle"] == "cancelled"
final = report("addacs", "ADD-2", "mandate_ref,reason_code,effective_on,new_sort_code,new_account_number\nDOD-7313-01,C,2026-12-12,30-00-00,87654321\n")
assert "final" in final["results"][0]["result"] # a cancelled mandate stays cancelled
def test_PAY_20_a_ddica_indemnity_claim_reverses_the_collection_and_marks_the_mandate_disputed(nb):
nb_holder[0] = nb
collected_run()
got = report("ddica", "DDICA-1", "mandate_ref,claimed_on\nDOD-7313-01,2026-12-03\n")
assert got["results"][0]["mandate"] == "disputed"
m = {m["mandate_ref"]: m for m in must(api("GET", "/dd-mandates", CASHIER))["mandates"]}["DOD-7313-01"]
assert (m["lifecycle"], m["disputed_on"]) == ("cancelled", "2026-12-03")
assert settle("INV-7313")["INV-7313"]["outstanding"] == "300.00"
def test_PAY_20_a_return_and_a_new_allocation_racing_for_one_receipt_collide_on_its_source_claim(nb):
nb_holder[0] = nb
collected_run()
nb.invoice("INV-X", 7311, "50.00")
from conftest import receipt as rc
pid = rc(amount="50.00", reference="free")["payment_id"]
with use_cases.repository() as repo:
live = repo.receipts.get(pid)
must(api("POST", f"/receipts/{pid}/reverse", CASHIER, {"bank_return_ref": "BR1"})) # reverse commits first ...
with mock.patch.object(t.AllocationRepository, "for_credit", lambda self, k, i: []), mock.patch.object(t.AllocationRepository, "for_source", lambda self, k, i: []), \
mock.patch.object(t.ReceiptRepository, "get", lambda self, i: live):
# ... the allocating cashier read the receipt while it was still live, and the claims as they were
status, body = api("POST", f"/receipts/{pid}/allocate", CASHIER, {"lines": [{"invoice_id": "INV-X", "amount": "50.00"}]})
assert status == 409 and "Someone else wrote" in str(body)
assert [r["n"] for r in table(t.pay_source_claim) if r["source_id"] == pid] == [1]
def test_PAY_20_reports_need_the_permission_and_a_known_mandate(nb):
nb_holder[0] = nb
collected_run()
from conftest import SALES
refused(api("POST", "/dd-reports/import", SALES, {"kind": "arudd", "report_ref": "Z", "content": ARUDD}), 403, "pay:dd_run:write")
got = report("arudd", "ARUDD-9", "mandate_ref,reason_code,returned_on\nNOPE-0001,0,2026-12-03\n")
assert got["results"][0]["result"].startswith("not applied: no mandate NOPE-0001")