dodil/erp-brick-expenses
DODIL ERP brick: expenses
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 default role, the two doors - `api(...)` and `tool(...)` - and in-memory neighbours (tests/fakes.py) whose every call is
checked against the callee's signed spec."""
import base64
import json
import os
import pathlib
from contextlib import contextmanager
from datetime import datetime, timezone
import anyio
import httpx
import httpx2 # the MCP client's HTTP library
import pytest
import yaml
from erp_platform import contract as _contract
from erp_platform.contract import checked_seams
from fakes import Neighbours
from mcp import ClientSession
from mcp.client.streamable_http import streamable_http_client
from sqlalchemy import create_engine
from brick_expenses.app import app
from brick_expenses.application import use_cases
from brick_expenses.infrastructure.tables import Repository, md
# Every outgoing seam call is checked against the CALLEE's signed spec (erp-method). A missing checkout is an error, never a skip.
HERE = pathlib.Path(__file__).resolve().parent
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})")
def _names_us(entry):
return "expenses" in [w.strip() for w in str(entry.get("with", "")).split(",")]
_load_callee = _contract.load_callee
def _load_with_lists_split(callee, method_dir=None):
"""The callee's seams as they SHOULD list us (spec gaps, recorded in AGENTS.md; erp_platform.contract reads only `seams:` and `with:`):
hr lists one route `with: expenses, pos` - erp-platform 0.9.13 splits that itself, kept for older checkouts;
gl GET /fx-rates/{currency} names tax, consolidation and fixed-assets, not expenses (EXP-05 depends on it, GL-33);
tax POST /documents/compute (TAX-53) and /documents/{ref}/check-contents (TAX-50) name expenses under `external_services:`, not `seams:`."""
spec, shared = _load_callee(callee, method_dir)
seams = [{**e, "with": "expenses"} if _names_us(e) and callee == "hr" else e for e in spec.get("seams") or []]
if callee == "gl":
seams = [{**e, "with": f"{e['with']}, expenses"} if str(e.get("route", "")).startswith("GET /fx-rates/") else e for e in seams]
if callee == "tax":
seams += [e for e in spec.get("external_services") or [] if _names_us(e) and str(e.get("route", "")).startswith("POST /documents/")]
return {**spec, "seams": seams}, shared
_contract.load_callee = _load_with_lists_split
checked_seams = checked_seams(caller="expenses", method_dir=METHOD_DIR) # autouse: any real Seam call
ROLES = yaml.safe_load((HERE.parent / "src" / "brick_expenses" / "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)
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)
yield e
e.dispose()
def person(email, permissions, sub=None):
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": []}),
"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)
# The people of the specification's stories: employee 3301 claims; 2280 is their manager; 1904 is the manager's manager.
EMP, MGR, MGR2 = "emp3301@uk.example", "mgr2280@uk.example", "mgr1904@uk.example"
EMPLOYEE, MANAGER, MANAGER2 = as_role("employee", EMP), as_role("manager", MGR), as_role("manager", MGR2)
FINANCE, FINANCE2 = as_role("finance", "fin1@uk.example"), as_role("finance", "fin2@uk.example")
OWNER, SYSTEM = as_role("owner"), as_role("system")
HR_CLIENT, PAYMENTS_CLIENT = as_role("hr_client", "hr-service@uk.example"), as_role("payments_client", "payments-service@uk.example")
NOBODY = person("nobody@uk.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)
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)
# ── a frozen business clock, the neighbours, and small builders the tests share ─────────────────
@pytest.fixture(autouse=True)
def clock():
"""Business 'today' is 2026-11-07 (the day 3301 submits in EXP-08), 09:00 UTC = 09:00 in London."""
from erp_platform import work
work.set_clock(at=datetime(2026, 11, 7, 9, 0, tzinfo=timezone.utc), 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."""
from erp_platform import work
before = work.now()
work.set_clock(at=datetime.fromisoformat(when), freeze=True)
try:
yield
finally:
work.set_clock(at=before, freeze=True)
@pytest.fixture(autouse=True)
def nb():
"""hr, the ledger and payments, in memory: employee 3301 (UK01, SALES-N), managed by 2280, whose manager is 1904."""
from erp_platform import seam
n = Neighbours()
n.employee(3301, EMP, manager=n.as_manager(2280, MGR), manager2=n.as_manager(1904, MGR2))
n.employee(2280, MGR, manager=n.as_manager(1904, MGR2))
n.employee(1904, MGR2)
seam.register_local("hr", n.hr)
seam.register_local("gl", n.gl)
seam.register_local("payments", n.payments)
yield n
for name in ("hr", "gl", "payments", "tax", "ar"):
try:
seam.unregister_local(name)
except Exception:
pass
def must(result, status=200):
"""A call that has to succeed: (status, body) -> body."""
got, body = result
assert got == status, (got, body)
return body
def refused(result, status=409, words=""):
got, body = result
assert got == status, (got, body)
assert words in json.dumps(body), (words, body)
return body
def new_report(who=None, currency="GBP", title="Manchester client visit"):
return must(api("POST", "/reports", who or EMPLOYEE, {"currency": currency, "title": title}), 201)
def add(report_id, who=None, **kw):
line = {"category_code": "TRAVEL_RAIL", "expense_date": "2026-11-04", "merchant": "Northern Rail", "amount": "86.40",
"receipt_key": f"raw/expenses/{report_id}/1/" + "a" * 64 + ".jpg", **kw}
return must(api("POST", f"/reports/{report_id}/lines", who or EMPLOYEE, line), 201)
def claim_551(who=None):
"""The story claim, in the own-money lines the first slice can hold: train 86.40, hotel 264.00, client dinner 96.00, taxi 105.00 = 551.40."""
r = new_report(who)
rid = r["report_id"]
add(rid, who, category_code="TRAVEL_RAIL", merchant="Northern Rail", amount="86.40")
add(rid, who, category_code="ACCOMMODATION", merchant="Manchester Hotel", amount="264.00", vat_amount="44.00", expense_date="2026-11-05")
add(rid, who, category_code="MEALS_CLIENT", merchant="The Ivy", amount="96.00", vat_amount="16.00")
add(rid, who, category_code="TRAVEL_RAIL", merchant="Trains to Leeds", amount="105.00")
return rid
def submit(rid, who=None):
return must(api("POST", f"/reports/{rid}/submit", who or EMPLOYEE))
def approve(rid, who):
return api("POST", f"/reports/{rid}/approve", who)
def report(rid, who=None):
return must(api("GET", f"/reports/{rid}", who or EMPLOYEE))
def seed_card(card_id="4417", holder=3301, txns=(), company="UK01", status="active", holder_email=None):
"""A company card and its imported transactions, straight in the tables (the feed import has its own tests): txns = [(ref, posted_on, merchant,
amount)] in GBP."""
from datetime import date
from erp_platform import work
with use_cases.repository() as repo:
repo.cards.put_card({"card_id": card_id, "company_code": company, "holder_person_id": holder, "provider": "Visa", "currency": "GBP",
"account_ref": "CORP-UK", "status": status, "holder_email": holder_email}, work.now())
repo.cards.add_txns([{"card_txn_id": f"card:{card_id}:{ref}", "card_id": card_id, "company_code": company, "holder_person_id": holder,
"posted_on": date.fromisoformat(posted), "merchant": merchant, "billed_amount": amount, "billed_currency": "GBP",
"original_amount": amount, "original_currency": "GBP", "status": "imported", "statement_ref": "2026-11"}
for ref, posted, merchant, amount in txns], work.now())
def policy(who, **kw):
"""finance puts a policy row (a dated rule)."""
body = {"company_code": "UK01", "valid_from": "2026-11-07", "scope": "per item", "mode": "warn", **kw}
return must(api("POST", "/policy-rules", who, body), 201)
@pytest.fixture(autouse=True)
def store():
"""The receipt store: a dict (the S3 client is brick-local, see infrastructure/objects.py)."""
from brick_expenses.infrastructure import objects
objects.ports.store = objects.MemoryStore()
yield objects.ports.store
objects.ports.store = None
JPEG = b"\xff\xd8\xff\xe0" + b"fake jpeg " * 20
def image(body=JPEG, kind="image/jpeg"):
return {"content_type": kind, "content_base64": base64.b64encode(body).decode()}