dodil/erp-brick-payments
DODIL ERP brick: payments
tests/conftest.py
"""Shared wiring: the real app with storage on a SQLite file (each test its own), personas that carry exactly the
permissions given, the two doors - `api(...)` and `tool(...)` - and in-memory neighbours (tests/fakes.py)."""
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.contract import Recorder, 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_payments.app import app
from brick_payments.application import use_cases
from brick_payments.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})")
checked_seams = checked_seams(caller="payments", method_dir=METHOD_DIR) # autouse: any real Seam call
_CONTRACT = Recorder()
@pytest.fixture(autouse=True)
def contract():
_CONTRACT.violations.clear()
yield _CONTRACT
assert not _CONTRACT.violations, "\n".join(_CONTRACT.violations)
ROLES = yaml.safe_load((pathlib.Path(__file__).resolve().parents[1] / "src" / "brick_payments" / "roles.yaml").read_text())["roles"]
ALL = sorted({p for perms in ROLES.values() for p in perms})
@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, roles=()):
payload = base64.urlsafe_b64encode(json.dumps({"permissions": list(permissions)}).encode()).decode()
return {"X-Dodil-User": json.dumps({"sub": email, "email": email, "app_roles": list(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])
CASHIER, COLLECTOR, APPROVER, SALES, SYSTEM = (as_role(r) for r in ("cashier", "collector", "approver", "sales", "system"))
APPROVER2 = as_role("approver", "owner2@shop.example")
NOBODY = person("nobody@shop.example", [])
BOSS = person("boss@shop.example", ALL)
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-20 in every test."""
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 nb():
"""Invoicing, the partner register and the ledger, in memory."""
from erp_platform import seam
n = Neighbours()
seam.register_local("ar", n.ar)
seam.register_local("business-partner", n.bp)
seam.register_local("gl", n.gl)
seam.register_local("tax", n.tax)
seam.register_local("purchasing", n.purchasing)
yield n
for name in ("ar", "business-partner", "gl", "tax", "purchasing"):
seam.unregister_local(name)
@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)
def must(result):
"""A call that has to succeed: (status, body) -> body."""
status, body = result
assert status == 200, (status, 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 bank(code="BANK1", gl="1200", currency="GBP", who=None, **more):
body = {"name": "Current account", "currency": currency, "account_number": "12345678", "company_code": "UK01",
**({"gl_account_id": gl} if gl else {}), **more}
return must(api("POST", f"/bank-accounts/{code}", who or APPROVER, body))
def receipt(bp_id=7311, amount="10000.00", currency="GBP", received_on="2026-11-20", reference="INV-2026-0041 0042", who=None, **kw):
return must(api("POST", "/receipts", who or CASHIER, {"bp_id": bp_id, "amount": amount, "currency": currency,
"received_on": received_on, "bank_reference": reference, **kw}))
def allocate(payment_id, lines, who=None):
return api("POST", f"/receipts/{payment_id}/allocate", who or CASHIER,
{"lines": [{"invoice_id": i, "amount": a} for i, a in lines]})
def settle(*ids, as_at=None, who=None):
body = must(api("GET", "/settlement", who or CASHIER, {"invoice_ids": ",".join(ids), **({"as_at": as_at} if as_at else {})}))
return {i["invoice_id"]: i for i in body["invoices"]}
def set_control(key, value, who=None):
return must(api("POST", f"/controls/{key}", who or APPROVER, {"value": value}))
def three_invoices(nb, bp_id=7311):
nb.invoice("INV-2026-0041", bp_id, "4000.00", number="INV-2026-0041", due_date="2026-11-10")
nb.invoice("INV-2026-0042", bp_id, "3500.00", number="INV-2026-0042", due_date="2026-11-12")
nb.invoice("INV-2026-0043", bp_id, "3000.00", number="INV-2026-0043", due_date="2026-11-15")