"""A seam must behave identically whether its neighbour is in this process (tier 1) or across the
network (tiers 2 and 3). Every case here runs the SAME neighbour app both ways and compares.
The HTTP side is a real uvicorn server on a free port, not a mock: the claim being tested is that
the in-process path is indistinguishable from the network one, so the network one has to be real.
"""
import base64
import json
import socket
import threading
import time
import pytest
import uvicorn
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import RedirectResponse
from erp_platform import seam
from erp_platform.seam import Seam, register_local, unregister_local
def _claims(request: Request) -> dict:
"""What a receiving brick does: decode the gateway JWT, take permissions from it."""
token = request.headers.get("X-Dodil-User-JWT", "")
try:
payload = token.split(".")[1]
return json.loads(base64.urlsafe_b64decode(payload + "=" * (-len(payload) % 4)))
except (IndexError, ValueError):
return {}
ledger = FastAPI()
@ledger.post("/journals")
def post_journal(body: dict, request: Request):
c = _claims(request)
if "gl:journal:post" not in c.get("permissions", []):
raise HTTPException(403, "gl:journal:post required")
return {"posted": True, "by": c.get("sub"), "lines": body.get("lines")}
@ledger.get("/closed")
def closed_period():
raise HTTPException(409, "period 2026-09 is closed")
@ledger.get("/boom")
def boom():
raise RuntimeError("ledger exploded")
@ledger.get("/moved")
def moved():
return RedirectResponse("/journals", status_code=307)
@ledger.get("/slow")
def slow():
time.sleep(3)
return {"late": True}
def _free_port() -> int:
with socket.socket() as s:
s.bind(("127.0.0.1", 0))
return s.getsockname()[1]
@pytest.fixture(scope="module")
def ledger_url():
port = _free_port()
server = uvicorn.Server(uvicorn.Config(ledger, host="127.0.0.1", port=port, log_level="critical"))
thread = threading.Thread(target=server.run, daemon=True)
thread.start()
deadline = time.time() + 10
while not server.started and time.time() < deadline:
time.sleep(0.02)
assert server.started, "uvicorn did not start"
yield f"http://127.0.0.1:{port}"
server.should_exit = True
thread.join(5)
ALL_PERMS = ["gl:journal:post"]
def _over_http(url, monkeypatch, permissions=ALL_PERMS, timeout=seam.TIMEOUT) -> Seam:
monkeypatch.setenv("LEDGER_BASE_URL", url)
unregister_local("ledger")
return Seam("ledger", "LEDGER_BASE_URL", roles=["bookkeeper"], permissions=permissions,
timeout=timeout)
def _in_process(monkeypatch, permissions=ALL_PERMS, timeout=seam.TIMEOUT) -> Seam:
monkeypatch.delenv("LEDGER_BASE_URL", raising=False)
register_local("ledger", ledger)
return Seam("ledger", "LEDGER_BASE_URL", roles=["bookkeeper"], permissions=permissions,
timeout=timeout)
@pytest.fixture(autouse=True)
def _clean_registry():
yield
unregister_local("ledger")
unregister_local("sales")
def _shape(r) -> tuple:
return (r.ok, r.status, r.reason, r.data)
# ── parity: one app, two transports, the same reply ──────────────────────────────────────
def test_success_is_identical(ledger_url, monkeypatch):
body = {"lines": [{"dr": "1000.00"}, {"cr": "1000.00"}]}
http = _over_http(ledger_url, monkeypatch).post("/journals", body, actor="alice@shop.example")
local = _in_process(monkeypatch).post("/journals", body, actor="alice@shop.example")
assert http.ok and local.ok
assert _shape(http) == _shape(local)
assert local.data == {"posted": True, "by": "alice@shop.example", "lines": body["lines"]}
def test_the_permission_guard_runs_in_process(ledger_url, monkeypatch):
"""The security property. A direct function call would skip the api/ check; this must not."""
http = _over_http(ledger_url, monkeypatch, permissions=[]).post("/journals", {}, "mallory")
local = _in_process(monkeypatch, permissions=[]).post("/journals", {}, "mallory")
assert local.status == 403 and not local.ok
assert _shape(http) == _shape(local)
def test_a_refusal_carries_its_reason(ledger_url, monkeypatch):
http = _over_http(ledger_url, monkeypatch).get("/closed")
local = _in_process(monkeypatch).get("/closed")
assert local.status == 409 and "period 2026-09 is closed" in local.reason
assert _shape(http) == _shape(local)
def test_a_crash_is_a_reply_not_an_exception(ledger_url, monkeypatch):
"""And reported as a crash. With httpx's default (raise_app_exceptions=True) it would come back
as "ledger unreachable" with status 0 — a reply, but a false one: the ledger is up and failing."""
http = _over_http(ledger_url, monkeypatch).get("/boom")
local = _in_process(monkeypatch).get("/boom") # must not raise
assert local.status == 500 and not local.ok
assert _shape(http) == _shape(local)
def test_a_redirect_is_not_followed(ledger_url, monkeypatch):
"""Starlette's TestClient follows redirects by default; the network path does not."""
http = _over_http(ledger_url, monkeypatch).get("/moved")
local = _in_process(monkeypatch).get("/moved")
assert local.status == 307
assert _shape(http) == _shape(local)
def test_a_timeout_is_reported_the_same_way(ledger_url, monkeypatch):
http = _over_http(ledger_url, monkeypatch, timeout=0.5).get("/slow")
local = _in_process(monkeypatch, timeout=0.5).get("/slow")
for r in (http, local):
assert not r.ok and r.status == 0
assert r.reason.startswith("ledger unreachable")
# ── absence, and which transport wins ────────────────────────────────────────────────────
def test_absent_is_a_no_op(monkeypatch):
monkeypatch.delenv("LEDGER_BASE_URL", raising=False)
s = Seam("ledger", "LEDGER_BASE_URL", roles=[], permissions=[])
r = s.post("/journals", {})
assert not s.installed and not r.ok and r.reason == "ledger is not installed"
def test_a_down_neighbour_is_a_reply(monkeypatch):
r = _over_http(f"http://127.0.0.1:{_free_port()}", monkeypatch).get("/closed")
assert not r.ok and r.reason.startswith("ledger unreachable")
def test_registration_wins_over_a_base_url(monkeypatch):
"""In this process beats leaving through the network — even to a URL that is down."""
monkeypatch.setenv("LEDGER_BASE_URL", f"http://127.0.0.1:{_free_port()}")
register_local("ledger", ledger)
s = Seam("ledger", "LEDGER_BASE_URL", roles=[], permissions=ALL_PERMS)
assert s.post("/journals", {"lines": []}, "alice").ok
# ── tier 1 must open no socket and must not deadlock ─────────────────────────────────────
def test_in_process_opens_no_socket(monkeypatch):
s = _in_process(monkeypatch)
def refuse(*_a, **_k):
raise AssertionError("tier 1 opened a network connection")
monkeypatch.setattr(socket.socket, "connect", refuse)
monkeypatch.setattr(seam.httpx, "request", refuse)
assert s.post("/journals", {"lines": []}, "alice").ok
def test_an_async_handler_can_call_a_neighbour_in_process(monkeypatch):
"""A shared event loop would deadlock here: the async handler blocks the loop its neighbour
needs. A fresh thread and loop per call cannot."""
monkeypatch.delenv("LEDGER_BASE_URL", raising=False)
monkeypatch.delenv("SALES_BASE_URL", raising=False)
register_local("ledger", ledger)
to_ledger = Seam("ledger", "LEDGER_BASE_URL", roles=[], permissions=ALL_PERMS, timeout=5)
sales = FastAPI()
@sales.post("/accept")
async def accept(request: Request):
reply = to_ledger.post("/journals", {"lines": ["quote-7"]}, _claims(request).get("sub", ""))
return {"accepted": True, "ledger": reply.data}
register_local("sales", sales)
to_sales = Seam("sales", "SALES_BASE_URL", roles=[], permissions=[], timeout=5)
started = time.time()
r = to_sales.post("/accept", {}, "alice")
assert r.ok, r.reason
assert r.data == {"accepted": True,
"ledger": {"posted": True, "by": "alice", "lines": ["quote-7"]}}
assert time.time() - started < 5, "nested in-process call looks deadlocked"