dodil/erp-brick-manufacturing
DODIL ERP brick: manufacturing
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, the two doors - `api(...)` and `tool(...)` - and in-memory neighbours (tests/fakes.py). Every outgoing seam call is checked
against the CALLEE's signed spec (erp-method): a missing checkout is an error, never a skip."""
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
from erp_platform.seam import Seam
from fakes import Neighbours
from mcp import ClientSession
from mcp.client.streamable_http import streamable_http_client
from sqlalchemy import create_engine
from brick_manufacturing.app import app
from brick_manufacturing.application import use_cases
from brick_manufacturing.infrastructure.tables import Repository, md
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})")
_CONTRACT = Recorder()
# SPEC GAPS (reported, not hidden): the callee's signed spec lists some of manufacturing's calls under another caller or not at all.
# product GET /products/{product_id} and /convert: product v6 has no `with: manufacturing` entry for them -> not checked.
# inventory GET /availability/{product_id} and /valuation/products/{product_id}: listed `with: sales-order` only -> checked AS sales-order
# (the same route, query and permission class). tests/test_seams.py fails the day a spec lists them, so this goes.
# Every OTHER call (the issue, receipt, valuation-adjustment and journal doors) is checked as manufacturing against the callee's spec.
AS_OTHER_CALLER = {("inventory", "GET"): "sales-order", ("gl", "GET"): "inventory"}
# inventory GET /lots/{product_id}/{lot_code}/trace: listed only as ONE row with GET /serials/... ("A and B"), `with: crm` -> the checker
# finds no seam for it -> not checked (tests/test_seams.py fails the day inventory lists it for manufacturing).
UNCHECKED = ("product", "ai-fake") # ai-fake: a capability provider the tests wire, not a module with a spec
UNCHECKED_PATHS = ("/lots/",)
@pytest.fixture(autouse=True)
def checked_seams(monkeypatch):
from erp_platform.contract import check
original = Seam._call
def call(self, method, path, actor, caller=None, **kw):
if self.name not in UNCHECKED and self.installed and not path.startswith(UNCHECKED_PATHS):
who = AS_OTHER_CALLER.get((self.name, method), "manufacturing")
bad = check(self.name, method, path, kw.get("json") if method == "POST" else kw.get("params"), caller=who,
method_dir=METHOD_DIR)
if bad:
msg = f"manufacturing -> {self.name} {method} {path} breaks the callee's spec:\n " + "\n ".join(bad)
_CONTRACT.violations.append(msg)
raise AssertionError(msg)
return original(self, method, path, actor, caller, **kw)
monkeypatch.setattr(Seam, "_call", call)
@pytest.fixture(autouse=True)
def contract():
_CONTRACT.violations.clear()
yield _CONTRACT
assert not _CONTRACT.violations, "\n".join(_CONTRACT.violations)
ROLES = yaml.safe_load((HERE.parent / "src" / "brick_manufacturing" / "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)
@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=(), 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):
return person(email or f"{role}@roastery.example", ROLES[role], sub=sub)
ENGINEER, PLANNER, OPERATOR, ACCOUNTANT, QUALITY, OWNER, SYSTEM = (
as_role(r) for r in ("process_engineer", "planner", "operator", "cost_accountant", "quality", "owner", "system"))
ACCOUNTANT2 = as_role("cost_accountant", "second.accountant@roastery.example")
NOBODY = person("nobody@roastery.example", [])
BOSS = person("boss@roastery.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-10 in every test (a BOM must be released for a LATER day)."""
from erp_platform import work
work.set_clock(at=datetime(2026, 11, 10, 9, 0, tzinfo=timezone.utc), freeze=True)
yield
work.reset_clock()
@pytest.fixture(autouse=True)
def nb():
"""Product, inventory and the ledger, in memory."""
from erp_platform import seam
n = Neighbours()
names = {"product": "product", "inventory": "inventory", "gl": "gl", "sales-order": "sales"}
for name, attr in names.items():
seam.register_local(name, getattr(n, attr))
yield n
for name in names:
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 set_day(when):
from erp_platform import work
work.set_clock(at=datetime.fromisoformat(when), freeze=True)
D16 = "2026-11-16T09:00:00+00:00"
def must(result, status=200):
got, body = result
assert got == status, (got, body)
return body
def refused(result, status=409, words="", gate=None):
got, body = result
assert got == status, (got, body)
assert words in json.dumps(body), (words, body)
if gate:
assert body.get("gate") == gate, body
return body
def line(product, qty, unit="KGM", **kw):
return {"product_id": product, "quantity": qty, "unit_code": unit, **kw}
def roast_bom(version=1, valid_from="2026-11-16", output="84.000", component="100.000", chaff="1.200", release=True, who=None):
"""BOM ROAST-ETH usage production: 100.000 KGM green beans -> 84.000 KGM roasted + 1.200 KGM chaff (MFG-02)."""
lines = [line("GREEN-ETH", component)]
if chaff:
lines.append(line("CHAFF", chaff, kind="by_product", nrv_unit_cost="0.0000"))
body = {"version": version, "valid_from": valid_from, "output_quantity": output, "output_unit_code": "KGM", "lines": lines,
"company_code": "UK01"}
out = must(api("POST", "/boms/ROAST-ETH:production/versions", who or ENGINEER, body))
if release:
out = must(api("POST", f"/boms/ROAST-ETH:production/versions/{version}/release", ENGINEER))
return out
def new_order(qty="84.000", due="2026-11-16", who=None, **kw):
return must(api("POST", "/production-orders", who or PLANNER,
{"company_code": "UK01", "product_id": "ROAST-ETH", "warehouse_id": "ROASTERY", "planned_quantity": qty,
"due_on": due, **kw}))["production_order_id"]
def release(oid, who=None):
with on_day(D16):
return api("POST", f"/production-orders/{oid}/release", who or PLANNER)
def released_order(qty="84.000", chaff="1.200"):
"""A released ROAST-ETH order (the clock is back on 2026-11-10 afterwards)."""
roast_bom(chaff=chaff)
oid = new_order(qty)
must(release(oid))
return oid
def issue(oid, qty="100.000", lot="G-2611-ETH-07", product="GREEN-ETH", who=None, **kw):
with on_day(D16):
return api("POST", f"/production-orders/{oid}/issue", who or OPERATOR,
{"location_id": "ROASTERY/RAW", "lines": [{"product_id": product, "quantity": qty, "lot_code": lot, **kw}]})
def output(oid, qty="82.500", lot="R-20261116-01", chaff="1.100", who=None, **kw):
body = {"location_id": "ROASTERY/FG", "quantity": qty, "lot_code": lot, **kw}
if chaff:
body["by_products"] = [{"product_id": "CHAFF", "quantity": chaff}]
with on_day(D16):
return api("POST", f"/production-orders/{oid}/output", who or OPERATOR, body)
def complete(oid, who=None):
with on_day(D16):
return api("POST", f"/production-orders/{oid}/complete", who or ACCOUNTANT)
def close(oid, who=None):
with on_day("2026-11-18T09:00:00+00:00"):
return api("POST", f"/production-orders/{oid}/close", who or ACCOUNTANT)
def get_order(oid, who=None):
return must(api("GET", f"/production-orders/{oid}", who or ACCOUNTANT))