dodil/erp-brick-inventory

DODIL ERP brick: inventory

tests/conftest.py

"""Shared wiring: the real app with storage on a SQLite file (each test its own), personas that carry exactly the permissions
given, the two doors - `api(...)` and `tool(...)` - and the neighbours as their spec routes answer (tests/fakes.py)."""
import base64
import json
import pathlib
from contextlib import contextmanager
from datetime import datetime, timezone

import anyio
import fakes
import httpx
import httpx2
import pytest
import yaml
from mcp import ClientSession
from mcp.client.streamable_http import streamable_http_client
from sqlalchemy import create_engine

from brick_inventory.app import app
from brick_inventory.application import unit as unit_mod
from brick_inventory.infrastructure import business_time
from brick_inventory.infrastructure.tables import Repository, md

_ROLES = yaml.safe_load((pathlib.Path(__file__).resolve().parents[1] / "src" / "brick_inventory" / "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, "timeout": 30})
    md.create_all(e)

    @contextmanager
    def repo():
        with e.begin() as conn:
            yield Repository(conn)
    monkeypatch.setattr(unit_mod, "repository", repo)
    from brick_inventory.application import use_cases
    monkeypatch.setattr(use_cases.INV, "idempotency_connection", e.begin)
    yield e
    e.dispose()


@pytest.fixture(autouse=True)
def clock():
    """Business 'today' is 2026-11-20 in every test (Europe/London)."""
    from erp_platform import work
    work.set_clock(at=datetime(2026, 11, 20, 9, 0, tzinfo=timezone.utc), freeze=True)
    business_time.SOURCE = business_time.EnvSource()
    yield
    work.reset_clock()


@pytest.fixture(autouse=True)
def neighbours():
    fakes.reset()
    fakes.install()
    yield
    fakes.reset()


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)


MANAGER, OPERATOR, CONTROLLER, OWNER, ACCOUNTANT, SALES, SYSTEM = (
    as_role(r) for r in ("warehouse_manager", "stock_operator", "stock_controller", "owner", "accountant", "sales_rep", "system"))
NOBODY = person("nobody@shop.example", [])
BOSS = person("boss@shop.example", ALL)


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)
    try:
        return r.status_code, r.json()
    except ValueError:
        return r.status_code, r.text


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)


def must(result):
    status, body = result
    assert status == 200, (status, body)
    return body


def refused(result, status, words=""):
    got, body = result
    assert got == status, (got, body)
    text = json.dumps(body) if not isinstance(body, str) else body
    assert words in text, (words, body)
    return body


@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)