dodil/erp-brick-time-off
DODIL ERP brick: time off
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 default role, the
two doors - `api(...)` and `tool(...)` - and an in-memory hr (tests/fakes.py) whose every call is checked against hr's signed spec."""
import base64
import json
import os
import pathlib
from contextlib import contextmanager
from datetime import date, datetime, timezone
import anyio
import httpx
import httpx2 # the MCP client's HTTP library
import pytest
import yaml
from erp_platform import contract as _contract
from erp_platform.contract import checked_seams
from fakes import GB_SCT, MON_WED, Hr
from mcp import ClientSession
from mcp.client.streamable_http import streamable_http_client
from sqlalchemy import create_engine
from brick_time_off.app import app
from brick_time_off.application import use_cases
from brick_time_off.infrastructure.tables import Repository, md
# Every outgoing seam call is checked against the CALLEE's signed spec (erp-method). A missing checkout is an error, never a skip.
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})")
def _names_us(entry):
named = [w.strip() for w in str(entry.get("with", "")).split(",")]
return "time-off" in named or "time-off" in " ".join(str(c) for c in entry.get("callers") or [])
_load_callee = _contract.load_callee
def _load_with_callers_read(callee, method_dir=None):
"""hr's spec lists the routes it offers every caller as `with: expenses` + `callers: [expenses EXP-19, ..., time-off TOF-19, ...]` (HR-56, HR-57,
HR-58) and contract.check reads only `with`, so a call from time-off would be reported as 'listed for [expenses], not for time-off'. Read an
entry whose `callers` names time-off as ours. (Gap for erp-platform contract.py: read `callers`, or for hr's spec: list `with: expenses,
time-off`; see the report.)"""
spec, shared = _load_callee(callee, method_dir)
return {**spec, "seams": [{**e, "with": "time-off"} if _names_us(e) else e for e in spec.get("seams") or []]}, shared
_contract.load_callee = _load_with_callers_read
checked_seams = checked_seams(caller="time-off", method_dir=METHOD_DIR) # autouse: any real Seam call
ROLES = yaml.safe_load((HERE.parent / "src" / "brick_time_off" / "roles.yaml").read_text())["roles"]
@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, 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, pid=None):
return person(email, ROLES[role], f"sub-{pid}" if pid else None)
# The people of the specification's stories: employee 4410 (London) asks; 2280 is their manager and his manager is 1904; 5120 works Monday to
# Wednesday; 4500 sits in Edinburgh; 3100 is HR (with its own person in hr, so HR can ask for leave too).
E4410, E2280, E1904, E5120, E4500, E3100 = ("emp4410@uk.example", "mgr2280@uk.example", "mgr1904@uk.example", "emp5120@uk.example",
"emp4500@uk.example", "hr3100@uk.example")
EMPLOYEE, MANAGER, MANAGER2 = as_role("employee", E4410, 4410), as_role("manager", E2280, 2280), as_role("manager", E1904, 1904)
PARTTIMER, EDINBURGH = as_role("employee", E5120, 5120), as_role("employee", E4500, 4500)
HR_ADMIN, OWNER, SYSTEM = as_role("hr_admin", E3100, 3100), as_role("owner", "owner@uk.example"), as_role("system", "tof-worker@uk.example")
HR_CLIENT = as_role("hr_client", "hr-service@uk.example")
PAYROLL_CLIENT = as_role("payroll_client", "payroll-service@uk.example")
NOBODY = person("nobody@uk.example", [])
SECOND_LOGIN_OF_2280 = person("mgr2280.alt@uk.example", ROLES["manager"], "sub-2280") # the same person under another email
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, hr, and small builders the tests share ─────────────────────────────
@pytest.fixture(autouse=True)
def clock():
"""Business 'today' is 2026-11-02 (the day 4410 opens My time off in TOF-01), 09:00 UTC = 09:00 in London."""
from erp_platform import work
work.set_clock(at=datetime(2026, 11, 2, 9, 0, tzinfo=timezone.utc), freeze=True)
yield
work.reset_clock()
@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)
@pytest.fixture(autouse=True)
def hr():
"""hr in memory: 4410 (London, Monday to Friday) managed by 2280, whose own derived manager is himself (the chief of his unit) and then 1904."""
from erp_platform import seam
n = Hr()
n.person(1904, E1904)
n.person(2280, E2280)
n.managers[2280] = [n.as_manager(2280), n.as_manager(1904)]
n.person(4410, E4410, manager=n.as_manager(2280), manager2=n.as_manager(1904))
n.person(5120, E5120, weekdays=MON_WED, fte="0.6000", manager=n.as_manager(2280), manager2=n.as_manager(1904))
n.person(4500, E4500, calendar=GB_SCT, manager=n.as_manager(2280), manager2=n.as_manager(1904))
n.person(3100, E3100, manager=n.as_manager(1904))
seam.register_local("hr", n.app)
yield n
seam.unregister_local("hr")
def must(result, status=200):
"""A call that has to succeed: (status, body) -> body."""
got, body = result
assert got == status, (got, body)
return body
def refused(result, status=409, words=""):
got, body = result
assert got == status, (got, body)
assert words in json.dumps(body), (words, body)
return body
def ask(first, last=None, who=None, **kw):
body = {"from_date": first, **({"to_date": last} if last else {}), **kw}
return api("POST", "/leave-requests", who or EMPLOYEE, body)
def request(first, last=None, who=None, **kw):
return must(ask(first, last, who, **kw), 201)
def approve(rid, who=None, **kw):
return api("POST", f"/leave-requests/{rid}/approve", who or MANAGER, kw or None)
def decided(first, last=None, who=None, by=None, **kw):
"""Ask, and have the manager approve: the request as approved."""
rid = request(first, last, who, **kw)["request_id"]
return must(approve(rid, by))
def get(rid, who=None):
return must(api("GET", f"/leave-requests/{rid}", who or EMPLOYEE))
def balance(who=None, **q):
return must(api("GET", "/balance", who or EMPLOYEE, q or None))
def entries_of(person_id=None):
from sqlalchemy import select
from brick_time_off.infrastructure import tables as t
with use_cases.repository() as repo:
q = select(t.tof_balance_entry.c.entry_ref, t.tof_balance_entry.c.kind, t.tof_balance_entry.c.quantity)
if person_id:
q = q.where(t.tof_balance_entry.c.person_id == person_id)
return sorted((r[0], r[1], str(r[2])) for r in repo.conn.execute(q))
def put_entry(person_id, ref, kind, quantity, year=2026, unit="days", on=date(2026, 1, 1), reason="test"):
from decimal import Decimal
with use_cases.repository() as repo:
repo.entries.add(ref, person_id, "ANNUAL", year, kind, Decimal(quantity), unit, on, "hr@uk.example", "sub-hr", reason, use_cases._now())
# ── the evidence store (fit notes) ─────────────────────────────────────────────────────────────
PDF = b"%PDF-1.4\n% a fit note\n"
@pytest.fixture
def store():
"""The evidence store: a dict (the S3 client is brick-local, see infrastructure/objects.py)."""
from brick_time_off.infrastructure import objects
objects.ports.store = objects.MemoryStore()
yield objects.ports.store
objects.ports.store = None