dodil/erp-brick-gl

DODIL ERP brick: general ledger

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 the brick's default roles, 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 mcp import ClientSession
from mcp.client.streamable_http import streamable_http_client
from sqlalchemy import create_engine

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

HERE = pathlib.Path(__file__).resolve().parent
_ROLES = yaml.safe_load((HERE.parents[0] / "src" / "brick_gl" / "roles.yaml").read_text())["roles"]
TODAY = "2026-10-22"


@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(use_cases, "repository", repo)
    yield e
    e.dispose()


@pytest.fixture(autouse=True)
def clock():
    """Business 'today' is 2026-10-22 (09:00 UTC) in every test."""
    from erp_platform import work
    work.set_clock(at=datetime(2026, 10, 22, 9, 0, tzinfo=timezone.utc), freeze=True)
    yield
    work.reset_clock()


@pytest.fixture(autouse=True)
def neighbours(monkeypatch):
    """No neighbour is installed unless a test registers one."""
    from erp_platform import seam
    for name in ("payments", "purchasing", "ar", "fixed-assets", "business-partner", "tax"):
        seam.unregister_local(name)
    yield


def person(email, permissions, sub=None, roles=()):
    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": list(roles)}),
            "X-Dodil-User-JWT": f"h.{payload.rstrip('=')}.s"}


def as_role(role, email=None, sub=None):
    return person(email or f"{role}@firm.example", _ROLES[role], sub=sub)


BOOKKEEPER, ACCOUNTANT, CONTROLLER, AUDITOR, OWNER, SYSTEM = (as_role(r) for r in (
    "bookkeeper", "accountant", "controller", "auditor", "owner", "system"))
ACCOUNTANT2 = as_role("accountant", "accountant2@firm.example")
# the posting modules' service clients hold exactly gl:journal:post and gl:period:read, nothing else
POSTER = person("ar-client@svc.example", ["gl:journal:post", "gl:period:read"])
PAYMENTS = person("payments-client@svc.example", ["gl:journal:post", "gl:period:read", "gl:company:read", "gl:report:read"])
TAXC = person("tax-client@svc.example", ["gl:fx:read"])
CONSOLIDATION = person("cons-client@svc.example", ["gl:group_data:extract", "gl:company:read", "gl:fx:read", "gl:period:read"])
NOBODY = person("nobody@firm.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)
    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):
    """A call that has to succeed: (status, body) -> body."""
    status, body = result
    assert status == 200, (status, body)
    return body


def refused(result, status, *words):
    s, body = result
    assert s == status, (s, body)
    text = body["detail"] if isinstance(body, dict) else str(body)
    for w in words:
        assert w in text, (w, text)
    return text


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