dodil/erp-brick-timesheets
DODIL ERP brick: timesheets
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 in-memory neighbours (tests/fakes.py) whose every call is checked against the
CALLEE's signed spec."""
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 as _contract
from erp_platform.contract import checked_seams
from fakes import Neighbours
from mcp import ClientSession
from mcp.client.streamable_http import streamable_http_client
from sqlalchemy import create_engine
from brick_timesheets.app import app
from brick_timesheets.application import use_cases
from brick_timesheets.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})")
# Routes timesheets' own spec (TS-21) says it calls, whose owner lists other callers only: a gap in gl v8's seams (GL-61 names inventory and
# purchasing, GL-32 names ar, fixed-assets, inventory and purchasing); and product v6's PRD-31 names sales-order alone. The request shape is the owner's own entry's.
UNLISTED = ("GET /cost-centres/{cost_centre}", "GET /periods/open", "GET /products/{product_id}/price")
def _names_us(entry):
"""hr's HR-62 lists its callers in `callers: [time-off TOF-19, timesheets TS-18, ...]` under `with: time-off`, and time-off's TOF-20 (`with: payroll`)
names timesheets only in its status; contract.check reads only `with`."""
withs = [w.strip() for w in str(entry.get("with", "")).split(",")]
return ("timesheets" in withs or any(str(c).startswith("timesheets ") for c in entry.get("callers") or [])
or "timesheets v1 TS-" in str(entry.get("status", "")) or entry.get("route", "").split("?")[0] in UNLISTED) # time-off TOF-20 says `with: payroll` and names timesheets only in its status
_load_callee = _contract.load_callee
def _load_with_callers_read(callee, method_dir=None):
"""Read an entry that names timesheets in `callers` as ours. (Gap for erp-platform contract.py: `_callers` should read `callers:` too.)"""
spec, shared = _load_callee(callee, method_dir)
return {**spec, "seams": [{**e, "with": "timesheets"} 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="timesheets", method_dir=METHOD_DIR) # autouse: any real Seam call
ROLES = yaml.safe_load((HERE.parent / "src" / "brick_timesheets" / "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=None, sub=None):
return person(email or f"{role}@uk.example", ROLES[role], sub)
# The people of the specification's stories: employee 4410 (SENIOR) records time, 4425 is a colleague; 2280 is their manager, 1904 the
# manager's manager; 3020 leads the projects.
EMP, EMP2, MGR, MGR2, LEAD = ("emp4410@uk.example", "emp4425@uk.example", "mgr2280@uk.example", "mgr1904@uk.example", "lead3020@uk.example")
EMPLOYEE, EMPLOYEE2 = as_role("employee", EMP), as_role("employee", EMP2)
MANAGER, MANAGER2, LEADER = as_role("manager", MGR), as_role("manager", MGR2), as_role("project_lead", LEAD)
FINANCE, FINANCE2 = as_role("finance", "fin1@uk.example"), as_role("finance", "fin2@uk.example")
OWNER, SYSTEM = as_role("owner"), as_role("system")
PAYROLL, TIMEOFF = as_role("payroll_client", "payroll-service@uk.example"), as_role("timeoff_client", "timeoff-service@uk.example")
NOBODY = person("nobody@uk.example", [])
WEEK, SHEET = "2026-W47", "4410:2026-W47" # Monday 2026-11-16 .. Sunday 2026-11-22
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 ─────────────────
@pytest.fixture(autouse=True)
def clock():
"""Business 'today' is Friday 2026-11-20, 15:00 UTC = 15:00 in London: the day 4410 submits week W47 in TS-03."""
from erp_platform import work
work.set_clock(at=datetime(2026, 11, 20, 15, 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)
APPROVAL_DAY = "2026-11-23T10:00:00+00:00" # the Monday after: the day TS-04 / TS-11 approve
@pytest.fixture(autouse=True)
def nb():
"""hr, projects, time-off, the ledger and sales-order, in memory: employee 4410 (UK01, home CONS-LON, SENIOR from 2025-04-01) managed by
2280, whose manager is 1904; 4425 beside them; 3020 leads P1 and P2."""
from erp_platform import seam
n = Neighbours()
m2280, m1904 = n.as_manager(2280, MGR), n.as_manager(1904, MGR2)
n.employee(4410, EMP, manager=m2280, manager2=m1904)
n.employee(4425, EMP2, manager=m2280, manager2=m1904)
n.employee(3020, LEAD, manager=m2280, manager2=m1904)
n.employee(2280, MGR, manager=m1904)
n.employee(1904, MGR2)
n.standard_projects()
n.absence(4410, "UK01-TOF-2026-00405", "ANNUAL", ["2026-11-20"]) # TS-01: annual leave on Friday 2026-11-20
seam.register_local("hr", n.hr)
seam.register_local("projects", n.projects_app)
seam.register_local("time-off", n.time_off)
seam.register_local("gl", n.gl)
seam.register_local("sales-order", n.orders)
seam.register_local("product", n.product)
seam.register_local("ar", n.ar)
n.prices[("CONSULT-SR", 7311)] = {"unit_price": "110.0000"}
yield n
for name in ("hr", "projects", "time-off", "gl", "sales-order", "product", "ar"):
seam.unregister_local(name)
@pytest.fixture(autouse=True)
def rates(engine):
"""Finance's rows: 4410 and 3020 are SENIOR from 2025-04-01 (52.0000 from 2026-01-01, the seeded placeholder), 4425 JUNIOR."""
from datetime import date
with use_cases.repository() as repo:
for pid, group in ((4410, "SENIOR"), (3020, "SENIOR"), (4425, "JUNIOR")):
repo.rates.put_person(pid, group, date(2025, 4, 1), "setup", use_cases._now())
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 open_week(who=None, week=WEEK):
return must(api("POST", "/sheets", who or EMPLOYEE, {"week": week}), 201)
def line(sheet_id, who=None, **kw):
body = {"activity_code": "DEV", "project_id": "pj:sales_order:quote:Q-2026-0042@v2", **kw}
return must(api("POST", f"/sheets/{sheet_id}/lines", who or EMPLOYEE, body), 201)
def hours(sheet_id, line_no, entries, who=None):
"""entries: {"2026-11-16": "7.50", ...} (a value may be (hours, note))."""
es = [{"work_date": d, "hours": h[0], "note": h[1]} if isinstance(h, tuple) else {"work_date": d, "hours": h} for d, h in entries.items()]
return must(api("POST", f"/sheets/{sheet_id}/lines/{line_no}/entries", who or EMPLOYEE, {"entries": es}))
P1, P2 = "pj:sales_order:quote:Q-2026-0042@v2", "pj:contract:CT-2026-0009"
def story_week(sheet_id=SHEET, who=None):
"""The week of TS-02, on top of the Friday leave: P1 DEV Mon 7.50 Tue 7.50 Wed 4.00 (19.00), ADMIN 3.50 Wed on CONS-LON, P2 MEETING Thu 7.50."""
who = who or EMPLOYEE
a = line(sheet_id, who, project_id=P1, activity_code="DEV")["lines"][-1]["line_no"]
b = must(api("POST", f"/sheets/{sheet_id}/lines", who, {"activity_code": "ADMIN", "cost_centre": "CONS-LON"}), 201)["lines"][-1]["line_no"]
c = line(sheet_id, who, project_id=P2, activity_code="MEETING")["lines"][-1]["line_no"]
hours(sheet_id, a, {"2026-11-16": "7.50", "2026-11-17": "7.50", "2026-11-18": "4.00"}, who)
hours(sheet_id, b, {"2026-11-18": "3.50"}, who)
hours(sheet_id, c, {"2026-11-19": "7.50"}, who)
return a, b, c
def submit(sheet_id=SHEET, who=None):
return must(api("POST", f"/sheets/{sheet_id}/submit", who or EMPLOYEE))
def approve(sheet_id=SHEET, who=None, at=APPROVAL_DAY):
with on_day(at):
return api("POST", f"/sheets/{sheet_id}/approve", who or MANAGER)
def sheet(sheet_id=SHEET, who=None, revision=None):
return must(api("GET", f"/sheets/{sheet_id}", who or EMPLOYEE, {"revision": revision} if revision else None))
def approved_week():
"""The story week opened, filled, submitted and approved on 2026-11-23: the starting point of TS-04/07/11/12."""
open_week()
story_week()
submit()
return must(approve())