dodil/erp-brick-purchasing
DODIL ERP brick: purchasing
tests/test_files.py
"""The bank files: BACS Standard 18 (PUR-59) and pain.001 (PUR-29)."""
import xml.etree.ElementTree as ET
from conftest import (
CLERK2,
OWNER,
PAYAPPROVER,
SYSTEM,
api,
bill,
confirmed,
must,
receive,
refused,
supplier_bank,
supplier_price,
world,
)
from brick_purchasing.domain import files
UK = 7020
def bacs_world(service_user="123456"):
world()
if service_user:
must(api("POST", "/settings/controls", OWNER, {"setting_key": "bacs_service_user_number", "value": service_user, "valid_from": "2026-11-20"}))
supplier_bank(UK, iban="12345678", sort="60-16-13", name="Acme Haulage Ltd")
def due_bill(supplier=UK, number="INV-1", price="18.0000", note="DN-1", due="2026-11-24", currency="GBP"):
supplier_price(bp=supplier, price=price)
o = confirmed(bp=supplier, lines=[{"product_id": "COLA-075", "quantity": "10"}])
receive(o, note=note)
s, b = bill(o, bp=supplier, number=number, due_date=due, unit_price=price)
assert s == 200, b
return must(api("POST", f"/bills/{b['bill_id']}/approve", PAYAPPROVER, {}))
def bacs_run(pay_date="2026-11-25", **kw):
return api("POST", "/payment-runs", CLERK2, {"bank_account_id": "BARC-01", "pay_date": pay_date, "file_format": "bacs_std18", **kw})
def test_pur59_a_bacs_standard_18_file_has_100_character_details_a_contra_and_totals_in_pence():
bacs_world()
due_bill(number="A")
due_bill(number="B", note="DN-2", price="25.5000")
run = must(bacs_run())
must(api("POST", f"/payment-runs/{run['run_id']}/approve", PAYAPPROVER, {}))
f = must(api("POST", f"/payment-runs/{run['run_id']}/export", SYSTEM, {}))
assert f["file_key"] == f"raw/purchasing/runs/{run['run_id']}.std18"
lines = f["content"].splitlines()
assert [ln[:4] for ln in lines[:4]] == ["VOL1", "HDR1", "HDR2", "UHL1"] and [ln[:4] for ln in lines[-3:]] == ["EOF1", "EOF2", "UTL1"]
assert all(len(ln) == 80 for ln in lines[:4] + lines[-3:])
details = lines[4:-3]
assert len(details) == 2 and all(len(d) == 100 for d in details) # ONE payment per supplier (both bills are the same supplier): details + the contra
detail, contra = details[0], details[-1]
assert detail[:6] == "601613" and detail[6:14] == "12345678" and detail[15:17] == "99" and detail[17:23] == "601613" and detail[23:31] == "87654321"
pence = int(detail[35:46])
assert pence == (216 + 306) * 100 and int(contra[35:46]) == pence
assert detail[82:100].strip() == "ACME HAULAGE LTD"
detail_sum, trailer_sum, count = files.bacs_totals(f["content"])
assert detail_sum == trailer_sum == pence and count == 1
again = must(api("POST", f"/payment-runs/{run['run_id']}/export", SYSTEM, {}))
assert again["replay"] is True and again["content"] == f["content"]
def test_pur59_two_suppliers_make_two_detail_records_and_the_contra_equals_their_sum():
bacs_world()
supplier_bank(7021, iban="87654321", sort="20-00-00", name="Beta Ltd")
due_bill(UK, "A")
due_bill(7021, "B", note="DN-2", price="25.5000")
run = must(bacs_run())
assert len(run["payments"]) == 2
must(api("POST", f"/payment-runs/{run['run_id']}/approve", PAYAPPROVER, {}))
f = must(api("POST", f"/payment-runs/{run['run_id']}/export", SYSTEM, {}))
details = f["content"].splitlines()[4:-3]
assert len(details) == 3 and all(len(d) == 100 for d in details)
assert sum(int(d[35:46]) for d in details[:2]) == int(details[2][35:46])
assert files.bacs_totals(f["content"])[2] == 2
def test_pur59_bacs_ready_refuses_with_the_reason_and_the_earliest_pay_date():
bacs_world(service_user="")
due_bill(due="2026-11-20")
run = must(bacs_run(pay_date="2026-11-23"))
gate = [g for g in run["readiness"] if g["gate"] == "bacs_ready"][0]
assert gate["met"] is False and "service user number" in gate["reason"] and "earliest 2026-11-25" in gate["reason"]
why = refused(api("POST", f"/payment-runs/{run['run_id']}/approve", PAYAPPROVER, {}))
assert "bacs_ready" in why and "2026-11-25" in why
def test_pur59_a_payee_without_a_uk_sort_code_and_account_is_not_bacs_ready():
bacs_world()
supplier_bank(7022, iban="DE89370400440532013000", sort="COBADEFF", name="Gmbh")
due_bill(7022, "G", note="DN-5")
run = must(bacs_run())
gate = [g for g in run["readiness"] if g["gate"] == "bacs_ready"][0]
assert gate["met"] is False and "no UK sort code" in gate["reason"]
def test_pur59_a_run_that_is_not_in_gbp_is_not_bacs_ready():
ok_, _ = files.bacs_ready([], "EUR", "123456", __import__("datetime").date(2026, 12, 1), __import__("datetime").date(2026, 11, 20), 3, "601613 87654321")[0], None
assert ok_.met is False and "not GBP" in ok_.reason
def test_pain001_09_has_the_same_payments_in_the_newer_schema():
world()
from test_stories_pay import approved_bill, propose
approved_bill()
run = must(propose(file_format="pain.001.001.09"))
must(api("POST", f"/payment-runs/{run['run_id']}/approve", PAYAPPROVER, {}))
f = must(api("POST", f"/payment-runs/{run['run_id']}/export", SYSTEM, {}))
root = ET.fromstring(f["content"])
ns = {"p": root.tag.split("}")[0][1:]}
assert "pain.001.001.09" in ns["p"] and root.find(".//p:ReqdExctnDt/p:Dt", ns).text == "2026-12-31"
assert root.find(".//p:DbtrAgt/p:FinInstnId/p:BICFI", ns).text == "BARCGB22"
def test_csv_file_for_a_bank_that_imports_csv():
world()
from test_stories_pay import approved_bill, propose
approved_bill()
run = must(propose(file_format="csv"))
must(api("POST", f"/payment-runs/{run['run_id']}/approve", PAYAPPROVER, {}))
f = must(api("POST", f"/payment-runs/{run['run_id']}/export", SYSTEM, {}))
rows = f["content"].splitlines()
assert rows[0].startswith("payment_id,payee") and "216.00,GBP" in rows[1] and f["file_key"].endswith(".csv")
# ── PAY-27: our own account comes from payments per file and is never kept ──────────────────────────
ACCOUNT_NUMBERS = ("87654321", "601613 87654321", "GB12BARC20000012345678")
_n = [0]
def approved_run(fmt):
"""An approved run in `fmt` over a fresh bill (BACS: the UK supplier with a sort code and account; pain.001: the usual supplier)."""
_n[0] += 1
if fmt == "bacs_std18":
due_bill(number=f"X{_n[0]}", note=f"DN-X{_n[0]}")
run = must(bacs_run())
else:
from test_stories_pay import approved_bill, propose
approved_bill()
run = must(propose(file_format=fmt))
must(api("POST", f"/payment-runs/{run['run_id']}/approve", PAYAPPROVER, {}))
return run
def test_pur19_the_file_uses_the_details_payments_served_and_asks_with_the_run_id_as_purpose(neighbours):
bacs_world()
run = approved_run("bacs_std18")
f = must(api("POST", f"/payment-runs/{run['run_id']}/export", SYSTEM, {}))
assert "601613" in f["content"] and "87654321" in f["content"]
assert neighbours.payments.reads == [("BARC-01", f"payment_run:{run['run_id']}")]
f2 = must(api("POST", f"/payment-runs/{run['run_id']}/export", SYSTEM, {}))
assert f2["replay"] and len(neighbours.payments.reads) == 1 # a replay serves the kept file: no second read
pain = approved_run("pain.001.001.03")
p = must(api("POST", f"/payment-runs/{pain['run_id']}/export", SYSTEM, {}))
assert "GB12BARC20000012345678" in p["content"] and "BARCGB22" in p["content"]
def test_pur19_nothing_but_the_file_holds_our_account_number(neighbours, engine):
from conftest import tool
from sqlalchemy import text
bacs_world()
run = approved_run("bacs_std18")
s, body = api("POST", f"/payment-runs/{run['run_id']}/export", SYSTEM, {})
assert s == 200
resp = {k: v for k, v in body.items() if k != "content"} # the file body is the one place it lives (and the object)
assert not any(n in json.dumps(resp) for n in ACCOUNT_NUMBERS)
for view in (must(api("GET", f"/payment-runs/{run['run_id']}", CLERK2)), must(api("GET", "/payment-runs", CLERK2))):
assert not any(n in json.dumps(view) for n in ACCOUNT_NUMBERS)
with engine.connect() as c:
rows = [str(r) for r in c.execute(text("SELECT * FROM pur_payment_run")).fetchall()]
assert rows and not any(n.replace(" ", "") in r.replace(" ", "") for r in rows for n in ACCOUNT_NUMBERS[:2])
# the MCP tool result for a replay carries the kept file only; the export of a fresh run returns no account outside `content`
again = must(api("POST", f"/payment-runs/{run['run_id']}/export", SYSTEM, {}))
assert again["replay"]
r2 = approved_run("bacs_std18")
st, out = tool(SYSTEM, "export_bank_file", {"run_id": r2["run_id"]})
assert st == "ok" and not any(n in json.dumps({k: v for k, v in out.items() if k != "content"}) for n in ACCOUNT_NUMBERS)
import json # noqa: E402
def test_pur19_payments_403_404_503_refuse_the_file_with_the_originator_account_gate(neighbours):
bacs_world()
for status, words in ((403, "pay:bank_account:read_unmasked"), (404, "no paying account"), (503, "could not log")):
run = approved_run("bacs_std18")
neighbours.payments.fail = status
msg = refused(api("POST", f"/payment-runs/{run['run_id']}/export", SYSTEM, {}), 409)
assert msg.startswith("originator_account:") and "fix:" in msg and words in msg, msg
neighbours.payments.fail = None
assert must(api("GET", f"/payment-runs/{run['run_id']}", CLERK2))["status"] == "approved" # nothing was written