dodil/erp-brick-projects

tests/conftest.py

"""Shared wiring: the real app with storage on a SQLite file (each test its own), the platform's public-link
table beside it, personas that carry exactly the permissions their role gives (tests/scenarios/roles.yaml),
the business clock under the test's control, and the two doors — `api(...)` and `tool(...)`."""
import base64
import json
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 public, work
from mcp import ClientSession
from mcp.client.streamable_http import streamable_http_client
from sqlalchemy import create_engine

from brick_projects.app import app
from brick_projects.application import use_cases
from brick_projects.infrastructure.tables import Repository, md

# The brick's own default roles (src/brick_projects/roles.yaml): what a customer starts from.
ROLES = yaml.safe_load((pathlib.Path(__file__).resolve().parents[1] / "src" / "brick_projects" / "roles.yaml")
                       .read_text())["roles"]
LEAD_PERMS, WORKER_PERMS = ROLES["project_lead"], ROLES["team_member"]
FINANCE_PERMS, MANAGER_PERMS, SCHEDULER_PERMS = ROLES["billing"], ROLES["project_manager"], ROLES["scheduler"]
PUBLIC_BASE = "https://status.shop.example"


@pytest.fixture(autouse=True)
def engine(tmp_path, monkeypatch):
    e = create_engine(f"sqlite:///{tmp_path / 'bucket.db'}", connect_args={"check_same_thread": False, "timeout": 30})
    md.create_all(e)
    public.md.create_all(e)
    public.configure(e.begin)
    monkeypatch.setenv("PUBLIC_LINK_BASE_URL", PUBLIC_BASE)
    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)
    clock("2026-11-02T09:00")
    yield e
    work.reset_clock()
    public.configure(None)
    e.dispose()


def clock(when: str) -> None:
    """Freeze the business clock at an ISO time (UTC)."""
    work.set_clock(at=datetime.fromisoformat(when).replace(tzinfo=timezone.utc), freeze=True)


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"}


LEAD = person("lena@agency.example", LEAD_PERMS + WORKER_PERMS, ["project_lead", "team_member"])  # may do work too
LEAD2 = person("leo@agency.example", LEAD_PERMS + WORKER_PERMS, ["project_lead", "team_member"])
WORKER = person("will@agency.example", WORKER_PERMS, ["team_member"])
FINANCE = person("fay@agency.example", FINANCE_PERMS, ["billing"])
MANAGER = person("max@agency.example", MANAGER_PERMS, ["project_manager"])
NOBODY = person("nobody@agency.example", [])
# A role NAME grants nothing: "manager" and "system" with a viewer's permissions are viewers.
NAMED_MANAGER = person("nina@agency.example", ROLES["viewer"], ["manager"])
NAMED_SYSTEM = person("sam@agency.example", ROLES["viewer"], ["system"])
SCHEDULER = person("scheduler@agency.example", SCHEDULER_PERMS, ["scheduler"])
WILL = "will@agency.example"


async def _call(method, path, headers, body=None):
    async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), 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 sale, as a contract adapter sends it ──────────────────────────────────────────────────

MIXED = [
    {"line_id": "L1", "title": "Implementation phase", "quantity": 2, "billing_kind": "one_off",
     "billing_plan": "single"},
    {"line_id": "L2", "title": "Hosting month", "quantity": 12, "billing_kind": "recurring",
     "billing_plan": "periodic"},
    {"line_id": "L3", "title": "Maintenance month", "quantity": 5, "billing_kind": "recurring",
     "billing_plan": "periodic"},
]


def plan(who=LEAD, lines=MIXED, source_id="C-100", bp_id="7311", **kw):
    body = {"source_kind": "contract", "source_id": source_id, "bp_id": bp_id, "lines": lines,
            "starts_on": "2026-11-01", "ends_on": "2026-12-31", **kw}
    return api("POST", "/plans", who, body)