dodil/erp-brick-pos
DODIL ERP brick: point of sale
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 the default roles (src/brick_pos/roles.yaml), 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)."""
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 import contract
from erp_platform.contract import Recorder
from erp_platform.seam import Seam
from fakes import Neighbours, seed_uk
from mcp import ClientSession
from mcp.client.streamable_http import streamable_http_client
from sqlalchemy import create_engine
from brick_pos.app import app
from brick_pos.application import use_cases
from brick_pos.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})")
# Two defects in the CALLEES' specs that make the contract check refuse calls the spec describes. Each is allowed by name and
# tests/test_seams.py checks it is STILL a defect, so the day the spec is fixed that test fails and the allowance goes
# (the unquoted-`on` defect, gap 2, is fixed in the spec: its allowance is gone):
# 1. product's spec lists GET /products/{product_id}?on= (PRD-28) and GET /products/{product_id}/convert (PRD-21) for inventory,
# purchasing and others, not for pos (both routes are allowed whole);
# 3. tax's v6 entries for pos (TAX-52 GET /rates, TAX-53 POST /documents/compute) sit under `external_services:` in tax's
# module.yaml, not under `seams:`, so contract.check cannot find them: they are checked here against that entry, exactly.
# 2. `{ field: on, ... }` in product PRD-40 and tax TAX-52 is unquoted YAML 1.1, which reads `on` as the boolean True, so a
# correct `on=` query parameter is reported "not in the spec's request_schema" (the two messages it causes are dropped).
_CONTRACT = Recorder()
# Routes pos reads that the OWNER lists for other callers and not for pos (the spec gaps recorded in AGENTS.md): each is checked against the entry
# of a caller the owner DOES list, so pos's request is still held to the signed schema. tests/test_seams.py fails the day pos is listed.
LENDERS = {("product", r"/products/[^/]+"): "inventory", ("product", r"/products/[^/]+/convert"): "sales-order",
("gl", r"/companies/[^/]+"): "ar", ("inventory", r"/warehouses/[^/]+"): "product", ("inventory", r"/locations"): "purchasing"}
def _lender(callee: str, method: str, path: str):
import re
return next((who for (c, pat), who in LENDERS.items() if c == callee and method == "GET" and re.fullmatch(pat, path)), None)
def _violations(callee: str, method: str, path: str, body, caller: str = "pos") -> list:
bad = contract.check(callee, method, path, body, caller=caller, method_dir=METHOD_DIR)
if bad and callee == "tax" and bad[0].startswith("tax: no seam"):
spec, shared = contract.load_callee("tax", METHOD_DIR)
hits = [e for e in spec.get("external_services") or [] if e.get("with") == "pos" and e.get("request_schema")
and e["route"].split()[0] == method and contract._route_regex(e["route"].split(None, 1)[1]).match(path)]
if hits:
bad = contract._schema_for(hits[0], method, path, body, shared)
return list(bad)
@pytest.fixture(autouse=True)
def contract_checked(monkeypatch):
"""Every Seam call goes through contract.check against the callee's module.yaml; a violation fails the call and, if a
brick swallowed it, the test at teardown."""
_CONTRACT.violations.clear()
original = Seam._call
def call(self, method, path, actor, caller=None, **kw):
bare = path.split("?", 1)[0]
if self.installed and contract._has_spec(self.name, METHOD_DIR):
body = kw.get("json") if method in ("POST", "PUT", "PATCH") else kw.get("params")
lender = _lender(self.name, method, bare)
bad = _violations(self.name, method, bare, body, lender or "pos")
if bad:
msg = f"pos -> {self.name} {method} {bare} breaks the callee's spec:\n " + "\n ".join(bad)
_CONTRACT.violations.append(msg)
raise contract.ContractViolation(msg)
return original(self, method, path, actor, caller, **kw)
monkeypatch.setattr(Seam, "_call", call)
yield _CONTRACT
assert not _CONTRACT.violations, "\n".join(_CONTRACT.violations)
ROLES = yaml.safe_load((HERE.parent / "src" / "brick_pos" / "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, 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": []}),
"X-Dodil-User-JWT": f"h.{payload.rstrip('=')}.s"}
def as_role(role, email=None, sub=None):
return person(email or f"{role}@shop.example", ROLES[role], sub)
CASHIER, CASHIER2, LEAD, LEAD2, OPS, ACCOUNTANT, OWNER, SYSTEM = (
as_role("cashier"), as_role("cashier", "cashier2@shop.example"), as_role("shift_lead"),
as_role("shift_lead", "lead2@shop.example"), as_role("retail_ops"), as_role("accountant"), as_role("owner"),
as_role("system"))
NOBODY = person("nobody@shop.example", [])
BOSS = person("boss@shop.example", ["pos:*:*"]) # a wildcard: it never reaches a machine permission
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 ─────────────────
def freeze(iso: str):
from erp_platform import work
work.set_clock(at=datetime.fromisoformat(iso), freeze=True)
@pytest.fixture(autouse=True)
def clock():
"""The business 'now' is 2026-11-14 09:00 UTC (London is on GMT: the business date is the 14th) in every test."""
from erp_platform import work
work.set_clock(at=datetime(2026, 11, 14, 9, 0, tzinfo=timezone.utc), freeze=True)
yield
work.reset_clock()
CURRENT = [None] # the neighbours of the running test (make_shop seeds the shop's company and warehouse in them)
@pytest.fixture(autouse=True)
def nb():
"""Product, tax, inventory, gl and payments, in memory, with the London shop's world seeded."""
from erp_platform import seam
n = Neighbours()
seed_uk(n)
CURRENT[0] = n
seam.register_local("product", n.product)
seam.register_local("tax", n.tax)
seam.register_local("inventory", n.inventory)
seam.register_local("manufacturing", n.manufacturing)
seam.register_local("gl", n.gl)
seam.register_local("payments", n.payments)
yield n
CURRENT[0] = None
for name in ("product", "tax", "inventory", "manufacturing", "gl", "payments"):
seam.unregister_local(name)
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, 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 make_shop(code="LON-SOHO", *, registers=("R1", "R2"), price_list="LIST-UK-SHOP", country="GB", fiscal="none",
staff=(CASHIER, CASHIER2, LEAD, LEAD2), company=None):
"""The shop, with the world POS-01 says it needs (given): gl's company and inventory's warehouse with its bar location."""
company = company or ("DE01" if country == "DE" else "UK01")
if CURRENT[0] is not None:
CURRENT[0].companies.setdefault(company, {"country": country, "functional_currency": "GBP"})
CURRENT[0].warehouses.setdefault(code, {"company_code": company, "status": "active", "locations": [(f"{code}/BAR", "internal")]})
body = {"shop_id": code, "name": code, "company_code": company, "warehouse_id": code, "consumption_location_id": f"{code}/BAR",
"price_list_id": price_list, "fiscal_profile": fiscal, "country": country}
must(api("POST", "/shops", OPS, body))
for r in registers:
must(api("POST", f"/shops/{code}/registers", OPS, {"register_id": r}))
for who in staff:
ident = json.loads(who["X-Dodil-User"])
must(api("POST", f"/shops/{code}/staff", OPS, {"sub": ident["sub"], "email": ident["email"]}))
return code
def open_till(shop="LON-SOHO", register="R1", opening_float="150.00", who=None):
return must(api("POST", "/sessions", who or CASHIER, {"shop_id": shop, "register_id": register, "opening_float": opening_float}))
def ring(session_id, items, eat_in=None, who=None):
"""items: [(product_id, quantity)]"""
body = {"session_id": session_id, "lines": [{"product_id": p, "quantity": str(q)} for p, q in items]}
if eat_in is not None:
body["eat_in"] = eat_in
return must(api("POST", "/receipts", who or CASHIER, body))
def pay(receipt, method="cash", amount=None, tip=None, who=None, **more):
body = {"method": method, "amount": amount or receipt["remaining"], "version": receipt["version"], **more}
if tip is not None:
body["tip_amount"] = tip
return must(api("POST", f"/receipts/{receipt['receipt_id']}/payments", who or CASHIER, body))
def finalise(receipt, who=None):
return must(api("POST", f"/receipts/{receipt['receipt_id']}/finalise", who or CASHIER, {"version": receipt["version"]}))
def sell(session_id, items, *, method="cash", tip=None, eat_in=None, who=None, **more):
"""Ring up, take one tender for the whole bill, finalise: one till sale."""
r = ring(session_id, items, eat_in, who)
r = pay(r, method, tip=tip, who=who, **({"terminal_txn_ref": f"T-{r['receipt_id']}-1", "terminal_status": "authorised"}
if method == "card" else {}), **more)
return finalise(r, who)