dodil/erp-brick-fixed-assets
DODIL ERP brick: fixed assets
scripts/live_check.py
"""ONE live check: the brick alone on a scratch DataK3 bucket (default w2falab, org dodil) — apply the schema, then one happy-path
story through ROUTES (areas -> class -> asset -> purchasing hands a bill line over -> read the values -> simulate a run) and one through
MCP (tool list -> list_assets). The pg wire is real; the neighbour gl is the in-process stand-in of tests/conftest.py
(it answers exactly what gl's spec routes answer).
BUCKET=w2falab python scripts/live_check.py
The schema is applied here (not with `erp-compose`, 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 datetime
import json
import os
import pathlib
import re
import subprocess
import sys
ROOT = pathlib.Path(__file__).resolve().parents[1]
sys.path[:0] = [str(ROOT / "src"), str(ROOT / "tests")]
BUCKET = os.environ.get("BUCKET", "w2falab")
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 conftest import ACC, FakeGL # noqa: E402
from erp_platform import db, seam # noqa: E402
from mcp import ClientSession # noqa: E402
from mcp.client.streamable_http import streamable_http_client # noqa: E402
from brick_fixed_assets.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"}
OWNER = person("owner@live", ["fa:settings:controls"])
ACCOUNTANT = person("accountant@live", ["fa:asset:write", "fa:asset:read", "fa:depreciation:simulate", "fa:report:read"])
PURCHASING = person("purchasing@live", ["fa:acquisition:intake"])
SYSTEM = person("system@live", ["fa:number_series:claim"])
def apply_schema():
sql = (ROOT / "src" / "brick_fixed_assets" / "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)
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()
names = [t.name for t in (await s.list_tools()).tools]
res = await s.call_tool(name, args)
return names, res.is_error, json.loads(res.content[0].text) if not res.is_error else res.content[0].text
def main():
apply_schema()
gl = FakeGL()
seam.register_local("gl", gl.app)
today = datetime.date.today().isoformat()
stamp = datetime.datetime.now().strftime("%H%M%S")
# the owner's first rows may start today; a re-run finds them there (dated rows are never back-dated): 200 or 409 both fine
for path, body in (("/settings/areas", {"area_code": "01", "name": "Book", "purpose": "book", "ledger_id": "10",
"company_code": "UK01", "currency": "GBP", "valid_from": today}),
("/settings/keys", {"key_code": "LIN5", "method": "straight_line", "life_periods": 60, "valid_from": today}),
("/settings/classes", {"class_code": "VEH", "name": "Vehicles", "valid_from": today,
"areas": [{"area_code": "01", "key_code": "LIN5"}]}),
("/settings/account-determination", {"class_code": "VEH", "area_code": "01", "valid_from": today, **ACC})):
s, b = api("POST", path, OWNER, body)
assert s in (200, 409), (path, s, b)
s, a = api("POST", "/assets", ACCOUNTANT, {"request_id": f"live-{stamp}", "company_code": "UK01", "class_code": "VEH",
"description": f"Live van {stamp}"})
assert s == 200, a
ref = f"pur:LIVE-{stamp}:1"
s, acq = api("POST", "/acquisitions", PURCHASING, {"acquisition_ref": ref, "source_kind": "purchase_bill", "company_code": "UK01",
"bill_id": f"LIVE-{stamp}", "line_no": 1, "value_date": today,
"amount": "12000.00", "currency": "GBP", "asset_id": a["asset_id"]})
assert s == 200 and acq["replay"] is False, acq
s, got = api("GET", f"/assets/{a['asset_id']}", ACCOUNTANT)
v = next(x for x in got["values"] if x["area_code"] == "01")
assert s == 200 and v["cost"] == "12000.00" and v["net_book_value"] == "12000.00", got
assert len(gl.journals) == 1 and next(iter(gl.journals.values()))["ledger_id"] == "10"
s, r = api("POST", "/depreciation-runs", ACCOUNTANT, {"company_code": "UK01", "period_id": today[:7]})
assert s == 200 and r["status"] == "simulated", r
print("route: asset", a["asset_id"], "cost", v["cost"], "journals to gl", len(gl.journals), "run", r["run_id"], r["totals"])
names, err, listed = anyio.run(mcp, ACCOUNTANT, "list_assets", {})
assert not err and "propose_retirement" in names and "approve_asset_transaction" in names, (err, listed)
print(f"mcp: {len(names)} tools; list_assets ok")
print("LIVE OK")
if __name__ == "__main__":
main()