"""Every request gets an id, carries it through both doors and every seam, and ends in ONE safe log line.
Real uvicorn servers and a real MCP client, as in test_bricks_doors.py: the claims under test — the id
reaches a use case on a worker thread, a neighbour sees the same id, a retry on that thread reaches the
request line — are about context crossing threads and processes, which a mock would simply assert.
"""
import json
import logging
import os
import socket
import subprocess
import sys
import threading
import time
import anyio
import httpx
import pytest
import uvicorn
from mcp import ClientSession
from mcp.client.streamable_http import create_mcp_http_client, streamable_http_client
from sqlalchemy.exc import OperationalError
from erp_platform import requestlog, retry
from erp_platform.bricks import Brick, Refused, serve
from erp_platform.seam import Seam, gateway_headers, register_local, unregister_local
SERVER_THREAD = "requestlog-uvicorn"
# ── the neighbour: echoes the request id its use case sees ─────────────────────────────────
ECHO = Brick("echo")
@ECHO.use_case(kind="query", route="GET /echo")
def echo(caller: dict) -> dict:
"""The request id this neighbour's use case runs under."""
return {"request_id": requestlog.request_id()}
LOCAL_ECHO = Seam("echo", "UNSET_ECHO_BASE_URL", roles=["sales"], permissions=[])
NET_ECHO = Seam("echo-net", "UNSET_ECHO_NET_BASE_URL", roles=["sales"], permissions=[])
# ── the brick under test ────────────────────────────────────────────────────────────────────
PING = Brick("ping")
FLAKY = {"left": 0}
class Pg(Exception):
sqlstate = "57P03"
@PING.use_case(kind="query", route="GET /whoami/{item_id}")
def whoami(caller: dict, item_id: str) -> dict:
"""The request id and thread this use case runs on."""
return {"request_id": requestlog.request_id(), "thread": threading.current_thread().name}
@PING.use_case(kind="query", route="GET /ask")
def ask_neighbour(caller: dict, where: str = "local") -> dict:
"""Call the echo brick over a seam and return the id it saw."""
r = (LOCAL_ECHO if where == "local" else NET_ECHO).get("/echo", actor=caller["email"])
assert r.ok, r.reason
return {"mine": requestlog.request_id(), "neighbour": r.data["request_id"]}
@PING.use_case(kind="query", route="GET /flaky")
def flaky(caller: dict) -> dict:
"""Fails with 'cannot connect now' until FLAKY runs out, then answers."""
if FLAKY["left"]:
FLAKY["left"] -= 1
raise OperationalError("SELECT", {}, Pg("cannot connect now"))
return {"ok": True}
@PING.use_case(kind="command", route="POST /explode")
def explode(caller: dict, note: str) -> dict:
"""Crash with a message that quotes customer data."""
raise ValueError(f"customer {caller['email']} wrote {note}")
@PING.use_case(kind="command", route="POST /guarded", permission="ping:thing:do")
def guarded(caller: dict) -> dict:
"""Needs a permission nobody here has."""
return {"ok": True}
# ── who is calling ──────────────────────────────────────────────────────────────────────────
def as_user(sub: str, email: str) -> dict:
return gateway_headers({"sub": sub, "email": email, "app_roles": ["sales"], "permissions": []})
ALICE = as_user("alice@shop.example", "alice@shop.example") # a sub that IS an address
USER = as_user("u-7f3a", "carol@shop.example") # an opaque sub, the production shape
SECRET_BITS = ("alice@shop.example", "carol@shop.example", ALICE["X-Dodil-User-JWT"].split(".")[1],
USER["X-Dodil-User-JWT"].split(".")[1], "hunter2-note")
# ── servers and captured lines ──────────────────────────────────────────────────────────────
def _serve(app):
with socket.socket() as s:
s.bind(("127.0.0.1", 0))
port = s.getsockname()[1]
server = uvicorn.Server(uvicorn.Config(app, host="127.0.0.1", port=port, log_level="critical"))
thread = threading.Thread(target=server.run, daemon=True, name=SERVER_THREAD)
thread.start()
deadline = time.time() + 10
while not server.started and time.time() < deadline:
time.sleep(0.02)
assert server.started
return server, thread, f"http://127.0.0.1:{port}"
@pytest.fixture(scope="module")
def base():
register_local("echo", ECHO.routes_app()) # tier 1: the neighbour in this process
echo_server, echo_thread, echo_url = _serve(ECHO.app()) # tier 2: the same neighbour over HTTP
NET_ECHO.base = echo_url
server, thread, url = _serve(serve([PING], name="ping", version="0.1.0"))
yield url
for s, t in ((server, thread), (echo_server, echo_thread)):
s.should_exit = True
t.join(5)
unregister_local("echo")
class _Lines(logging.Handler):
def __init__(self):
super().__init__(logging.DEBUG)
self.records: list = []
def emit(self, record):
self.records.append(record)
def lines(self, rid: str, timeout: float = 3.0) -> list:
"""The request lines for `rid`. Waited for: a line is written after its response has gone."""
deadline = time.time() + timeout
while True:
found = [json.loads(r.getMessage()) for r in self.records if r.name == "erp_platform.request"]
found = [f for f in found if f["request_id"] == rid]
if found or time.time() > deadline:
return found
time.sleep(0.02)
@pytest.fixture
def logs():
handler = _Lines()
loggers = [logging.getLogger(n) for n in ("erp_platform.request", "erp_platform.retry")]
old = [lg.level for lg in loggers]
for lg in loggers:
lg.addHandler(handler)
lg.setLevel(logging.INFO)
yield handler
for lg, lvl in zip(loggers, old):
lg.removeHandler(handler)
lg.setLevel(lvl)
def _one(logs, rid: str) -> dict:
found = logs.lines(rid)
assert len(found) == 1, found
return found[0]
async def _mcp(base: str, headers: dict, tool: str, args: dict):
async with create_mcp_http_client(headers=headers) as http:
async with streamable_http_client(f"{base}/mcp", http_client=http) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
return await session.call_tool(tool, args)
def via_mcp(base, headers, tool, args=None):
return anyio.run(_mcp, base, headers, tool, args or {})
def _assert_clean(line: dict) -> None:
assert tuple(line) == requestlog.FIELDS
text = json.dumps(line)
for bit in SECRET_BITS:
assert bit not in text, f"{bit!r} leaked into the request line"
assert "@" not in text
# ── the id ──────────────────────────────────────────────────────────────────────────────────
def test_an_id_is_minted_when_the_caller_sends_none(base, logs):
r = httpx.get(f"{base}/whoami/Q-1", headers=USER)
rid = r.headers["x-request-id"]
assert r.status_code == 200 and len(rid) == 32 and int(rid, 16) >= 0
assert r.json()["request_id"] == rid
assert _one(logs, rid)["status"] == 200
def test_a_sane_caller_id_is_kept(base, logs):
rid = "edge.trace:01J9-ab_c"
r = httpx.get(f"{base}/whoami/Q-1", headers={**USER, "X-Request-Id": rid})
assert r.headers["x-request-id"] == rid and r.json()["request_id"] == rid
assert _one(logs, rid)["use_case"] == "whoami"
@pytest.mark.parametrize("bad", ["x" * 129, "abc def", "id;rm -rf /", "<script>", b"\xfc-umlaut", ""])
def test_an_unsafe_caller_id_is_replaced(base, bad):
r = httpx.get(f"{base}/whoami/Q-1", headers={**USER, "X-Request-Id": bad})
rid = r.headers["x-request-id"]
assert rid != bad and len(rid) == 32 and r.json()["request_id"] == rid
def test_the_id_reaches_a_use_case_on_a_worker_thread(base):
"""anyio runs a sync use case in a copy of the request's context — checked, not assumed."""
r = httpx.get(f"{base}/whoami/Q-1", headers={**USER, "X-Request-Id": "thread-check"})
body = r.json()
assert body["request_id"] == "thread-check"
assert body["thread"] != SERVER_THREAD # it really ran off the event loop
def test_the_id_is_on_the_sign_in_gates_401(base, logs):
r = httpx.post(f"{base}/mcp", headers={"X-Request-Id": "gate-401"}, json={})
assert r.status_code == 401 and r.headers["x-request-id"] == "gate-401"
line = _one(logs, "gate-401")
assert (line["door"], line["outcome"], line["status"], line["sub"]) == ("mcp", "refused", 401, None)
def test_the_id_is_on_a_routes_401(base, logs):
r = httpx.get(f"{base}/whoami/Q-1", headers={"X-Request-Id": "route-401"})
assert r.status_code == 401 and r.headers["x-request-id"] == "route-401"
assert _one(logs, "route-401")["outcome"] == "refused"
def test_a_crash_is_a_500_with_the_id_and_a_line_without_the_message(base, logs):
r = httpx.post(f"{base}/explode", headers={**ALICE, "X-Request-Id": "crash-1"},
json={"note": "hunter2-note"})
assert r.status_code == 500 and r.headers["x-request-id"] == "crash-1"
assert r.json() == {"detail": "internal error", "request_id": "crash-1"}
line = _one(logs, "crash-1")
_assert_clean(line)
assert (line["status"], line["outcome"], line["error"], line["use_case"]) == \
(500, "error", "ValueError", "explode")
# ── the line ────────────────────────────────────────────────────────────────────────────────
def test_a_line_has_exactly_the_documented_fields_and_no_customer_data(base, logs):
httpx.get(f"{base}/whoami/Q-1", headers={**USER, "X-Request-Id": "fields-1"})
line = _one(logs, "fields-1")
_assert_clean(line)
assert line["method"] == "GET" and line["path"] == "/whoami/{item_id}" # the template, not Q-1
assert (line["door"], line["sub"], line["retries"], line["outcome"], line["error"]) == \
("http", "u-7f3a", 0, "ok", None)
assert isinstance(line["duration_ms"], float)
def test_a_sub_that_is_an_address_is_pseudonymised(base, logs):
httpx.get(f"{base}/whoami/Q-1", headers={**ALICE, "X-Request-Id": "pseudo-1"})
httpx.get(f"{base}/whoami/Q-2", headers={**ALICE, "X-Request-Id": "pseudo-2"})
a, b = _one(logs, "pseudo-1"), _one(logs, "pseudo-2")
_assert_clean(a)
assert a["sub"].startswith("anon:") and a["sub"] == b["sub"] # still correlates per person
def test_a_refusal_is_logged_as_refused(base, logs):
r = httpx.post(f"{base}/guarded", headers={**USER, "X-Request-Id": "refused-1"}, json={})
assert r.status_code == 403
line = _one(logs, "refused-1")
assert (line["outcome"], line["use_case"], line["status"]) == ("refused", "guarded", 403)
def test_an_unmatched_path_is_logged_without_an_address_in_it(base, logs):
httpx.get(f"{base}/people/alice@shop.example", headers={**USER, "X-Request-Id": "nf-1"})
line = _one(logs, "nf-1")
_assert_clean(line)
assert line["path"] == "/people/{redacted}" and line["status"] == 404
# ── the agent door ──────────────────────────────────────────────────────────────────────────
def test_a_tool_call_is_logged_by_its_use_case_under_the_callers_id(base, logs):
result = via_mcp(base, {**ALICE, "X-Request-Id": "mcp-1"}, "whoami", {"item_id": "Q-1"})
assert json.loads(result.content[0].text)["request_id"] == "mcp-1"
lines = logs.lines("mcp-1")
assert all(line["door"] == "mcp" for line in lines)
tool = [line for line in lines if line["use_case"] == "whoami"]
assert len(tool) == 1 and tool[0]["outcome"] == "ok" and tool[0]["path"] == "/mcp"
for line in lines:
_assert_clean(line)
def test_a_tool_crash_is_an_error_line_although_mcp_answers_200(base, logs):
result = via_mcp(base, {**ALICE, "X-Request-Id": "mcp-crash"}, "explode", {"note": "hunter2-note"})
assert result.is_error
tool = [line for line in logs.lines("mcp-crash") if line["use_case"] == "explode"]
assert len(tool) == 1
_assert_clean(tool[0])
assert (tool[0]["status"], tool[0]["outcome"], tool[0]["error"]) == (200, "error", "ValueError")
# ── seams ───────────────────────────────────────────────────────────────────────────────────
@pytest.mark.parametrize("where", ["local", "net"])
def test_a_neighbour_sees_the_same_id_over_a_seam(base, logs, where):
"""Tier 1 (in process) and tier 2 (HTTP): the neighbour's use case and its line share the id."""
rid = f"seam-{where}"
r = httpx.get(f"{base}/ask", params={"where": where}, headers={**ALICE, "X-Request-Id": rid})
assert r.status_code == 200, r.text
assert r.json() == {"mine": rid, "neighbour": rid}
by_case = {line["use_case"]: line for line in logs.lines(rid)}
assert set(by_case) == {"ask_neighbour", "echo"}, by_case
for line in by_case.values():
_assert_clean(line)
# ── retries ─────────────────────────────────────────────────────────────────────────────────
def test_retries_are_counted_on_the_line_and_named_in_the_warning(base, logs, monkeypatch):
monkeypatch.setattr(retry, "BACKOFF", (0.01,))
FLAKY["left"] = 2
r = httpx.get(f"{base}/flaky", headers={**USER, "X-Request-Id": "retry-1"})
assert r.status_code == 200
assert _one(logs, "retry-1")["retries"] == 2
warnings = [json.loads(rec.getMessage()) for rec in logs.records if rec.name == "erp_platform.retry"]
mine = [w for w in warnings if w["request_id"] == "retry-1"]
assert [w["attempt"] for w in mine] == [1, 2] and mine[0]["sqlstate"] == "57P03"
assert all(rec.levelno == logging.WARNING for rec in logs.records if rec.name == "erp_platform.retry")
# ── configuration ───────────────────────────────────────────────────────────────────────────
_PROBE = """
import anyio, httpx
from erp_platform.bricks import Brick, serve
async def main():
app = serve([Brick("probe")], name="probe", version="0")
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://t") as c:
await c.get("/healthz", headers={"X-Request-Id": "probe-1"})
anyio.run(main)
"""
def _probe(**env) -> str:
environ = {k: v for k, v in os.environ.items() if k != "LOG_LEVEL"} | env
out = subprocess.run([sys.executable, "-c", _PROBE], capture_output=True, text=True, env=environ,
timeout=60)
assert out.returncode == 0, out.stderr
return out.stdout
def test_with_no_setup_the_line_goes_to_stdout():
lines = [json.loads(s) for s in _probe().splitlines() if s.startswith("{")]
assert [line["request_id"] for line in lines] == ["probe-1"]
assert lines[0]["path"] == "/healthz" and lines[0]["outcome"] == "ok"
def test_log_level_silences_it():
assert "probe-1" not in _probe(LOG_LEVEL="WARNING")
def test_a_bad_log_level_does_not_stop_the_app():
assert "probe-1" in _probe(LOG_LEVEL="LOUD")