"""0.9.6: erp_platform.after_response (work after the commit and the answer) and Refused(gate=, fix=)."""
import base64
import json
import logging
from types import SimpleNamespace
import anyio
import pytest
from fastapi.testclient import TestClient
from mcp.server.mcpserver.exceptions import ToolError
from sqlalchemy import Column, MetaData, String, Table, create_engine, insert, select
from erp_platform import after_response, join_after_response
from erp_platform.after import response_scope
from erp_platform.bricks import Brick, Refused, _tool
RAN: list = []
md = MetaData()
notes = Table("t_note", md, Column("id", String, primary_key=True))
ENGINE = create_engine("sqlite://", connect_args={"check_same_thread": False},
poolclass=__import__("sqlalchemy.pool", fromlist=["StaticPool"]).StaticPool)
md.create_all(ENGINE)
B = Brick("after", instructions="after_response under test.")
def _note(tag):
RAN.append(tag)
@B.use_case(kind="command", route="POST /ok/{tag}", idempotent=True)
def ok(caller: dict, tag: str) -> dict:
"""Commits, then asks for work after the response."""
with ENGINE.begin() as conn:
conn.execute(insert(notes).values(id=tag))
after_response(_note, tag, conn=conn)
return {"tag": tag, "ran_yet": tag in RAN}
@B.use_case(kind="command", route="POST /rolled-back/{tag}", idempotent=True)
def rolled_back(caller: dict, tag: str) -> dict:
"""Registers bound work, then its transaction rolls back — and the use case still answers."""
try:
with ENGINE.begin() as conn:
after_response(_note, tag, conn=conn)
raise ValueError("abandon")
except ValueError:
pass
return {"tag": tag}
@B.use_case(kind="command", route="POST /refused/{tag}", idempotent=True)
def refused(caller: dict, tag: str) -> dict:
"""Registers work, then refuses."""
after_response(_note, tag)
raise Refused(409, f"{tag} has no e-invoice unit code yet", gate="einvoice_unit_code",
fix="give it before a document uses it")
@B.use_case(kind="command", route="POST /failing/{tag}", idempotent=True)
def failing(caller: dict, tag: str) -> dict:
"""Its after-work fails; the answer stands."""
after_response(lambda: 1 / 0)
after_response(_note, tag)
return {"tag": tag}
def headers():
payload = base64.urlsafe_b64encode(json.dumps({"permissions": []}).encode()).decode()
return {"X-Dodil-User": json.dumps({"sub": "a@x.example", "email": "a@x.example", "app_roles": ["sales"]}),
"X-Dodil-User-JWT": f"h.{payload.rstrip('=')}.s"}
@pytest.fixture(autouse=True)
def _clean():
RAN.clear()
yield
join_after_response()
@pytest.fixture(scope="module")
def http():
return TestClient(B.routes_app())
def mcp(name, **args):
uc = next(u for u in B.use_cases if u.name == name)
return anyio.run(lambda: _tool(uc)(ctx=SimpleNamespace(headers=headers()), **args))
def test_http_runs_it_after_the_answer_and_the_test_client_joins_it(http):
r = http.post("/ok/h1", headers=headers())
assert r.status_code == 200 and r.json() == {"tag": "h1", "ran_yet": False} # not before the answer
assert RAN == ["h1"] # but before the call returned
def test_mcp_runs_it_after_the_answer(http):
out = mcp("ok", tag="m1")
join_after_response()
assert RAN == ["m1"] and "m1" in json.dumps(out, default=str)
@pytest.mark.parametrize("door", ["http", "mcp"])
def test_a_rolled_back_transaction_runs_nothing(http, door):
if door == "http":
assert http.post("/rolled-back/r1", headers=headers()).status_code == 200
else:
mcp("rolled_back", tag="r1")
join_after_response()
assert RAN == []
@pytest.mark.parametrize("door", ["http", "mcp"])
def test_a_refusal_runs_nothing_and_carries_its_gate(http, door):
if door == "http":
r = http.post("/refused/x1", headers=headers())
assert r.status_code == 409 and r.json() == {
"detail": "x1 has no e-invoice unit code yet: give it before a document uses it",
"gate": "einvoice_unit_code", "reason": "x1 has no e-invoice unit code yet",
"fix": "give it before a document uses it"}
else:
with pytest.raises(ToolError, match="^409: x1 has no e-invoice unit code yet: give it before"):
mcp("refused", tag="x1")
join_after_response()
assert RAN == []
@pytest.mark.parametrize("door", ["http", "mcp"])
def test_a_failing_job_is_logged_and_the_answer_stands(http, door, caplog):
caplog.set_level(logging.WARNING, logger="erp_platform.after_response")
if door == "http":
assert http.post("/failing/f1", headers=headers()).json() == {"tag": "f1"}
else:
mcp("failing", tag="f1")
join_after_response()
assert RAN == ["f1"] # the next job still ran
assert any("ZeroDivisionError" in r.message and "after_response_failed" in r.message for r in caplog.records)
def test_a_plain_refusal_body_is_unchanged():
assert Refused(404, "no quote Q-9").body() == {"detail": "no quote Q-9"}
assert Refused(422, "bad", fields={"field": "qty"}).body() == {"detail": "bad", "field": "qty"}
with pytest.raises(ValueError):
Refused(422, "bad", fields={"detail": "x"})
def test_outside_a_door_it_runs_at_once_or_at_the_end_of_a_scope():
after_response(_note, "direct")
join_after_response()
assert RAN == ["direct"]
with ENGINE.begin() as conn:
with pytest.raises(RuntimeError, match="response_scope"):
after_response(_note, "open-txn", conn=conn)
with response_scope():
with ENGINE.begin() as conn:
after_response(_note, "scoped", conn=conn)
assert RAN == ["direct"]
join_after_response()
assert RAN == ["direct", "scoped"]
with pytest.raises(KeyError):
with response_scope():
after_response(_note, "never")
raise KeyError("x")
join_after_response()
assert RAN == ["direct", "scoped"]