dodil/erp-brick-hr

tests/conftest.py

"""Shared test wiring: the real app with storage on a SQLite FILE (each test its own), the platform's link table on
the same engine, no capability providers (rule 8: every action a manual task), a stub drafter, 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_hr.application import use_cases
from brick_hr.infrastructure import models_adapter
from brick_hr.infrastructure.tables import HrRepository, md

ROLES = yaml.safe_load((pathlib.Path(__file__).parent / "scenarios" / "roles.yaml").read_text())["roles"]


class StubDrafter:
    def __init__(self):
        self.calls, self.fail = [], False

    def draft(self, instructions, fields, model):
        self.calls.append({"instructions": instructions, "fields": dict(fields), "model": model})
        if self.fail:
            raise models_adapter.DraftFailed("Models unreachable: stub")
        return f"Hello {fields.get('preferred_name') or ''} — draft by {model}."


@pytest.fixture
def engine(tmp_path, monkeypatch):
    e = create_engine(f"sqlite:///{tmp_path / 'hr.db'}", connect_args={"check_same_thread": False, "timeout": 30})

    @event.listens_for(e, "connect")
    def _wal(conn, _rec):
        conn.execute("PRAGMA journal_mode=WAL")
    md.create_all(e)
    public.md.create_all(e)

    @contextmanager
    def repo():
        with e.begin() as conn:
            yield HrRepository(conn)
    monkeypatch.setattr(use_cases, "repository", repo)
    public.configure(e.begin)
    monkeypatch.setenv("PUBLIC_LINK_BASE_URL", "https://public.hr.test")
    capabilities.reset_providers()
    seam.unregister_local("business-partner")
    yield e
    public.configure(None)
    capabilities.reset_providers()
    work.reset_clock()
    e.dispose()


@pytest.fixture
def drafter(monkeypatch):
    d = StubDrafter()
    monkeypatch.setattr(models_adapter, "DRAFTER", d)
    return d


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:
    """The identity a use case receives, for calling it directly (rules-level tests)."""
    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):
        self.app = app

    async def _go(self, method, path, who, body, params, extra):
        async with httpx.AsyncClient(transport=httpx.ASGITransport(app=self.app), base_url="http://hr") as c:
            h = dict(who or {})
            h.update(extra or {})
            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):
        r = anyio.run(self._go, method, path, who, body, params, headers)
        try:
            return r.status_code, r.json()
        except ValueError:
            return r.status_code, r.text


@pytest.fixture
def api():
    from brick_hr.app import app
    return Api(app)


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)
                    tools = None
    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_hr.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()