"""Shared test wiring: the real app with storage on a SQLite FILE (each test its own), the platform's link tables on
the SAME engine (so the send and its links, and a signature and its link claim, share one transaction — tier 1), the
capability claims there too, no mail provider (rule 8: every delivery a manual task), business-partner replaced by an
in-memory directory of partners (tests/test_tier1.py uses the real brick), and the business clock frozen where a test
sets it. Personas carry exactly the permissions of tests/scenarios/roles.yaml.
"""
import base64
import json
import pathlib
from contextlib import contextmanager
import anyio
import httpx
import pytest
import yaml
from sqlalchemy import create_engine, event
from erp_platform import capabilities, public, seam, work
from brick_contracts.application import use_cases
from brick_contracts.infrastructure import business_partner, repositories
from brick_contracts.infrastructure.tables import ContractsRepository, md
ROLES = yaml.safe_load((pathlib.Path(__file__).parent / "scenarios" / "roles.yaml").read_text())["roles"]
PUBLIC_BASE = "https://public.contracts.test"
# Short on purpose: erp_platform.public's Link.use(conn) writes its seen/used PROJECTION on a second connection while
# the signing transaction (which holds SQLite's write lock) is still open; the projection is best-effort (the event
# row is the record) and gives up after its retries — with SQLite's default 30 s busy wait that is 90 s per signature.
SQLITE_TIMEOUT = 2.0
class Partners:
"""business-partner, as contracts sees it through its seam (GET /partners/{bp_id})."""
def __init__(self):
self.rows = {}
def add(self, bp_id, legal_name, roles, kind="organisation", work_email="", status="active"):
self.rows[int(bp_id)] = {"bp_id": int(bp_id), "bp_key": f"k{bp_id}", "legal_name": legal_name,
"kind": kind, "roles": sorted(roles), "status": status, "can_transact": True,
"country": "GB", "tax_id": "", "work_email": work_email, "domain": ""}
return int(bp_id)
def read(self, bp_id, actor=""):
p = self.rows.get(int(bp_id))
if p is None:
return business_partner.PartnerReply(False, "not_found", reason=f"no business partner {bp_id}")
return business_partner.PartnerReply(True, "found", dict(p))
@pytest.fixture
def partners(monkeypatch):
p = Partners()
monkeypatch.setattr(business_partner, "read", p.read)
return p
@pytest.fixture
def engine(tmp_path, monkeypatch, partners):
e = create_engine(f"sqlite:///{tmp_path / 'contracts.db'}", connect_args={"check_same_thread": False, "timeout": SQLITE_TIMEOUT})
@event.listens_for(e, "connect")
def _wal(conn, _rec):
conn.execute("PRAGMA journal_mode=WAL")
md.create_all(e)
public.md.create_all(e)
capabilities.tables("contracts").metadata.create_all(e)
@contextmanager
def repo():
with e.begin() as conn:
yield ContractsRepository(conn)
monkeypatch.setattr(use_cases, "repository", repo)
public.configure(e.begin)
use_cases.CT.capability_connection = e.begin
use_cases.CT.idempotency_connection = e.begin
monkeypatch.setenv("PUBLIC_LINK_BASE_URL", PUBLIC_BASE)
capabilities.reset_providers()
repositories.forget_snapshots()
yield e
public.configure(None)
use_cases.CT.capability_connection = None
use_cases.CT.idempotency_connection = None
capabilities.reset_providers()
work.reset_clock()
repositories.forget_snapshots()
e.dispose()
def claims(role: str, n: int = 1, sub: str = None) -> dict:
s = sub or f"{role}.{n}"
return {"sub": s, "email": f"{s}@scenario.test", "app_roles": [role], "permissions": list(ROLES[role])}
def headers(role: str, n: int = 1, sub: str = None) -> dict:
return seam.gateway_headers(claims(role, n, sub))
def caller(role: str, n: int = 1, sub: str = None) -> dict:
c = claims(role, n, sub)
return {"sub": c["sub"], "email": c["email"], "roles": c["app_roles"], "permissions": c["permissions"],
"connection": "local", "tenant": None, "auth_source": "pool"}
class Api:
def __init__(self, app, base="http://contracts"):
self.app, self.base = app, base
async def _go(self, method, path, who, body, params, extra, form):
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=self.app), base_url=self.base) as c:
h = dict(who or {})
h.update(extra or {})
if form is not None:
return await c.request(method, path, headers=h, data=form)
return await c.request(method, path, headers=h, json=body, params=params)
def __call__(self, method, path, who=None, body=None, params=None, headers=None, form=None, raw=False):
r = anyio.run(self._go, method, path, who, body, params, headers, form)
if raw:
return r
try:
return r.status_code, r.json()
except ValueError:
return r.status_code, r.text
@pytest.fixture
def api():
from brick_contracts.app import app
return Api(app)
@pytest.fixture
def door():
"""The composition's PUBLIC app with contracts' handlers (no pool, no MCP)."""
from brick_contracts.application.use_cases import CT
return Api(public.app(CT.public_handlers, name="contracts-public"), base=PUBLIC_BASE)
async def _tool(app, who, name, args):
import httpx2
from mcp import ClientSession
from mcp.client.streamable_http import streamable_http_client
async with app.router.lifespan_context(app):
async with httpx2.AsyncClient(transport=httpx2.ASGITransport(app=app), headers=who, base_url="http://b") as http:
async with streamable_http_client("http://b/mcp", http_client=http) as (r, w):
async with ClientSession(r, w) as s:
await s.initialize()
res = await s.call_tool(name, args)
text = res.content[0].text if res.content else ""
if res.is_error:
return "refused", text.split(f"Error executing tool {name}: ", 1)[-1]
return "ok", json.loads(text)
@pytest.fixture
def tool():
from brick_contracts.app import app
return lambda who, name, args: anyio.run(_tool, app, who, name, args)
def b64(data: bytes) -> str:
return base64.b64encode(data).decode()
ADMIN, ADMIN2 = headers("contract_admin"), headers("contract_admin", 2)
@pytest.fixture
def library(engine, api):
"""The gb pack and the seed library, approved."""
from seed import seed
work.set_clock(at=work.parse_instant("2026-10-19T10:00 Europe/London"), freeze=True)
seed(lambda p, b: api("POST", p, ADMIN, b), lambda p, b: api("POST", p, ADMIN2, b))
return api