dodil/erp-brick-sales-order
DODIL ERP brick: sales order
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, and the two doors — `api(...)` and `tool(...)`."""
import base64
import json
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 contract_checks import CONTRACT, METHOD_DIR
from erp_platform import idempotency, join_after_response, work
from erp_platform.contract import checked_seams
from fakes import (
FakeGl,
FakeInventory,
FakeInvoicing,
FakeMaps,
FakeOdoo,
FakePartners,
FakePayments,
FakeProduct,
FakePurchasing,
FakeTax,
)
from mcp import ClientSession
from mcp.client.streamable_http import streamable_http_client
from sqlalchemy import create_engine
from brick_sales_order.app import app
from brick_sales_order.application import use_cases
from brick_sales_order.infrastructure import seams
from brick_sales_order.infrastructure.tables import Repository, md
ALL = ["so:number_series:claim", "so:order:approve", "so:order:cancel", "so:order:confirm", "so:order:deliver", "so:order:finish",
"so:order:import", "so:order:intake", "so:order:invoice", "so:order:read", "so:order:suggest", "so:order:write",
"so:sales_order:read_all", "so:settings:controls", "so:settings:write", "so:pricing:write", "so:rebate:write", "so:rebate:approve", "so:rebate:accrue",
"so:memo:approve", "so:order:block"]
# Every outgoing seam call (the real adapters over neighbour apps) is checked against the CALLEE's signed spec. business-partner's
# signed spec lists no seams at all, so there is no entry to check GET /partners/{bp_id} against: a spec gap, reported, not exempted
# for any other callee.
checked_seams = checked_seams(caller="sales-order",
method_dir=METHOD_DIR, skip=("business-partner",)) # autouse
@pytest.fixture(autouse=True)
def contract():
"""The fakes replace whole ports (they never reach a Seam), so each checks its call against the callee's spec."""
CONTRACT.violations.clear()
yield CONTRACT
assert not CONTRACT.violations, "\n".join(CONTRACT.violations)
@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(use_cases.SO, "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(use_cases, "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.payments, N.invoicing, N.partners."""
class N:
product, tax, payments, invoicing, partners = FakeProduct(), FakeTax(), FakePayments(), FakeInvoicing(), FakePartners()
maps, odoo, inventory, gl, purchasing = FakeMaps(), FakeOdoo(), FakeInventory(), FakeGl(), FakePurchasing()
for name in ("product", "tax", "payments", "invoicing", "partners", "maps", "odoo", "gl", "purchasing"):
monkeypatch.setattr(seams.ports, name, getattr(N, name))
return N
@pytest.fixture
def inventory(monkeypatch, neighbours):
"""Inventory is part of the installation (without this fixture it is not, and every line is delivered by hand)."""
fake = FakeInventory()
monkeypatch.setattr(seams.ports, "inventory", fake)
return fake
@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=()):
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"}
ALICE = person("alice@shop.example", ALL)
NOBODY = person("nobody@shop.example", [])
_ROLES = yaml.safe_load((pathlib.Path(__file__).resolve().parents[1] / "src" / "brick_sales_order" / "roles.yaml").read_text())["roles"]
def as_role(role, email=None):
return person(email or f"{role}@shop.example", _ROLES[role])
REP, REP2, ADMIN, OWNER, FULFILLER, BILLER, SYSTEM = (as_role("sales_rep"), as_role("sales_rep", "rep2@shop.example"), as_role("order_admin"),
as_role("owner"), as_role("fulfiller"), as_role("biller"), as_role("system"))
ADMIN2 = as_role("order_admin", "admin2@shop.example")
INVENTORY = {**person("inventory@shop.example", ["so:order:deliver", "so:order:read", "so:return:inspect"]), "X-Dodil-Auth-Source": "platform"} # inventory's service client
PURCHASING = {**person("purchasing@shop.example", ["so:order:deliver", "so:order:read"]), "X-Dodil-Auth-Source": "platform"} # purchasing's service client
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)
join_after_response() # work run after the response (the cancel's re-promise) has finished before the next call
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)
ADDRESS = {"lines": ["1 High Street"], "city": "London", "postcode": "N1 1AA", "country": "GB"}
def hand(who=REP, bp_id=7311, lines=None, **kw):
"""A draft by hand: 2 x PK24 of COLA-075 and 3 hours of SETUP unless told otherwise."""
lines = lines or [{"product_id": "COLA-075", "quantity": "2", "unit_code": "PK24"}, {"product_id": "SETUP", "quantity": "3", "unit_code": "HUR"}]
if any(x.get("usage") == "drop_ship" for x in lines): # v8: a drop-ship line sends the delivery address object to purchasing
kw.setdefault("ship_to_address", ADDRESS)
return must(api("POST", "/sales-orders", who, {"bp_id": bp_id, "currency": "GBP", "lines": lines, **kw}))
def confirm(order, who=REP):
return must(api("POST", f"/sales-orders/{order['sales_order_id']}/confirm", who, {}))
def confirmed(who=REP, **kw):
return confirm(hand(who, **kw), who)
def deliver(order, line_no, qty, ref, who=FULFILLER, **kw):
return api("POST", f"/sales-orders/{order['sales_order_id']}/deliveries", who, {"line_no": line_no, "quantity": str(qty), "delivery_ref": ref, **kw})
def get(order_id, who=ADMIN):
return must(api("GET", f"/sales-orders/{order_id}", who))