dodil/erp-brick-fixed-assets
DODIL ERP brick: fixed assets
tests/conftest.py
"""Shared wiring: the real app with storage on a SQLite file (each test its own), personas that carry exactly the permissions
the brick's default roles give them, the two doors — `api(...)` and `tool(...)` — and a stand-in gl that answers exactly
what gl's SPEC routes answer (shape and status codes): a fake that disagrees with the real callee is the bug that hides a
broken seam."""
import base64
import json
import os
import pathlib
import re
from contextlib import contextmanager
from datetime import datetime
from decimal import Decimal
import anyio
import httpx
import httpx2 # the MCP client's HTTP library
import pytest
import yaml
from erp_platform import seam, work
from erp_platform.contract import checked_seams
from mcp import ClientSession
from mcp.client.streamable_http import streamable_http_client
from sqlalchemy import create_engine
from brick_fixed_assets.app import app
from brick_fixed_assets.application import use_cases
from brick_fixed_assets.infrastructure import business_time
from brick_fixed_assets.infrastructure.tables import Repository, md
HERE = pathlib.Path(__file__).resolve().parent
ROLES = yaml.safe_load((HERE.parent / "src" / "brick_fixed_assets" / "roles.yaml").read_text())["roles"]
# Every outgoing seam call is checked against the CALLEE's signed spec (erp-method). A missing checkout is an error, never a skip.
METHOD_DIR = os.environ.get("ERP_METHOD_DIR") or str(HERE.parents[1] / "erp-method")
if not (pathlib.Path(METHOD_DIR) / "modules" / "shared.yaml").exists():
raise RuntimeError(f"seam contract checks need an erp-method checkout: set ERP_METHOD_DIR (tried {METHOD_DIR})")
checked_seams = checked_seams(caller="fixed-assets", method_dir=METHOD_DIR) # autouse; pytest collects it from here
TODAY = "2026-11-20T09:00:00+00:00" # business 'today' is 2026-11-20 (Europe/London, GMT)
@pytest.fixture(autouse=True)
def engine(tmp_path, monkeypatch):
e = create_engine(f"sqlite:///{tmp_path / 'bucket.db'}", connect_args={"check_same_thread": False})
md.create_all(e)
@contextmanager
def repo():
with e.begin() as conn:
yield Repository(conn)
monkeypatch.setattr(use_cases, "repository", repo)
monkeypatch.setattr(business_time, "SOURCE", lambda conn: [])
monkeypatch.delenv("BUSINESS_TIMEZONE", raising=False)
yield e
e.dispose()
@pytest.fixture(autouse=True)
def clock():
work.set_clock(at=datetime.fromisoformat(TODAY), freeze=True)
yield
work.reset_clock()
@contextmanager
def on_day(when):
"""Run a block with the business clock at `when` (an ISO instant), and put it back."""
before = work.now()
work.set_clock(at=datetime.fromisoformat(when), freeze=True)
try:
yield
finally:
work.set_clock(at=before, freeze=True)
def person(email, permissions, sub=None, roles=()):
payload = base64.urlsafe_b64encode(json.dumps({"permissions": list(permissions)}).encode()).decode()
return {"X-Dodil-User": json.dumps({"sub": sub or email, "email": email, "app_roles": list(roles)}),
"X-Dodil-User-JWT": f"h.{payload.rstrip('=')}.s"}
def as_role(role, email=None, sub=None):
return person(email or f"{role}@uk.example", ROLES[role], sub=sub)
ACCOUNTANT, APPROVER, TAX, CUSTODIAN, OWNER, SYSTEM = (as_role(r) for r in (
"asset_accountant", "finance_approver", "tax_preparer", "asset_custodian", "owner", "system"))
APPROVER2 = as_role("finance_approver", "second.approver@uk.example")
ACCOUNTANT2 = as_role("asset_accountant", "second.accountant@uk.example")
# the neighbours' service clients: purchasing hands bills over; gl asks what is posted and the balances
PURCHASING = person("purchasing@svc.example", ["fa:acquisition:intake", "fa:asset:read", "fa:asset:read_own"])
GL_CLIENT = person("gl@svc.example", ["fa:depreciation:read", "fa:asset:read"])
ALL = sorted({p for perms in ROLES.values() for p in perms if p not in use_cases.MACHINE})
async def _call(method, path, headers, body=None):
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(transport=transport, base_url="http://brick") as c:
if method == "GET":
return await c.get(path, headers=headers, params=body)
return await c.request(method, path, headers=headers, json=body)
def api(method, path, headers, body=None):
r = anyio.run(_call, method, path, headers, body)
return r.status_code, r.json()
async def _mcp(headers, fn):
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 await fn(s)
def tool(headers, name, args):
res = anyio.run(_mcp, headers, lambda s: s.call_tool(name, args))
text = res.content[0].text
return ("refused", text.split(": ", 1)[1]) if res.is_error else ("ok", json.loads(text))
def tools(headers):
listed = anyio.run(_mcp, headers, lambda s: s.list_tools())
return sorted(t.name for t in listed.tools)
def must(result):
"""A call that has to succeed: (status, body) -> body."""
status, body = result
assert status == 200, (status, body)
return body
# ── a stand-in gl: exactly what gl's SPEC routes answer ───────────────────────────────────────
class FakeGL:
"""POST /journal-entry/journals, GET /periods/open, GET /companies/{c}/ledgers, GET /fx-rates, GET /migration-map.
201 {journal_id, entry_no, period_id, status: posted, replay: false, lines}; 200 the same with replay: true; 409
{gate: period_open | journal_conflict}; 422 {gate: balanced | purposes_mapped | ledger_assigned | bounds}."""
def __init__(self):
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
self.journals, self.calls = {}, []
self.closed = set() # (ledger_id, "YYYY-MM") closed
self.ledgers = {"UK01": [{"ledger_id": "10", "kind": "leading", "accounting_principle": "UK GAAP"},
{"ledger_id": "20", "kind": "non_leading", "accounting_principle": "IFRS"}],
"UK02": [{"ledger_id": "11", "kind": "leading", "accounting_principle": "UK GAAP"},
{"ledger_id": "21", "kind": "non_leading", "accounting_principle": "IFRS"}]}
self.fx = {}
self.map = {}
self.fail_next = None
self.app = FastAPI()
gl = self
@self.app.post("/journal-entry/journals")
async def post(request: Request):
b = await request.json()
gl.calls.append(b)
if gl.fail_next:
code, body = gl.fail_next
gl.fail_next = None
return JSONResponse(body, status_code=code)
d = sum(Decimal(x["debit"]) for x in b["lines"] if "debit" in x)
c = sum(Decimal(x["credit"]) for x in b["lines"] if "credit" in x)
if d != c:
return JSONResponse({"gate": "balanced", "reason": f"debits {d} credits {c}", "fix": "balance it"}, 422)
if len(b["lines"]) > 5000:
return JSONResponse({"gate": "bounds", "reason": "more than 5,000 lines", "fix": "split by class"}, 422)
expected = seam.derived_id(b["source_document"], b["event"], b["ledger_id"]) if b.get("ledger_id") \
else seam.derived_id(b["source_document"], b["event"])
if str(b["journal_id"]) != str(expected): # gl's own gate: the id is derived from the source, never invented
return JSONResponse({"gate": "journal_id", "reason": f"journal_id {b['journal_id']} is not derived_id("
f"source_document, event[, ledger_id]) = {expected}", "fix": "derive it"}, 422)
ledger = str(b["ledger_id"])
if not re.fullmatch(r"[A-Z0-9]{2,8}", ledger):
return JSONResponse({"gate": "ledger_assigned", "reason": "ledger_id is a ledger code", "fix": "send the code"}, 422)
period = b["journal_date"][:7]
if (ledger, period) in gl.closed:
return JSONResponse({"gate": "period_open", "reason": f"{period} is closed in ledger {ledger}",
"fix": "reopen the period"}, 409)
if b["journal_id"] in gl.journals:
old = gl.journals[b["journal_id"]]
if old["lines"] != b["lines"]:
return JSONResponse({"gate": "journal_conflict", "reason": "same id, different lines", "fix": "x"}, 409)
return JSONResponse({"journal_id": b["journal_id"], "entry_no": old["entry_no"], "period_id": period,
"status": "posted", "replay": True, "lines": len(b["lines"])}, 200)
gl.journals[b["journal_id"]] = {**b, "entry_no": len(gl.journals) + 1}
return JSONResponse({"journal_id": b["journal_id"], "entry_no": len(gl.journals), "period_id": period,
"status": "posted", "replay": False, "lines": len(b["lines"])}, 201)
@self.app.get("/periods/open")
async def periods(date: str, company_code: str = "UK01", ledger_id: str = "10"):
p = date[:7]
is_open = (str(ledger_id), p) not in gl.closed
return {"date": date, "company_code": company_code, "period_id": p, "open": is_open,
"status": "open" if is_open else "closed", "reason": "" if is_open else f"{p} is closed",
"fix": "" if is_open else "reopen the period"}
@self.app.get("/companies/{company_code}/ledgers")
async def ledgers(company_code: str):
if company_code not in gl.ledgers:
return JSONResponse({"detail": "unknown company"}, 404)
return {"company_code": company_code, "functional_currency": "GBP", "fiscal_year_end_month": 12,
"ledgers": gl.ledgers[company_code]}
@self.app.get("/fx-rates/{currency}")
async def fx(currency: str, on: str = "", to: str = "GBP", type: str = "spot"):
if currency not in gl.fx:
return JSONResponse({"gate": "fx_rate", "reason": "no rate", "fix": "enter it"}, 404)
return {"currency": currency, "to": to, "type": type, "on": on, "rate_date": on, "rate": gl.fx[currency],
"source": "test"}
@self.app.get("/migration-map/{db}/{model}")
async def mmap(db: str, model: str, ids: str = ""):
return [{"odoo_model": model, "odoo_id": i, "ours": gl.map[(model, int(i))]} for i in ids.split(",")
if i and (model, int(i)) in gl.map]
def lines_of(self, source_document_prefix=None, event=None):
return [j for j in self.journals.values()
if (source_document_prefix is None or j["source_document"].startswith(source_document_prefix))
and (event is None or j["event"] == event)]
def balance(self, account_id, ledger_id=None):
"""Debits minus credits posted to an account (optionally one ledger)."""
total = Decimal(0)
for j in self.journals.values():
if ledger_id is not None and str(j["ledger_id"]) != str(ledger_id):
continue
for ln in j["lines"]:
if str(ln.get("account_id")) == str(account_id):
total += Decimal(ln.get("debit", "0")) - Decimal(ln.get("credit", "0"))
return total
@pytest.fixture
def gl():
g = FakeGL()
seam.register_local("gl", g.app)
yield g
seam.unregister_local("gl")
# ── the setup most tests need ─────────────────────────────────────────────────────────────────
ACC = {"cost_account_id": 150, "accumulated_depreciation_account_id": 151, "depreciation_expense_account_id": 771,
"gain_account_id": 491, "loss_account_id": 791, "revaluation_reserve_account_id": 320,
"impairment_account_id": 772, "impairment_reversal_account_id": 492}
def setup_policy(gl_fake=None, ifrs=True, tax=True, classes=("VEH",), start="2026-01-01"):
"""The owner records areas 01 -> ledger 10, 30 (IFRS, revaluation model) -> ledger 20, CA (UK tax), the key LIN5 and the
classes with their accounts — all from `start`, the first row of each (which may start on `start` when it is today or
later; the test clock is 2026-11-20, so rows are dated today)."""
with on_day(start + "T09:00:00+00:00"):
must(api("POST", "/settings/areas", OWNER, {"area_code": "01", "name": "Book", "purpose": "book", "ledger_id": "10",
"company_code": "UK01", "currency": "GBP"}))
if ifrs:
must(api("POST", "/settings/areas", OWNER, {"area_code": "30", "name": "IFRS", "purpose": "ifrs",
"ledger_id": "20", "company_code": "UK01", "currency": "GBP",
"value_model": "revaluation"}))
if tax:
must(api("POST", "/settings/areas", OWNER, {"area_code": "CA", "name": "Tax", "purpose": "tax"}))
must(api("POST", "/settings/keys", OWNER, {"key_code": "LIN5", "method": "straight_line", "life_periods": 60}))
must(api("POST", "/settings/keys", OWNER, {"key_code": "LIN4", "method": "straight_line", "life_periods": 48}))
areas = [{"area_code": "01", "key_code": "LIN5"}]
if ifrs:
areas.append({"area_code": "30", "key_code": "LIN5"})
if tax:
areas.append({"area_code": "CA", "key_code": "LIN5"})
for c in classes:
must(api("POST", "/settings/classes", OWNER, {"class_code": c, "name": c.title(), "areas": areas,
"default_ca_pool": "main"}))
for code in ("01", "30") if ifrs else ("01",):
must(api("POST", "/settings/account-determination", OWNER, {"class_code": c, "area_code": code, **ACC}))
must(api("POST", "/settings/controls", OWNER, {"setting_key": "intercompany_clearing_account_id", "value": "2900"}))
must(api("POST", "/settings/controls", OWNER, {"setting_key": "asset_sale_clearing_account_id", "value": "1300"}))
def create_asset(request_id="req-1", company="UK01", cls="VEH", description="Ford Transit", who=None, **kw):
return must(api("POST", "/assets", who or ACCOUNTANT, {"request_id": request_id, "company_code": company,
"class_code": cls, "description": description, **kw}))
def acquire(asset_id, amount="12000.00", value_date="2026-10-20", ref="pur:BILL-1:1", **kw):
"""Purchasing hands over a bill line (POST /acquisitions)."""
body = {"acquisition_ref": ref, "source_kind": "purchase_bill", "company_code": "UK01", "bill_id": ref.split(":")[1],
"line_no": int(ref.split(":")[2]), "value_date": value_date, "amount": amount, "currency": "GBP",
"asset_id": asset_id, **kw}
return api("POST", "/acquisitions", PURCHASING, body)