dodil/erp-brick-consolidation

DODIL ERP brick: consolidation

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, and the two doors — `api(...)` and `tool(...)`. Generated by erp-compose scaffold."""
import base64
import json
from contextlib import contextmanager
from datetime import datetime, timezone

import anyio
import httpx
import httpx2  # the MCP client's HTTP library
import pytest
from contract_checks import CONTRACT, METHOD_DIR
from erp_platform import work
from erp_platform.contract import checked_seams
from fakes import FakeGl, FakeStore
from mcp import ClientSession
from mcp.client.streamable_http import streamable_http_client
from sqlalchemy import create_engine

from brick_consolidation.app import app
from brick_consolidation.application import use_cases
from brick_consolidation.infrastructure import seams
from brick_consolidation.infrastructure.tables import Repository, md

ALL = ["cons:carry_forward:derive", "cons:data:accept", "cons:data:collect", "cons:data:load", "cons:data:upload", "cons:eliminate:run", "cons:group:read", "cons:inventory:write", "cons:journal:approve", "cons:journal:post_auto", "cons:journal:reverse", "cons:journal:write", "cons:migration:load", "cons:migration:read", "cons:migration:reconcile", "cons:migration:sign", "cons:number_series:claim", "cons:period:open", "cons:period:reopen", "cons:period:signoff", "cons:report:read", "cons:settings:controls", "cons:settings:write", "cons:statement:materialise", "cons:structure:approve", "cons:structure:write", "cons:translate:run"]


# Every outgoing seam call (the real adapter over gl's app) is checked against gl's signed spec; the fake replaces the whole port and
# checks the same request itself (contract_checks.spec_check).
checked_seams = checked_seams(caller="consolidation", method_dir=METHOD_DIR)      # autouse


@pytest.fixture(autouse=True)
def contract():
    CONTRACT.violations.clear()
    yield CONTRACT
    assert not CONTRACT.violations, "\n".join(CONTRACT.violations)


@pytest.fixture(autouse=True)
def gl(monkeypatch):
    """gl, as its spec says it answers; tests set its companies, figures and rates."""
    fake = FakeGl()
    monkeypatch.setattr(seams.ports, "gl", fake)
    return fake


@pytest.fixture(autouse=True)
def store(monkeypatch):
    fake = FakeStore()
    monkeypatch.setattr(seams.ports, "files", fake)
    return fake


@pytest.fixture(autouse=True)
def odoo(monkeypatch):
    """No Odoo is wired unless a test installs a reader (the CON-34 loader then runs)."""
    monkeypatch.setattr(seams.ports, "odoo", seams.NoOdoo())


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


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


import pathlib  # noqa: E402

import yaml  # noqa: E402

_ROLES = yaml.safe_load((pathlib.Path(__file__).resolve().parents[1] / "src" / "brick_consolidation" / "roles.yaml").read_text())["roles"]


def as_role(role, email=None, sub=None):
    """A person holding exactly the role's permissions; `sub` apart from the email when a test needs two identities of one person."""
    h = person(email or f"{role}@group.example", _ROLES[role])
    if sub:
        user = json.loads(h["X-Dodil-User"])
        user["sub"] = sub
        h["X-Dodil-User"] = json.dumps(user)
    return h


ALICE = person("alice@shop.example", ALL)
NOBODY = person("nobody@shop.example", [])


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)