dodil/erp-brick-ar
DODIL ERP brick: accounts receivable
tests/conftest.py
"""Shared wiring: the real app with storage on a SQLite file (each test its own), personas that carry exactly the permissions of
a role (no role name is ever sent), the two doors — `api(...)` and `tool(...)` — a frozen business clock, and the neighbours."""
import base64
import json
import os
import pathlib
import sys
from contextlib import contextmanager
from datetime import datetime, timezone
from decimal import Decimal
import anyio
import httpx
import httpx2 # the MCP client's HTTP library
import pytest
import yaml
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
sys.path.insert(0, str(pathlib.Path(__file__).parent))
from fakes import World # noqa: E402
from brick_ar.app import app # noqa: E402
from brick_ar.application import use_cases # noqa: E402
from brick_ar.infrastructure.tables import Repository, md # noqa: E402
# 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(pathlib.Path(__file__).resolve().parents[2] / "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="ar", method_dir=METHOD_DIR) # autouse: any real Seam call
_ROLES = yaml.safe_load((pathlib.Path(__file__).resolve().parents[1] / "src" / "brick_ar" / "roles.yaml").read_text())["roles"]
@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)
from erp_platform import idempotency
idempotency.md.create_all(e)
with e.begin() as conn:
Repository(conn).seed()
@contextmanager
def repo():
with e.begin() as conn:
yield Repository(conn)
monkeypatch.setattr(use_cases, "repository", repo)
use_cases.AR.idempotency_connection = e.begin
yield e
use_cases.AR.idempotency_connection = None
e.dispose()
@pytest.fixture(autouse=True)
def clock():
"""Business 'today' is 2026-11-20 (09:00 UTC) in every test: rows dated before it are history, after it are scheduled."""
from erp_platform import work
work.set_clock(at=datetime(2026, 11, 20, 9, 0, tzinfo=timezone.utc), freeze=True)
yield
work.reset_clock()
@pytest.fixture(autouse=True)
def no_neighbours():
"""Nothing is installed unless the test asks (`world`)."""
from erp_platform import seam
for n in ("tax", "payments", "gl", "business-partner", "sales-order", "product", "purchasing", "peppol"):
seam.unregister_local(n)
yield
@pytest.fixture
def world():
w = World().everything()
yield w
w.remove("tax", "payments", "gl", "business-partner", "sales-order", "product", "purchasing", "peppol")
def person(email, permissions):
payload = base64.urlsafe_b64encode(json.dumps({"permissions": list(permissions)}).encode()).decode()
return {"X-Dodil-User": json.dumps({"sub": email, "email": email, "app_roles": []}),
"X-Dodil-User-JWT": f"h.{payload.rstrip('=')}.s"}
def as_role(role, email=None):
return person(email or f"{role}@shop.example", _ROLES[role])
ROLES = _ROLES
BILLER, COLLECTOR, APPROVER, OWNER, SALES, SYSTEM = (as_role(r) for r in ("biller", "collector", "approver", "owner", "sales", "system"))
NOBODY = person("nobody@shop.example", [])
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)
try:
return r.status_code, r.json()
except ValueError:
return r.status_code, {"detail": r.text}
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 ": " in text else text) if res.is_error else ("ok", json.loads(text))
def tools(headers):
return sorted(t.name for t in anyio.run(_mcp, headers, lambda s: s.list_tools()).tools)
def must(result):
status, body = result
assert status == 200, (status, body)
return body
# ── small builders the tests share ────────────────────────────────────────────────────────────
BP = 7311
def line(desc="Managed hosting", qty="2", price="100.00", **kw):
return {"description": desc, "quantity": qty, "unit_price": price, "unit_code": "MON", "tax_category": "standard", **kw}
def billing(bp=BP, **kw):
"""The customer's billing profile, in force from today (a customer's first row may start today)."""
body = {"bill_to_name": "Acme Ltd", "bill_to_address": "1 High St, London", "invoice_email": "ap@acme.example",
"terms_id": "NET30", "currency": "GBP", **kw}
return must(api("POST", f"/customers/{bp}/billing", APPROVER, body))
def hand_draft(lines=None, bp=BP, **kw):
return must(api("POST", "/invoices", BILLER, {"bp_id": bp, "currency": "GBP", "lines": lines or [line()], **kw}))
def issue(invoice_id, who=None, world=None, version=None):
body = must(api("POST", f"/invoices/{invoice_id}/issue", who or BILLER, {} if version is None else {"version": version}))
if world is not None and world.payments.app is not None:
world.payments.totals[invoice_id] = Decimal(body["total"])
return body
def issued(world=None, lines=None, bp=BP, **kw):
"""A customer with a profile, a hand-raised draft, issued: the invoice as the customer received it."""
billing(bp)
d = hand_draft(lines, bp, **kw)
return issue(d["invoice_id"], world=world)
def raise_so(source_id="so:SO-1:1", lines=None, bp=BP, who=SYSTEM, **kw):
body = {"source_kind": "sales_order", "source_id": source_id, "bp_id": bp, "currency": "GBP",
"lines": lines or [line(sales_order_id="SO-1", so_line_no=1)], **kw}
return api("POST", "/invoices/raise", who, body)
@contextmanager
def on_day(iso, hour=9):
"""Run a block with the business clock at a day (09:00 UTC), and put it back."""
from erp_platform import work
before = work.now()
work.set_clock(at=datetime.fromisoformat(f"{iso}T{hour:02d}:00:00+00:00"), freeze=True)
try:
yield
finally:
work.set_clock(at=before, freeze=True)
def control(key, value, valid_from=None, who=None):
body = {"key": key, "value": value, **({"valid_from": valid_from} if valid_from else {})}
return api("POST", "/controls", who or OWNER, body)
def account_map():
"""receivable 1100, revenue 4000, VAT S 2200 / Z 2210 — in force from today (each is the first row of its key)."""
for purpose, code, acct in (("receivable", None, 1100), ("revenue", None, 4000), ("tax", "S", 2200), ("tax", "Z", 2210),
("bad_debt", None, 6500), ("deposit", None, 2300)):
must(api("POST", "/account-map", APPROVER, {"purpose": purpose, "account_id": acct, **({"tax_code": code} if code else {})}))
def approved_credit(world, invoice_id, **body):
d = must(api("POST", "/credit-notes", BILLER, {"invoice_id": invoice_id, "reason": "service credit", **body}))
world.payments.credit_faces[d["credit_id"]] = Decimal(d["total"])
return must(api("POST", f"/credit-notes/{d['credit_id']}/approve", APPROVER, {}))
@pytest.fixture
def odoo(monkeypatch, world):
"""A small fake Odoo, its partners and taxes already mapped by business-partner and tax's migrations, and payments holding the
open balances as Odoo had them (payments migrates receipts and reconciliations first)."""
from fakes import Odoo
o = Odoo()
monkeypatch.setattr(use_cases, "odoo_reader", lambda: o)
world.bp.migration_map.update({10: 7311, 11: 7312})
world.tax.migration_map.update({1: "S", 2: "Z"})
return o
def migrate_payments(world, odoo):
"""payments' migration has run: each migrated invoice is registered with Odoo's total and what Odoo had settled."""
for m in odoo.move_rows:
if m["move_type"] == "out_invoice" and m["state"] == "posted":
iid = f"odoo:{odoo.db}:{m['id']}"
world.payments.totals[iid] = Decimal(str(m["amount_total"])).quantize(Decimal("0.01"))
paid = Decimal(str(m["amount_total"])).quantize(Decimal("0.01")) - Decimal(str(m["amount_residual"])).quantize(Decimal("0.01"))
if paid:
world.payments.receive(iid, str(paid), ref=f"odoo-pay-{m['id']}")