dodil/erp-brick-gl
DODIL ERP brick: general ledger
scripts/live_check.py
"""ONE live check: the brick alone on a scratch DataK3 bucket (default w2gllab, org dodil) - apply the schema, then one happy-path story through a
ROUTE (books -> post -> replay -> trial balance) and one through MCP (the same books: propose a manual journal -> read it -> reverse it ->
trial balance). The pg wire is real. Neighbours are absent (every one is optional).
BUCKET=w2gllab python scripts/live_check.py
The schema is applied here (not with `erp-compose migrate`, which needs the module signed in the published method). The token is the
logged-in CLI session's, read from `dodil data connect` and never printed.
"""
import base64
import json
import os
import pathlib
import re
import subprocess
import sys
import time
ROOT = pathlib.Path(__file__).resolve().parents[1]
sys.path[:0] = [str(ROOT / "src")]
BUCKET = os.environ.get("BUCKET", "w2gllab")
out = subprocess.run(["dodil", "data", "connect", BUCKET], capture_output=True, text=True, check=True).stdout
os.environ["DODIL_TOKEN"] = re.search(r"token:([^@]+)@", out).group(1)
os.environ["BUCKET"] = BUCKET
import anyio # noqa: E402
import httpx # noqa: E402
import httpx2 # noqa: E402
from erp_platform import db # noqa: E402
from erp_platform.seam import derived_id # noqa: E402
from mcp import ClientSession # noqa: E402
from mcp.client.streamable_http import streamable_http_client # noqa: E402
from brick_gl.app import app # 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"}
ACCOUNTANT = person("accountant@live", ["gl:account:write", "gl:company:write", "gl:company:read", "gl:settings:write", "gl:period:close", "gl:period:read",
"gl:journal:read", "gl:journal:write", "gl:reversal:approve", "gl:report:read"])
OWNER = person("owner@live", ["gl:settings:controls"])
POSTER = person("ar@live", ["gl:journal:post", "gl:period:read"])
def apply_schema():
sql = (ROOT / "src" / "brick_gl" / "schema.sql").read_text()
stmts = re.findall(r"(CREATE TABLE IF NOT EXISTS.*?\);)", sql, re.S)
with db.engine.begin() as c:
for s in stmts:
c.exec_driver_sql(s)
print(f"schema applied: {len(stmts)} tables on {BUCKET}")
async def route(method, path, headers, body=None):
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://brick") as c:
r = await c.request(method, path, headers=headers, json=body) if method != "GET" else await c.get(path, headers=headers, params=body)
return r.status_code, r.json()
def api(*a):
return anyio.run(route, *a)
async def mcp(headers, name, args):
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()
res = await s.call_tool(name, args)
return res.is_error, json.loads(res.content[0].text) if not res.is_error else res.content[0].text
def main():
apply_schema()
run = str(int(time.time()))
code = "L" + run[-3:]
print("company", code)
ok = True
def step(label, status, detail=""):
nonlocal ok
good = status == 200
ok = ok and good
print(f"{'route' if label.startswith('route') else 'mcp '}: {label.split(': ', 1)[1]:<34} {status} {detail}")
return good
s, b = api("POST", "/charts", ACCOUNTANT, {"chart_id": "LIVE", "name": "Live chart", "kind": "operational"})
step("route: chart", s)
accounts = {}
for c_, name, typ, sub in (("1100", "Receivables", "asset", "receivable"), ("1200", "Bank", "asset", "bank"), ("2200", "VAT", "liability", "tax"),
("3000", "Retained earnings", "equity", "retained_earnings"), ("4000", "Revenue", "income", "other"),
("6100", "Rent", "expense", "other"), ("2100", "Accruals", "liability", "other")):
s, b = api("POST", "/accounts", ACCOUNTANT, {"chart_id": "LIVE", "code": c_, "name": name, "type": typ, "subtype": sub})
accounts[c_] = b.get("account_id")
step("route: accounts", s, f"{len(accounts)} accounts")
s, b = api("POST", "/companies", ACCOUNTANT, {"company_code": code, "name": "Live Ltd", "country": "GB", "functional_currency": "GBP", "chart_id": "LIVE"})
step("route: company", s, b.get("ledgers"))
for purpose, acc in (("receivable", "1100"), ("revenue", "4000"), ("bank", "1200"), ("retained_earnings", "3000"), ("rounding", "6100")):
s, _ = api("POST", "/account-map", ACCOUNTANT, {"chart_id": "LIVE", "purpose": purpose, "account_id": accounts[acc], "valid_from": "2026-01-01"})
step("route: account map", 200 if s in (200, 422) else s, "(422 on a rerun: the map row exists)" if s == 422 else "")
s, b = api("POST", "/fiscal-years/open-year", ACCOUNTANT, {"fiscal_year": "FY2026", "company_code": code})
step("route: open FY2026", s, f"{len(b.get('periods', []))} periods")
doc = f"INV-{run}"
body = {"journal_id": str(derived_id(doc, "issue")), "source_module": "ar", "source_document": doc, "event": "issue", "journal_date": "2026-10-15",
"currency": "GBP", "company_code": code, "memo": "live invoice",
"lines": [{"purpose": "receivable", "debit": "1200.00"}, {"purpose": "revenue", "credit": "1000.00"}, {"account_id": accounts["2200"], "credit": "200.00"}]}
s, b = api("POST", "/journal-entry/journals", POSTER, body)
step("route: ar posts an invoice", s, f"{b.get('number')} replay={b.get('replay')}")
s, again = api("POST", "/journal-entry/journals", POSTER, body)
step("route: the same call again", s, f"replay={again.get('replay')}")
ok = ok and again.get("replay") is True and again.get("journal_id") == b.get("journal_id")
s, o = api("GET", "/periods/open", POSTER, {"date": "2026-10-15", "company_code": code})
step("route: period open?", s, f"{o.get('period_id')} open={o.get('open')}")
s, tb = api("GET", "/reports/trial-balance", ACCOUNTANT, {"company_code": code, "as_at": "2026-10-31"})
step("route: trial balance", s, f"debit {tb.get('total_debit')} credit {tb.get('total_credit')} difference {tb.get('difference')}")
ok = ok and tb.get("total_debit") == "1200.00" and float(tb.get("difference")) == 0.0
# MCP
s, _ = api("POST", "/controls", OWNER, {"key": "manual_journal_threshold", "value": 100000})
err, j = anyio.run(mcp, ACCOUNTANT, "propose_manual", {"journal_date": "2026-10-20", "company_code": code, "memo": "live accrual",
"lines": [{"account_id": accounts["6100"], "debit": "300.00"}, {"account_id": accounts["2100"], "credit": "300.00"}]})
step("mcp: propose_manual", 200 if not err else 500, j.get("number") if not err else j)
err2, r = anyio.run(mcp, ACCOUNTANT, "get_journal", {"journal_id": j["journal_id"]})
step("mcp: get_journal", 200 if not err2 else 500, f"{r.get('status')} lines={len(r.get('lines', []))}")
err3, rv = anyio.run(mcp, ACCOUNTANT, "reverse_journal", {"journal_id": j["journal_id"], "reason": "live check"})
step("mcp: reverse_journal", 200 if not err3 else 500, rv.get("status") if not err3 else rv)
err4, t2 = anyio.run(mcp, ACCOUNTANT, "get_trial_balance", {"company_code": code, "as_at": "2026-10-31"})
step("mcp: get_trial_balance", 200 if not err4 else 500, f"difference {t2.get('difference')} rows={len(t2.get('rows', []))}")
print("LIVE CHECK", "PASSED" if ok else "FAILED")
return 0 if ok else 1
if __name__ == "__main__":
sys.exit(main())