dodil/erp-brick-ar
DODIL ERP brick: accounts receivable
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 the migrations IN ORDER (00001_init .. 00004_v4b, so the ADD COLUMN and the
new tables are exercised), then the stories that depend on DataK3's behaviour:
- a deposit through a payment request, a settle replay (one deposit invoice), and two finals racing for one deposit's remainder
(the claim row: exactly one wins; on DataK3 EITHER racer may lose at commit)
- a Peppol invoice: sent, delivered, rejected, a duplicate status adding nothing, timestamps read back in order
- deferred revenue: a catch-up release, then a second run adding nothing
- a doubtful-debt provision proposed, approved by another person, approved again
- a manual recurring invoice raised by the daily pass
- the MCP tool count
Tax, payments and the access point are the stand-ins of tests/fakes.py, registered in process. 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.
PATH=$PATH:~/.local/bin BUCKET=w2arlab DODIL_ORG=dodil python scripts/live_check.py
"""
import base64
import json
import os
import pathlib
import re
import sys
import threading
from contextlib import contextmanager
from datetime import timedelta
from decimal import Decimal
ROOT = pathlib.Path(__file__).resolve().parents[1]
sys.path[:0] = [str(ROOT / "src"), str(ROOT / "tests")]
os.environ.setdefault("BUCKET", "w2arlab")
os.environ.setdefault("DODIL_ORG", "dodil")
import anyio # noqa: E402
import httpx # noqa: E402
import httpx2 # noqa: E402
import yaml # 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_ar.app import app # noqa: E402
from brick_ar.application import use_cases as uc # noqa: E402
from brick_ar.infrastructure.tables import Repository, md # noqa: E402
def person(email, perms):
payload = base64.urlsafe_b64encode(json.dumps({"permissions": 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_ar" / "roles.yaml").read_text())["roles"]
BILLER, APPROVER, OWNER, SYSTEM = (person(f"{r}@live", ROLES[r]) for r in ("biller", "approver", "owner", "system"))
PROPOSER = person("proposer@live", ROLES["biller"])
def apply_migrations() -> int:
n = 0
for d in sorted((ROOT / "src" / "brick_ar" / "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 main():
from fakes import Payments, Peppol, Tax
eng = lc.engine()
pay, peppol = Payments(), Peppol()
lc.drop_all(list(md.tables))
print(f"migrations applied: {apply_migrations()} statements on {os.environ['BUCKET']}")
@contextmanager
def repo():
with eng.begin() as conn:
yield Repository(conn)
uc.repository = repo
with repo() as r:
r.seed()
today = uc._today(r)
seam.register_local("tax", Tax().app)
seam.register_local("pay", pay.app)
seam.register_local("peppol", peppol.app)
first_this = today.replace(day=1)
prev_end = first_this - timedelta(days=1) # the last day of last month: a period that has ended
prev_period = f"{prev_end.year:04d}-{prev_end.month:02d}"
BP, BP2 = 880001, 880002
def billing(bp, **kw):
s, b = api("POST", f"/customers/{bp}/billing", APPROVER, {"bill_to_name": "Live Ltd", "bill_to_address": "1 Test St, Leeds, GB",
"invoice_email": "ap@live.example", "terms_id": "NET30", "currency": "GBP", **kw})
lc.check(f"billing profile {bp}", s == 200, b)
line = {"description": "Managed hosting", "quantity": "1", "unit_price": "100.00", "unit_code": "MON", "tax_category": "standard"}
billing(BP)
# ── a deposit through a payment request; the settle replays ───────────────────────────────────
s, b = api("POST", "/controls", OWNER, {"key": "deposit_mode", "value": "request"})
lc.check("deposit_mode request", s == 200, b)
s, rq = api("POST", "/payment-requests", BILLER, {"bp_id": str(BP), "currency": "GBP", "net_amount": "5000.00", "sales_order_id": "SO-L", "request_no": 1})
lc.check("payment request raised", s == 200 and rq["lifecycle"] == "draft", rq)
s, b = api("POST", f"/payment-requests/{rq['request_id']}/issue", BILLER, {})
lc.check("payment request issued", s == 200 and b["lifecycle"] == "issued", b)
body = {"payment_id": "LIVE-P1", "received_on": today.isoformat(), "amount": "6000.00", "currency": "GBP"}
s1, a = api("POST", f"/payment-requests/{rq['request_id']}/settle", SYSTEM, body)
s2, a2 = api("POST", f"/payment-requests/{rq['request_id']}/settle", SYSTEM, body)
dep = a.get("deposit_invoice_id")
lc.check("settle: a deposit invoice for the money received", s1 == 200 and dep and a["status"] == "settled", a)
lc.check("settle replay: the same deposit invoice, no second row", s2 == 200 and a2.get("deposit_invoice_id") == dep
and n_rows("ar_request_settlement", f"request_id = '{rq['request_id']}'") == 1, a2)
# ── two finals racing for one deposit's remainder ─────────────────────────────────────────────
finals = []
for k in ("A", "B"):
s, f = api("POST", "/invoices/raise", SYSTEM, {"source_kind": "sales_order", "source_id": f"so:LIVE-{k}:1", "bp_id": str(BP), "currency": "GBP",
"kind": "final", "lines": [{**line, "unit_price": "10000.00"}],
"deposit_deductions": [{"invoice_id": dep, "amount": "5000.00"}]})
lc.check(f"final {k} drafted with a deduction line", s == 200 and any(x.get("deducts_invoice_id") == dep for x in f["lines"]), f)
finals.append(f["invoice_id"])
results = {}
def issue(iid):
results[iid] = api("POST", f"/invoices/{iid}/issue", BILLER, {})[0]
threads = [threading.Thread(target=issue, args=(i,)) for i in finals]
[t.start() for t in threads]
[t.join() for t in threads]
lc.check("deposit claim: exactly one of two racing finals issued, the other was refused 409", sorted(results.values()) == [200, 409], results)
lc.check("the deposit's deductions never exceed its net", Decimal(str(lc.rows(
f"SELECT COALESCE(SUM(net), 0) AS s FROM ar_deposit_claim WHERE deposit_invoice_id = '{dep}'")[0]["s"])) == Decimal("5000.00"))
# ── Peppol ─────────────────────────────────────────────────────────────────────────────────────
for k, v in (("supplier_name", "Dodil Ltd"), ("supplier_address", "1 Mill Rd, London, GB"), ("supplier_country", "GB"),
("einvoice_sender_id", "0088:5060000000001"), ("einvoice_provider", "peppol")):
s, b = api("POST", "/settings", APPROVER, {"key": k, "value": v})
lc.check(f"setting {k}", s == 200, b)
billing(BP2, einvoice_channel="peppol", peppol_participant_id="0088:5060012349998")
s, d = api("POST", "/invoices", BILLER, {"bp_id": str(BP2), "currency": "GBP", "lines": [line]})
s, i = api("POST", f"/invoices/{d['invoice_id']}/issue", BILLER, {})
lc.check("peppol invoice issued (validated before a number)", s == 200 and i["invoice_no"], i)
s, sent = api("POST", f"/invoices/{d['invoice_id']}/send", BILLER, {})
mid = (sent.get("send") or {}).get("message_id")
lc.check("peppol send: UBL handed over, message id recorded", s == 200 and mid and "Invoice" in peppol.sent[-1]["xml"], sent)
for st, reason, at in (("delivered", None, "2026-10-20T09:00:00Z"), ("RE", "PO number missing", "2026-10-21T10:30:00Z"),
("RE", "PO number missing", "2026-10-21T10:30:00Z")):
s, r = api("POST", f"/einvoices/{mid}/status", SYSTEM, {"status": st, "at": at, **({"reason": reason} if reason else {})})
lc.check("e-invoice statuses: delivered then rejected with its reason, the duplicate added nothing",
s == 200 and [x["status"] for x in r["statuses"]] == ["delivered", "RE"] and r["rejected"] == "PO number missing", r)
lc.check("one row per (message, status)", n_rows("ar_einvoice_status", f"message_id = '{mid}'") == 2)
s, g = api("GET", f"/invoices/{d['invoice_id']}", BILLER)
lc.check("the rejection shows on the invoice, which is unchanged", s == 200 and g["lifecycle"] == "issued"
and (g.get("einvoice") or {}).get("rejection") == "PO number missing", g.get("einvoice"))
# ── deferred revenue ─────────────────────────────────────────────────────────────────────────
start = (prev_end.replace(day=1) - timedelta(days=1)).replace(day=1) # the first of the month before last month
end = start.replace(year=start.year + 1) - timedelta(days=1)
s, d = api("POST", "/invoices", BILLER, {"bp_id": str(BP), "currency": "GBP", "lines": [
{**line, "description": "Annual support", "unit_price": "12000.00", "service_from": start.isoformat(), "service_to": end.isoformat()}]})
s, inv = api("POST", f"/invoices/{d['invoice_id']}/issue", BILLER, {})
lc.check("annual service invoice issued", s == 200, inv)
s, a = api("POST", "/revenue/release", SYSTEM, {"period_id": prev_period})
s2, b = api("POST", "/revenue/release", SYSTEM, {"period_id": prev_period})
lc.check("release: the catch-up months once", s == 200 and len(a["released"]) == 2, a)
lc.check("release twice: the second run adds nothing", s2 == 200 and b["released"] == [] and b["already_released"] == 2
and n_rows("ar_deferral_release", f"invoice_id = '{d['invoice_id']}' AND kind <> 'guard'") == 2, b) # guard rows serialise release/credit/void
# ── the provision ───────────────────────────────────────────────────────────────────────────
pay.totals[d["invoice_id"]] = Decimal(inv["total"])
pay.aged["31-60"].append({"invoice_id": d["invoice_id"], "bp_id": BP, "outstanding": inv["total"], "days_past_due": 40, "disputed": False})
s, b = api("POST", "/provision-rates", APPROVER, {"bucket": "31-60", "rate_pct": "2"})
lc.check("provision rate", s == 200, b)
s, run = api("POST", "/provisions", PROPOSER, {"period_id": prev_period})
lc.check("provision proposed: net of VAT at the bucket rate", s == 200 and run["total"] == "240.00", run)
s, ok = api("POST", f"/provisions/{run['run_id']}/approve", APPROVER, {})
s2, ok2 = api("POST", f"/provisions/{run['run_id']}/approve", APPROVER, {})
lc.check("provision approved by another person, approving again changes nothing", s == 200 and ok["lifecycle"] == "approved"
and s2 == 200 and ok2["version"] == ok["version"], (ok, ok2))
# ── a manual recurring invoice ──────────────────────────────────────────────────────────────
s, sch = api("POST", "/schedules/manual", BILLER, {"bp_id": str(BP), "currency": "GBP", "description": "Desk rental", "unit_price": "450.00",
"starts_on": today.isoformat(), "periods": 12})
lc.check("manual schedule keyed man:<uuid>|1", s == 200 and sch["schedule_id"].startswith("man:") and sch["schedule_id"].endswith("|1"), sch)
s, raised = api("POST", "/schedules/raise-due", SYSTEM, {})
lc.check("the daily pass raised its first period as a draft", s == 200 and f"rec:{sch['schedule_id']}:1" in raised["raised"], raised)
# ── the agent door ───────────────────────────────────────────────────────────────────────────
everyone = person("all@live", sorted({u.permission for u in uc.AR.use_cases if u.permission}))
names = anyio.run(mcp_tools, everyone)
expected = sorted(u.name for u in uc.AR.use_cases if u.tool)
new = {"create_recurring_invoice", "release_deferred_revenue", "record_einvoice_status", "propose_doubtful_provision",
"approve_doubtful_provision", "send_credit_einvoice", "hand_over_credit_note"}
lc.check(f"MCP: {len(names)} tools, one per tool-enabled use case, the slice-B ones among them", names == expected and new <= set(names),
(len(names), len(expected)))
print(f"MCP tools: {len(names)}")
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)))