dodil/erp-brick-purchasing
DODIL ERP brick: purchasing
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
neighbours as in-memory fakes that answer their SPEC routes, and the two doors — `api(...)` and `tool(...)`."""
import base64
import json
import os
import pathlib
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 cryptography.fernet import Fernet
from erp_platform import contract, idempotency, work
from fakes import (
FakeAr,
FakeBp,
FakeFa,
FakeGl,
FakeInventory,
FakeMaps,
FakeObjects,
FakeOdoo,
FakePayments,
FakeProduct,
FakeTax,
)
from mcp import ClientSession
from mcp.client.streamable_http import streamable_http_client
from sqlalchemy import create_engine
from brick_purchasing.app import app
from brick_purchasing.application import base
from brick_purchasing.infrastructure import seams
from brick_purchasing.infrastructure.tables import Repository, md
D = Decimal
ROLES_FILE = pathlib.Path(__file__).resolve().parents[1] / "src" / "brick_purchasing" / "roles.yaml"
_ROLES = yaml.safe_load(ROLES_FILE.read_text())["roles"]
ALL = sorted({p for perms in _ROLES.values() for p in perms})
@pytest.fixture(autouse=True)
def bank_key(monkeypatch):
monkeypatch.setenv("PURCHASING_BANK_KEY", Fernet.generate_key().decode())
@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)
idempotency.md.create_all(e)
monkeypatch.setattr(base.PURCHASING, "idempotency_connection", e.begin)
with e.begin() as conn:
Repository(conn).seed()
@contextmanager
def repo():
with e.begin() as conn:
yield Repository(conn)
monkeypatch.setattr(base, "repository", repo)
yield e
e.dispose()
@pytest.fixture(autouse=True)
def neighbours(monkeypatch):
"""Every neighbour an in-memory fake; tests reach them as N.product, N.tax, N.gl, N.inv, ..."""
class N:
bp, product, tax, gl, inventory = FakeBp(), FakeProduct(), FakeTax(), FakeGl(), FakeInventory()
payments, ar, fa, objects, odoo, maps = FakePayments(), FakeAr(), FakeFa(), FakeObjects(), FakeOdoo(), FakeMaps()
for name in ("bp", "product", "tax", "gl", "inventory", "payments", "ar", "fa", "objects", "odoo", "maps"):
monkeypatch.setattr(seams.ports, name, getattr(N, name))
return N
# Every call a port fake receives is checked against the CALLEE's signed spec (erp-method). A missing checkout is an
# error, never a skip. (callee module, port attribute, method, spec path) for each write the brick makes.
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})")
GUARDS = [("gl", "post_journal", "gl", "/journal-entry/journals"), ("inventory", "receive", "inventory", "/receipts"),
("inventory", "return_to_supplier", "inventory", "/returns-to-supplier"),
("inventory", "subcontract_provision", "inventory", "/subcontract-provisions"),
("inventory", "valuation_adjustment", "inventory", "/valuation-adjustments"),
("fa", "acquisition", "fixed-assets", "/acquisitions")]
@pytest.fixture(autouse=True)
def _contract(neighbours, monkeypatch):
rec = contract.Recorder()
for attr, meth, callee, path in GUARDS:
contract.guard(monkeypatch, getattr(neighbours, attr), meth, callee=callee, method="POST", path=path,
caller="purchasing", method_dir=METHOD_DIR, recorder=rec)
pay, inner = neighbours.payments, neighbours.payments.account_details
def account_details(bank_account_id, purpose): # a GET: checked as path + query against payments' PAY-27
rec.verify("payments", "GET", f"/bank-accounts/{bank_account_id}/account-details", {"purpose": purpose}, "purchasing", METHOD_DIR)
return inner(bank_account_id, purpose)
monkeypatch.setattr(pay, "account_details", account_details)
tax = neighbours.tax
inner_check, inner_freeze = tax.check_contents, tax.freeze
def check_contents(document_ref, contents): # tax's TAX-24, with the v8 contents
rec.verify("tax", "POST", f"/documents/{document_ref}/check-contents", {"direction": "purchase", "contents": contents}, "purchasing", METHOD_DIR)
return inner_check(document_ref, contents)
def freeze(document_ref, contents=None):
rec.verify("tax", "POST", f"/documents/{document_ref}/freeze", {"contents": contents} if contents else {}, "purchasing", METHOD_DIR)
return inner_freeze(document_ref, contents)
monkeypatch.setattr(tax, "check_contents", check_contents)
monkeypatch.setattr(tax, "freeze", freeze)
gl, inner_cc, inner_co = neighbours.gl, neighbours.gl.cost_centre, neighbours.gl.company
def cost_centre(cc):
rec.verify("gl", "GET", f"/cost-centres/{cc}", None, "purchasing", METHOD_DIR)
return inner_cc(cc)
def company(code):
rec.verify("gl", "GET", f"/companies/{code}", None, "purchasing", METHOD_DIR)
return inner_co(code)
monkeypatch.setattr(gl, "cost_centre", cost_centre)
monkeypatch.setattr(gl, "company", company)
yield
assert not rec.violations, "\n".join(sorted(set(rec.violations)))
@pytest.fixture(autouse=True)
def clock():
"""Business 'today' is 2026-11-20 in every test (tests move it with `on_day`)."""
work.set_clock(at=datetime(2026, 11, 20, 9, 0, tzinfo=timezone.utc), freeze=True)
yield
work.reset_clock()
@contextmanager
def on_day(when):
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, roles=(), 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": list(roles)}),
"X-Dodil-User-JWT": f"h.{payload.rstrip('=')}.s"}
def as_role(role, email=None, sub=None, extra=()):
return person(email or f"{role}@shop.example", list(_ROLES[role]) + list(extra), sub=sub)
ALICE = person("alice@shop.example", ALL)
NOBODY = person("nobody@shop.example", [])
REQUESTER, REQUESTER2 = as_role("requester", "requester@shop.example"), as_role("requester", "req2@shop.example")
APPROVER, APPROVER2, DIRECTOR = (as_role("spend_approver", "approver@shop.example"), as_role("spend_approver", "approver2@shop.example"),
as_role("spend_approver", "director@shop.example"))
BUYER, BUYER2 = as_role("buyer", "buyer@shop.example"), as_role("buyer", "buyer2@shop.example")
RECEIVER = as_role("receiver", "receiver@shop.example")
CLERK, CLERK2 = as_role("payables_clerk", "clerk@shop.example"), as_role("payables_clerk", "clerk2@shop.example")
PAYAPPROVER, PAYAPPROVER2 = as_role("payables_approver", "finance@shop.example"), as_role("payables_approver", "finance2@shop.example")
OWNER, SYSTEM = as_role("owner", "owner@shop.example"), as_role("system", "system@shop.example")
INVENTORY = as_role("inventory_client", "inventory@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)
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
def refused(result, status=409):
s, body = result
assert s == status, (s, body)
return body.get("detail") or body.get("reason") or json.dumps(body)
def get(path, who=ALICE, **params):
return must(api("GET", path, who, params or None))
def post(path, body=None, who=ALICE):
return must(api("POST", path, who, body or {}))
# ── builders the tests share ──────────────────────────────────────────────────────────────────────
SUPPLIER_A, SUPPLIER_B = 7001, 7002
def set_limit(email, amount, category="*", start="2026-11-20", who=OWNER, currency="GBP", sub=None):
return must(api("POST", "/approval-limits", who, {"approver_email": email, "limit_amount": str(amount), "category": category, "valid_from": start,
"currency": currency, **({"approver_sub": sub} if sub else {})}))
def seed_accounts():
with on_day("2026-01-01T09:00:00+00:00"):
for purpose, cat, acct in (("payables", "*", "2000"), ("expense", "*", "5000"), ("input_vat", "*", "1300")):
must(api("POST", "/account-map", ALICE, {"purpose": purpose, "category": cat, "account_id": acct, "valid_from": "2026-01-02"}))
def supplier_price(bp=SUPPLIER_A, product="COLA-075", unit="PK24", price="18.0000", min_quantity="1", **kw):
with on_day("2026-01-01T09:00:00+00:00"):
return must(api("POST", "/supplier-prices", BUYER, {"bp_id": bp, "product_id": product, "unit_code": unit, "unit_price": price,
"min_quantity": min_quantity, "valid_from": "2026-01-02", **kw}))
def line(product="COLA-075", quantity="10", **kw):
return {"product_id": product, "quantity": quantity, **kw}
def draft(who=BUYER, bp=SUPPLIER_A, lines=None, **kw):
return must(api("POST", "/purchase-orders", who, {"bp_id": bp, "currency": "GBP", "lines": lines if lines is not None else [line()], **kw}))
def confirmed(who=BUYER, bp=SUPPLIER_A, lines=None, **kw):
o = draft(who, bp, lines, **kw)
return must(api("POST", f"/purchase-orders/{o['purchase_order_id']}/confirm", who, {}))
def receive(order, line_no=1, quantity="10", note="DN-1", on="2026-11-19", who=RECEIVER, location="WH/STOCK", unit=None):
return api("POST", f"/purchase-orders/{order['purchase_order_id']}/receipts", who,
{"delivery_note": note, "received_on": on, "location_id": location, "lines": [{"line_no": line_no, "quantity": quantity, **({"unit_code": unit} if unit else {})}]})
def bill(order=None, who=CLERK, bp=SUPPLIER_A, number="INV-5521", date_="2026-11-19", lines=None, unit_price="18.0000", quantity="10", unit="PK24", **kw):
body = {"bp_id": bp, "supplier_invoice_no": number, "invoice_date": date_, "currency": "GBP",
"lines": lines if lines is not None else [{"order_line_no": 1, "quantity": quantity, "unit_code": unit, "unit_price": unit_price}], **kw}
if order is not None:
body["purchase_order_id"] = order["purchase_order_id"]
return api("POST", "/bills", who, body)
def supplier_bank(bp=SUPPLIER_A, iban="GB29NWBK60161331926819", sort="601613", name="Supplier Ltd"):
"""Approved bank details in force on 2026-01-02 (entered by the clerk, approved by finance after a call-back)."""
with on_day("2026-01-01T09:00:00+00:00"):
must(api("POST", f"/suppliers/{bp}/bank", CLERK, {"iban_or_account": iban, "valid_from": "2026-01-02", "bic_or_sort_code": sort, "account_name": name,
"country": "GB"}))
return must(api("POST", f"/suppliers/{bp}/bank/2026-01-02/approve", PAYAPPROVER, {"call_back_note": "rang 020 7946 0000, spoke to Jo"}))
def world(threshold="1000000", limits=True, start="2026-11-20"):
"""The usual set-up: accounts, an order threshold high enough that buyers confirm directly, a supplier with approved bank details."""
seed_accounts()
must(api("POST", "/settings/controls", OWNER, {"setting_key": "order_approval_threshold", "value": threshold, "valid_from": start}))
supplier_bank(SUPPLIER_A)
supplier_bank(SUPPLIER_B)
if limits:
set_limit("approver@shop.example", "100000.00", start=start)
set_limit("approver2@shop.example", "100000.00", start=start)
set_limit("director@shop.example", "500000.00", start=start)