dodil/erp-platform

templates/brick-example/tests/test_doors.py

"""The use cases through both doors, with an in-memory repository instead of a bucket.

What runs is the real app — routes, /mcp, permissions, row visibility — with only storage swapped.
The repository itself is tested against DataK3 in erp-platform (scripts/live_repository_check.py).
"""
import base64
import copy
import json
from contextlib import contextmanager

import anyio
import httpx
import httpx2                     # the MCP client's HTTP library
import pytest
from mcp import ClientSession
from mcp.client.streamable_http import streamable_http_client

from brick_example.app import app
from brick_example.application import use_cases


class InMemoryQuotes:
    def __init__(self, store: dict):
        self.store = store

    def get(self, quote_id):
        return copy.deepcopy(self.store.get(quote_id))      # a load is a copy, like a database read

    def save(self, q):
        self.store[q.quote_id] = copy.deepcopy(q)

    def ids_owned_by(self, owner):
        return sorted(k for k, q in self.store.items() if owner is None or q.owner == owner)


@pytest.fixture(autouse=True)
def memory(monkeypatch):
    store: dict = {}

    @contextmanager
    def repo():
        yield InMemoryQuotes(store)
    monkeypatch.setattr(use_cases, "repository", repo)
    return store


def person(email, roles, permissions):
    payload = base64.urlsafe_b64encode(json.dumps({"permissions": permissions}).encode()).decode()
    return {"X-Dodil-User": json.dumps({"sub": email, "email": email, "app_roles": roles}),
            "X-Dodil-User-JWT": f"h.{payload.rstrip('=')}.s"}


ALL = ["example:quote:write", "example:quote:send", "example:quote:accept"]
ALICE = person("alice@shop.example", ["sales"], ALL)
BOB = person("bob@shop.example", ["sales"], 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:
        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 _tool(headers, name, args):
    # The MCP app's lifespan has to run for its session manager; drive the real app through it.
    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()
                    res = await 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 tool(headers, name, args):
    return anyio.run(_tool, headers, name, args)


def test_a_quote_from_draft_to_accepted_through_the_api():
    assert api("POST", "/quotes", ALICE, {"quote_id": "Q-1", "customer": "Acme"})[0] == 200
    status, q = api("POST", "/quotes/Q-1/lines", ALICE,
                    {"description": "Banner", "quantity": 3, "unit_price": "33.33"})
    assert status == 200 and q["total"] == "99.99"
    assert api("POST", "/quotes/Q-1/send", ALICE)[1]["status"] == "sent"
    assert api("POST", "/quotes/Q-1/accept", ALICE)[1]["status"] == "accepted"


def test_an_agent_does_the_same_through_mcp():
    assert tool(ALICE, "create_quote", {"quote_id": "Q-2", "customer": "Acme"})[0] == "ok"
    verdict, q = tool(ALICE, "add_line", {"quote_id": "Q-2", "description": "Banner",
                                          "quantity": 2, "unit_price": "10.50"})
    assert verdict == "ok" and q["total"] == "21.00"


def test_a_broken_rule_says_why_through_both_doors():
    api("POST", "/quotes", ALICE, {"quote_id": "Q-3", "customer": "Acme"})
    assert api("POST", "/quotes/Q-3/send", ALICE) == (409, {"detail": "an empty quote cannot be sent"})
    assert tool(ALICE, "send_quote", {"quote_id": "Q-3"}) == \
        ("refused", "409: an empty quote cannot be sent")


def test_another_persons_quote_does_not_exist_for_you():
    api("POST", "/quotes", ALICE, {"quote_id": "Q-4", "customer": "Acme"})
    assert api("GET", "/quotes/Q-4", BOB)[0] == 404
    assert tool(BOB, "get_quote", {"quote_id": "Q-4"}) == ("refused", "404: no quote Q-4")
    assert api("GET", "/quotes", BOB)[1] == {"quotes": []}


def test_without_the_permission_nothing_is_written(memory):
    viewer = person("vic@shop.example", ["sales"], [])
    assert api("POST", "/quotes", viewer, {"quote_id": "Q-5", "customer": "Acme"})[0] == 403
    assert "Q-5" not in memory


def test_the_process_view_shows_the_quote_workflow_as_bpmn():
    status, body = api("GET", "/process/bpmn", ALICE)
    assert status == 200 and body["processes"] == ["quote"]
    for i in ("Task_draft_quote", "Task_send_quote", "Task_accept_quote", "Flow_Task_draft_quote__Task_send_quote"):
        assert f'id="{i}"' in body["bpmn"]