dodil/erp-platform

tests/test_democlock.py

"""The business clock moved in process (work.set_clock) and the demo clock route (erp_platform.democlock):
through both doors, behind its permission — and impossible on a live deployment."""
import base64
import json
from datetime import datetime, timedelta, timezone

import anyio
import httpx
import httpx2
import pytest
from mcp import ClientSession
from mcp.client.streamable_http import streamable_http_client

from erp_platform import democlock, work
from erp_platform.bricks import Brick

CLK = Brick("clk", version="0.1.0")
democlock.mount(CLK)
APP = CLK.app()


@pytest.fixture(autouse=True)
def real_clock():
    work.reset_clock()
    yield
    work.reset_clock()


def person(perms):
    payload = base64.urlsafe_b64encode(json.dumps({"permissions": perms}).encode()).decode().rstrip("=")
    return {"X-Dodil-User": json.dumps({"sub": "p", "email": "p@x", "app_roles": ["demo"]}),
            "X-Dodil-User-JWT": f"h.{payload}.s"}


DEMO, NOBODY = person(["clk:clock:set"]), person([])


def api(method, path, headers, body=None):
    async def go():
        async with httpx.AsyncClient(transport=httpx.ASGITransport(app=APP), base_url="http://clk") as c:
            return await c.request(method, path, headers=headers, json=body)
    r = anyio.run(go)
    return r.status_code, r.json()


def tool(headers, name, args):
    async def go():
        async with APP.router.lifespan_context(APP):
            async with httpx2.AsyncClient(transport=httpx2.ASGITransport(app=APP), headers=headers,
                                          base_url="http://clk") as http:
                async with streamable_http_client("http://clk/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))
    return anyio.run(go)


def test_set_clock_freezes_advances_and_resets():
    try:
        at = datetime(2026, 3, 29, 0, 30, tzinfo=timezone.utc)
        assert work.set_clock(at=at, freeze=True) == at
        assert work.now() == at
        assert work.advance_clock(timedelta(hours=1)) == at + timedelta(hours=1)
        work.set_clock(offset=timedelta(days=3))
        assert abs(work.now() - (datetime.now(timezone.utc) + timedelta(days=3))) < timedelta(seconds=5)
    finally:
        work.reset_clock()
    assert abs(work.now() - datetime.now(timezone.utc)) < timedelta(seconds=5)


@pytest.mark.parametrize("env", ["prod", "live", "PROD"])
def test_the_clock_is_never_moved_in_production(monkeypatch, env):
    monkeypatch.setenv("DODIL_ENV", env)
    with pytest.raises(work.DemoClockRefused):
        work.set_clock(at=datetime(2026, 1, 1, tzinfo=timezone.utc))
    with pytest.raises(work.DemoClockRefused):
        work.advance_clock(timedelta(days=1))


def test_a_clock_set_before_prod_is_refused_at_every_read(monkeypatch):
    try:
        work.set_clock(offset=timedelta(days=1))
        monkeypatch.setenv("DODIL_ENV", "prod")
        with pytest.raises(work.DemoClockRefused):
            work.now()
    finally:
        work.reset_clock()


@pytest.mark.parametrize("text,utc", [
    ("2026-03-29T00:30 Europe/London", "2026-03-29T00:30:00+00:00"),     # GMT, before the change
    ("2026-03-29T01:30 Europe/London", "2026-03-29T01:30:00+00:00"),     # does not exist: lands 02:30 BST
    ("2026-03-29T02:30 Europe/London", "2026-03-29T01:30:00+00:00"),     # BST
    ("2026-10-25T01:30 Europe/London", "2026-10-25T00:30:00+00:00"),     # happens twice: the first
    ("2026-10-23T08:00Z", "2026-10-23T08:00:00+00:00"),
])
def test_instants_are_read_in_their_zone(text, utc):
    assert work.parse_instant(text).isoformat() == utc


def test_the_route_moves_the_clock_through_both_doors():
    assert api("POST", "/demo-clock", DEMO, {"at": "2026-11-02T09:00 Europe/London"})[1]["now"] == "2026-11-02T09:00:00Z"
    assert work.now() == datetime(2026, 11, 2, 9, 0, tzinfo=timezone.utc)
    verdict, state = tool(DEMO, "set_clk_demo_clock", {"advance": "+1d"})
    assert verdict == "ok" and state["now"] == "2026-11-03T09:00:00Z" and state["frozen"] is True
    status, seen = api("GET", "/demo-clock", NOBODY)                 # reading it needs no permission
    assert status == 200 and (seen["now"], seen["moved"], seen["frozen"]) == ("2026-11-03T09:00:00Z", True, True)
    assert api("POST", "/demo-clock", DEMO, {"reset": True})[1]["moved"] is False


def test_the_route_needs_its_permission_and_one_instruction():
    assert api("POST", "/demo-clock", NOBODY, {"reset": True}) == (403, {"detail": "missing permission: clk:clock:set"})
    assert tool(NOBODY, "set_clk_demo_clock", {"reset": True}) == ("refused", "403: missing permission: clk:clock:set")
    assert api("POST", "/demo-clock", DEMO, {"at": "2026-11-02T09:00Z", "reset": True})[0] == 422
    assert api("POST", "/demo-clock", DEMO, {"advance": "three days"})[0] == 422
    assert work._SET == {}


@pytest.mark.parametrize("env", ["prod", "live"])
def test_the_route_does_not_exist_on_a_live_deployment(monkeypatch, env):
    monkeypatch.setenv("DODIL_ENV", env)
    assert api("GET", "/demo-clock", DEMO) == (404, {"detail": "not found"})
    assert api("POST", "/demo-clock", DEMO, {"advance": "+3d"}) == (404, {"detail": "not found"})
    assert tool(DEMO, "set_clk_demo_clock", {"advance": "+3d"}) == ("refused", "404: not found")
    assert work._SET == {}