"""Shared wiring: the real app with storage on a SQLite file (each test its own), personas that carry
exactly the permissions given, and the two doors — `api(...)` and `tool(...)`. Generated by erp-compose scaffold."""
import base64
import json
import pathlib
from contextlib import contextmanager
import anyio
import httpx
import httpx2 # the MCP client's HTTP library
import pytest
import yaml
from contract_checks import METHOD_DIR
from erp_platform.contract import checked_seams
from mcp import ClientSession
from mcp.client.streamable_http import streamable_http_client
from sqlalchemy import create_engine
from brick_product.app import app
from brick_product.application import use_cases
from brick_product.infrastructure.tables import Repository, md
ALL = ["product:category:write", "product:customer_price_list:write", "product:price_list:check", "product:price_list:publish", "product:price_list:read", "product:price_list:write", "product:product:archive", "product:product:read", "product:product:suggest", "product:product:write", "product:settings:write"]
# Every outgoing seam call is checked against the CALLEE's signed spec; business-partner's lists no seams (a reported gap).
checked_seams = checked_seams(caller="product", method_dir=METHOD_DIR,
skip=("business-partner",)) # autouse
@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"}
ALICE = person("alice@shop.example", ALL)
NOBODY = person("nobody@shop.example", [])
# The brick's DEFAULT ROLES (src/brick_product/roles.yaml), as people: the permissions and no role name.
_ROLES = yaml.safe_load((pathlib.Path(__file__).resolve().parents[1] / "src" / "brick_product" / "roles.yaml").read_text())["roles"]
def as_role(role, email=None):
return person(email or f"{role}@shop.example", _ROLES[role])
CATALOGUE, PRICING, SALES, SYSTEM, OWNER = (as_role(r) for r in ("catalogue_manager", "pricing_manager", "sales_rep", "system", "owner"))
# A selling brick's service client (quote, orders, invoices, stock): the one machine permission that records a unit as used.
SELLING = person("selling@shop.example", ["product:unit:mark_used", "product:product:read", "product:price_list:read", "product:migration:read"])
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, and small builders the tests share ───────────────────────────────
@pytest.fixture(autouse=True)
def clock():
"""Business 'today' is 2026-11-20 in every test: rows dated before it are in force, after it are scheduled."""
from datetime import datetime, timezone
from erp_platform import work
work.set_clock(at=datetime(2026, 11, 20, 9, 0, tzinfo=timezone.utc), freeze=True)
yield
work.reset_clock()
def must(result):
"""A call that has to succeed: (status, body) -> body."""
status, body = result
assert status == 200, (status, body)
return body
def mk_product(pid="HOST-M", name="Managed hosting, per month", activate=True, who=None, **kw):
fields = {"base_unit_code": "MON", "category_code": None, "tax_category": "standard", "billing_kind": "recurring",
"invoicing_policy": "ordered", "product_type_code": "SERVICE", **kw} # a type is mandatory (PRD-35)
body = must(api("POST", "/products", who or CATALOGUE, {"product_id": pid, "name": name,
**{k: v for k, v in fields.items() if v is not None}}))
if activate:
body = must(api("POST", f"/products/{pid}/activate-product", who or CATALOGUE, {}))
return body
def row(pid, valid_from, price, **kw):
return {"product_id": pid, "valid_from": valid_from, "unit_price": price, **kw}
SETUP = "2026-10-01T09:00:00+00:00" # lists are built on this day: every row starts after it, however old the test's 'today'
DEFAULT = "2026-10-02" # the day after SETUP: a default_from that is valid when published on it
@contextmanager
def on_day(when):
"""Run a block with the business clock at `when` (an ISO instant), and put it back."""
from datetime import datetime
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 mk_list(list_id="LIST-GBP", currency="GBP", rows=(), rules=(), publish=True, who=None, at=None, **kw):
"""Open, fill and publish a list ON THE SETUP DAY (nothing published starts on or before the day it is published),
then return to the test's own day. Rules without a start get the day after setup."""
who = who or PRICING
with on_day(at or SETUP):
made = must(api("POST", "/price-lists", who, {"price_list_id": list_id, "currency": currency, **kw}))
if rows or rules:
dated = [{"valid_from": DEFAULT, **r} for r in rules]
made = must(api("POST", f"/price-lists/{list_id}/set-prices", who, {"rows": list(rows), "rules": dated}))
if publish:
made = must(api("POST", f"/price-lists/{list_id}/publish-price-list", who, {}))
return made
def publish(list_id, who=None, at=None):
"""Publish a draft on the setup day (or `at`): (status, body)."""
with on_day(at or SETUP):
return api("POST", f"/price-lists/{list_id}/publish-price-list", who or PRICING, {})
def price(pid, who=SALES, **q):
"""find_price through the API: (status, body)."""
return api("GET", f"/products/{pid}/price", who, {"currency": "GBP", **q})
def categories(*codes):
for code, parent in codes:
must(api("POST", f"/product-categories/{code}", CATALOGUE, {"parent_code": parent} if parent else {}))
@pytest.fixture(autouse=True)
def partner_register():
"""A stand-in partner register (business-partner's GET /partners/{bp_id}) that knows every customer except those
numbered 9_000_000 and up. A customer can only be assigned to a price list if the register has them; the tests of
that rule (tests/test_seams.py) install their own, or none."""
from erp_platform import seam
from fastapi import FastAPI, Response
app = FastAPI()
@app.get("/partners/{bp_id}")
def partner(bp_id: int, response: Response):
response.status_code = 404 if bp_id >= 9_000_000 else 200
return {"detail": "no such partner"} if bp_id >= 9_000_000 else {"bp_id": bp_id, "can_transact": True}
seam.register_local("business-partner", app)
yield
seam.unregister_local("business-partner")