dodil/erp-brick-payments
DODIL ERP brick: payments
scripts/live_check.py
"""ONE live check: the brick alone on a scratch DataK3 bucket (org dodil) over the REAL pg wire, with real transactions
(erp_platform.livecheck.engine). The schema is applied from ALL the migrations IN ORDER (00001_init .. 00005_slice_c), then the stories that
depend on DataK3's behaviour:
- a receipt, an allocation, two allocations racing for one receipt (the claim row: on DataK3 EITHER racer may lose at commit), a reversal
- a camt.053 statement imported twice (the second adds nothing), a line matched, a posting unmatched twice (the second replays nothing;
the reversal journals carry distinct keys)
- a feed delivered twice the same day as separate statements (no false continuity gap)
- PAY-27: a machine identity reads a bank account unmasked and an audit row is written; a person role is refused 403
- the MCP tool count
Invoicing, the partner register, the ledger and tax are the stand-ins of tests/fakes.py, registered in process, and every seam call is checked
against the callee's signed spec (ERP_METHOD_DIR, the unsigned seam-gap branch). The token is the logged-in CLI session's, taken by
`dodil data connect` inside erp_platform.livecheck and never printed. The exit code is erp_platform.livecheck.finish's: a crashed run fails.
ERP_METHOD_DIR=../erp-method PATH=$PATH:~/.local/bin BUCKET=w2paylab DODIL_ORG=dodil python scripts/live_check.py
"""
import base64
import json
import os
import pathlib
import re
import sys
import threading
ROOT = pathlib.Path(__file__).resolve().parents[1]
sys.path[:0] = [str(ROOT / "src"), str(ROOT / "tests")]
os.environ.setdefault("BUCKET", "w2paylab")
os.environ.setdefault("DODIL_ORG", "dodil")
METHOD_DIR = os.environ.get("ERP_METHOD_DIR") or str(ROOT.parents[0] / "erp-method")
import anyio # noqa: E402
import httpx # noqa: E402
import httpx2 # noqa: E402
import yaml # noqa: E402
from erp_platform import contract as _contract # noqa: E402
from erp_platform import livecheck as lc # noqa: E402
from erp_platform import seam # noqa: E402
from mcp import ClientSession # noqa: E402
from mcp.client.streamable_http import streamable_http_client # noqa: E402
from brick_payments.app import app # noqa: E402
from brick_payments.application import use_cases as uc # noqa: E402
from brick_payments.infrastructure.tables import md # noqa: E402
def person(email, perms):
payload = base64.urlsafe_b64encode(json.dumps({"permissions": list(perms)}).encode()).decode().rstrip("=")
return {"X-Dodil-User": json.dumps({"sub": email, "email": email, "app_roles": []}), "X-Dodil-User-JWT": f"h.{payload}.s"}
ROLES = yaml.safe_load((ROOT / "src" / "brick_payments" / "roles.yaml").read_text())["roles"]
CASHIER, APPROVER, SYSTEM = (person(f"{r}@live", ROLES[r]) for r in ("cashier", "approver", "system"))
# Routes payments calls that the callee's signed spec does not list for payments (reported to the method as spec gaps).
_GAPS = (("ar", "GET /invoice-totals"), ("ar", "GET /credit-notes"), ("ar", "POST /writeoffs"), ("ar", "GET /migration/open-balances"))
VIOLATIONS: list = []
def install_contract():
plain = _contract.check
def check(callee, method, path, body=None, **kw):
bad = plain(callee, method, path, body, **kw)
if any(callee == c and f"{method} {path}".startswith(p) for c, p in _GAPS):
return [b for b in bad if "is listed for" not in b and "no seam" not in b]
return bad
_contract.check = check
original = seam.Seam._call
def call(self, method, path, actor, caller=None, **kw):
mod = {"bp": "business-partner"}.get(self.name, self.name)
if mod != "business-partner" and self.installed and (pathlib.Path(METHOD_DIR) / "modules" / mod / "module.yaml").exists():
bad = _contract.check(mod, method, path, kw.get("json") if method in ("POST", "PUT", "PATCH") else kw.get("params"),
caller="payments", method_dir=METHOD_DIR)
if bad:
VIOLATIONS.append(f"payments -> {mod} {method} {path}: {bad}")
return original(self, method, path, actor, caller, **kw)
seam.Seam._call = call
def apply_migrations() -> int:
n = 0
for d in sorted((ROOT / "src" / "brick_payments" / "migrations").iterdir()):
for stmt in re.split(r";\s*\n", (d / "up.sql").read_text()):
body = "\n".join(x for x in stmt.splitlines() if not x.strip().startswith("--")).strip().rstrip(";")
if body:
lc.pg(body)
n += 1
return n
async def route(method, path, headers, body=None):
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://brick", timeout=120) as c:
r = await c.request(method, path, headers=headers, json=body) if method != "GET" else await c.get(path, headers=headers, params=body)
try:
return r.status_code, r.json()
except ValueError:
return r.status_code, {"detail": r.text}
def api(*a):
return anyio.run(route, *a)
async def mcp_tools(headers):
async with app.router.lifespan_context(app):
async with httpx2.AsyncClient(transport=httpx2.ASGITransport(app=app), headers=headers, base_url="http://brick") as http:
async with streamable_http_client("http://brick/mcp", http_client=http) as (r, w):
async with ClientSession(r, w) as s:
await s.initialize()
return sorted(t.name for t in (await s.list_tools()).tools)
def n_rows(table, where="TRUE"):
return int(lc.rows(f"SELECT COUNT(*) AS n FROM {table} WHERE {where}")[0]["n"])
def camt(entries, stmt_id, account="GB00BANK"):
"""camt.053 entries: (ref, 'CRDT'|'DBIT', amount, booked_on, text, counterparty)."""
body = "".join(
f"<Ntry><NtryRef>{r}</NtryRef><Amt Ccy='GBP'>{a}</Amt><CdtDbtInd>{d}</CdtDbtInd><BookgDt><Dt>{day}</Dt></BookgDt>"
f"<NtryDtls><TxDtls><RltdPties><Dbtr><Nm>{nm}</Nm></Dbtr><Cdtr><Nm>{nm}</Nm></Cdtr></RltdPties>"
f"<RmtInf><Ustrd>{u}</Ustrd></RmtInf></TxDtls></NtryDtls></Ntry>" for r, d, a, day, u, nm in entries)
return ("<?xml version='1.0'?><Document xmlns='urn:iso:std:iso:20022:tech:xsd:camt.053.001.02'><BkToCstmrStmt><Stmt>"
f"<Id>{stmt_id}</Id><Acct><Id><IBAN>{account}</IBAN></Id></Acct>{body}</Stmt></BkToCstmrStmt></Document>")
def main():
from fakes import Neighbours
install_contract()
eng = lc.engine()
nb = Neighbours()
lc.drop_all(list(md.tables))
print(f"migrations applied: {apply_migrations()} statements on {os.environ['BUCKET']}")
uc.repository = lambda: uc._on_engine(eng) # the brick's own commit-time conflict handling
with uc.repository() as r:
r.seed()
today = uc._today(r).isoformat()
for name in ("ar", "bp", "gl", "tax", "purchasing"):
seam.register_local(name, getattr(nb, name))
def must(name, res):
lc.check(name, res[0] == 200, res)
return res[1]
# ── an account, a receipt, an allocation, a collision, a reversal ─────────────────────────────
must("bank account BANK1", api("POST", "/bank-accounts/BANK1", APPROVER, {"name": "Current account", "currency": "GBP", "account_number": "12345678",
"company_code": "UK01", "gl_account_id": "1200", "purposes": "receive,pay"}))
for iid in ("INV-L-1", "INV-L-2", "INV-L-3"):
nb.invoice(iid, 7311, "1000.00", number=iid, issued_on=today, due_date=today)
def new_receipt(ref):
return must(f"receipt {ref} recorded", api("POST", "/receipts", CASHIER, {"bp_id": 7311, "amount": "1000.00", "currency": "GBP",
"received_on": today, "bank_reference": ref}))["payment_id"]
def alloc(p, inv):
return ("POST", f"/receipts/{p}/allocate", CASHIER, {"lines": [{"invoice_id": inv, "amount": "1000.00"}]})
pid, pid2 = new_receipt("LIVE-R1"), new_receipt("LIVE-R2")
must("receipt allocated to an invoice", api(*alloc(pid, "INV-L-1")))
results = {}
def go(inv):
try:
results[inv] = api(*alloc(pid2, inv))[0]
except Exception as exc: # noqa: BLE001 - a crash is a result too (it was a 500 before the commit-time 409)
results[inv] = type(exc).__name__
threads = [threading.Thread(target=go, args=(i,)) for i in ("INV-L-2", "INV-L-3")]
[t.start() for t in threads]
[t.join() for t in threads]
lc.check("allocation collision: two writers, one receipt - exactly one won, the other was refused 409 (either may lose at commit)",
sorted(results.values()) == [200, 409], results)
must("receipt reversed (bank return)", api("POST", f"/receipts/{pid}/reverse", CASHIER, {"bank_return_ref": "RET-L1"}))
# ── a statement: twice, a match, unmatch twice ────────────────────────────────────────────────
content = camt([("H1", "DBIT", "31350.00", today, "HMRC VAT", "HMRC"), ("D1", "DBIT", "12.50", today, "SERVICE CHARGE", "Bank")], "LIVE-1")
imp = {"format": "camt053", "content": content, "bank_account_id": "BANK1"}
first = must("statement imported", api("POST", "/statements/import", CASHIER, imp))
again = must("statement imported again", api("POST", "/statements/import", CASHIER, imp))
lc.check("importing the same statement twice adds nothing", first["lines_created"] == 2 and again["lines_created"] == 0
and again["skipped_duplicate"] == 2 and n_rows("pay_statement_line", "statement_id LIKE '%LIVE-1'") == 2, (first, again))
must("vat control account", api("POST", "/controls/vat_control_account_id", APPROVER, {"value": "2200"}))
nb.liabilities = [{"period_key": "26AD", "charge_type": "vat", "outstanding_amount": "31350.00", "due_date": today}]
m = must("a line matched to tax's liability", api("POST", "/statement-lines/bank:BANK1:H1/match", CASHIER, {"kind": "tax_liability", "ref": "26AD:vat"}))
lc.check("the match posted a journal", m["lifecycle"] == "matched" and str(m["journal_id"]).isdigit(), m)
post = {"account_id": "7900", "tax_code": "E"}
must("line posted", api("POST", "/statement-lines/bank:BANK1:D1/post", CASHIER, post))
must("line unmatched with a reason", api("POST", "/statement-lines/bank:BANK1:D1/unmatch", CASHIER, {"reason": "wrong account"}))
s2, b2 = api("POST", "/statement-lines/bank:BANK1:D1/unmatch", CASHIER, {"reason": "again"})
lc.check("unmatch twice: the second is refused, nothing to undo", s2 == 409 and "nothing to undo" in str(b2), (s2, b2))
must("line posted again", api("POST", "/statement-lines/bank:BANK1:D1/post", CASHIER, post))
must("line unmatched again", api("POST", "/statement-lines/bank:BANK1:D1/unmatch", CASHIER, {"reason": "and again"}))
revs = [k for k in nb.journals if k.startswith("bank_line_reversal:bank:BANK1:D1@")]
lc.check("two unmatches, two reversal journals under distinct keys", len(revs) == 2 and len(set(revs)) == 2, revs)
# ── the feed, delivered twice the same day ────────────────────────────────────────────────────
must("feed consent", api("POST", "/bank-accounts/BANK1/feed-consent", APPROVER, {"provider": "truelayer"}))
def tx(i, amount):
return {"transaction_id": i, "booked_on": today, "amount": amount, "direction": "credit", "counterparty_name": "Acme Ltd", "remittance": f"ref {i}"}
def feed(lines, balance):
return api("POST", "/statements/ingest", SYSTEM, {"bank_account_id": "BANK1", "as_of": today, "booked_balance": balance, "lines": lines})
must("feed delivery 1", feed([tx("T1", "100.00")], "1100.00"))
d2 = must("feed delivery 2 (T1 repeated, T3 new)", feed([tx("T1", "100.00"), tx("T3", "5.00")], "1105.00"))
lc.check("second same-day delivery: no false continuity gap, statement balanced", "continuity" not in d2 and d2["statement_lifecycle"] == "balanced", d2)
lc.check("one line per bank transaction id", n_rows("pay_statement_line", "line_id LIKE '%T1' OR line_id LIKE '%T3'") == 2)
# ── PAY-27: the machine read is audited, a person's role is refused ───────────────────────────
must("house bank", api("POST", "/house-banks/BARC", APPROVER, {"name": "Barclays", "sort_code": "20-00-00", "sun": "123456", "bic": "BARCGB22"}))
must("account with details", api("POST", "/bank-accounts/BARC-01", APPROVER, {
"name": "Dodil Trading Ltd", "currency": "GBP", "account_number": "31926819", "company_code": "UK01", "gl_account_id": "1200",
"purposes": "receive,pay", "house_bank_id": "BARC"}))
url = "/bank-accounts/BARC-01/account-details"
denied = api("GET", url, CASHIER)[0]
lc.check("a person role reads no account details (403), and the refusal leaves no audit row", denied == 403 and n_rows("pay_bank_account_read") == 0, denied)
got = must("machine identity reads the account unmasked", api("GET", url, SYSTEM))
lc.check("unmasked number returned, and exactly one audit row names the caller",
got.get("account_number") == "31926819" and n_rows("pay_bank_account_read", "bank_account_id = 'BARC-01' AND caller = 'system@live'") == 1, got)
# ── the agent door ───────────────────────────────────────────────────────────────────────────
everyone = person("all@live", sorted({u.permission for u in uc.PAY.use_cases if u.permission}))
names = anyio.run(mcp_tools, everyone)
expected = sorted(u.name for u in uc.PAY.use_cases if u.tool)
lc.check(f"MCP: {len(names)} tools, one per tool-enabled use case; the machine read is not one", names == expected and "get_account_details" not in names,
(len(names), len(expected)))
print(f"MCP tools: {len(names)}")
lc.check("every seam call kept to the callee's signed spec", not VIOLATIONS, VIOLATIONS[:3])
if __name__ == "__main__":
try:
main()
finally: # replaces an exception in flight: finish() records a crash as a failure
raise SystemExit(lc.finish(list(md.tables)))