"""0.9.13: the per-request identity map (erp_platform.repository.request_scope) and the version check without a SELECT.
An aggregate is read once per request tree per key per caller. These tests pin what must never change with it: two
requests never share a row, two callers never do (a seam call on a person's behalf and the brick's own grant
included), a write drops what it touched, and a concurrent save still loses with 409."""
import json
import threading
from datetime import datetime, timezone
import httpx
import pytest
from sqlalchemy import Column, Integer, MetaData, String, Table, TIMESTAMP, create_engine, event, text
from erp_platform import repository
from erp_platform.bricks import Brick, Refused
from erp_platform.repository import Aggregate, Conflict, Part, request_scope
from erp_platform.seam import Seam, _InProcessTransport, gateway_headers, register_local, unregister_local
from erp_platform.testing import count_statements
NOW = datetime(2026, 10, 25, 9, 0, tzinfo=timezone.utc)
md = MetaData()
orders = Table("zz_order", md, Column("order_id", String, primary_key=True), Column("owner", String),
Column("note", String), Column("version", Integer),
Column("created_at", TIMESTAMP(timezone=True)), Column("updated_at", TIMESTAMP(timezone=True)))
lines = Table("zz_line", md, Column("order_id", String, primary_key=True),
Column("line_no", Integer, primary_key=True, autoincrement=False), Column("qty", String),
Column("created_at", TIMESTAMP(timezone=True)))
ORDER = Aggregate(Part(orders, ("order_id",)), Part(lines, ("order_id", "line_no")), version="version")
STATE: dict = {}
@pytest.fixture
def engine(tmp_path):
e = create_engine(f"sqlite:///{tmp_path / 'idmap.db'}", connect_args={"check_same_thread": False, "timeout": 30})
md.create_all(e)
with e.begin() as c:
for oid, owner in (("o1", "ann@co.example"), ("o2", "bob@co.example")):
ORDER.save(c, {"zz_order": [{"order_id": oid, "owner": owner, "note": "new"}],
"zz_line": [{"order_id": oid, "line_no": 1, "qty": "1"}]}, now=NOW, expected_version=0)
STATE["engine"] = e
yield e
e.dispose()
def load(e, oid="o1"):
with e.connect() as c:
return ORDER.load(c, {"order_id": oid})
def test_outside_a_request_nothing_is_cached(engine):
with count_statements(engine) as n:
load(engine), load(engine)
assert n.statements == 4
def test_in_a_request_an_aggregate_is_read_once_and_each_answer_is_a_copy(engine):
with request_scope({"sub": "ann@co.example"}):
first = load(engine)
first["zz_order"][0]["note"] = "edited in place" # never what the next load answers
with count_statements(engine) as n:
second = load(engine)
assert n.statements == 0
assert second["zz_order"][0]["note"] == "new" and second.snapshot.version == 1
def test_two_requests_never_share_rows(engine):
with request_scope({"sub": "ann@co.example"}):
load(engine)
with count_statements(engine) as n, request_scope({"sub": "ann@co.example"}):
load(engine)
assert n.statements == 2
# ... nor two at the same time, on two threads, for the same person
seen, gate = [], threading.Barrier(2)
def one():
with request_scope({"sub": "ann@co.example"}) as tree:
load(engine)
gate.wait()
seen.append(tree)
ts = [threading.Thread(target=one) for _ in range(2)]
[t.start() for t in ts]
[t.join() for t in ts]
assert seen[0] is not seen[1]
def test_two_callers_in_one_request_tree_never_answer_each_others_reads(engine):
with request_scope({"sub": "ann@co.example"}):
load(engine)
with request_scope({"sub": "inventory", "service": "inventory"}): # a seam callee on the brick's own grant
with count_statements(engine) as n:
load(engine)
load(engine)
with request_scope({"sub": "bob@co.example"}):
with count_statements(engine) as m:
load(engine)
assert n.statements == 2 and m.statements == 2
def test_a_save_drops_the_entry_and_the_next_load_reads_again(engine):
with request_scope({"sub": "ann@co.example"}):
got = load(engine)
rows = {k: [dict(r) for r in v] for k, v in got.items()}
rows["zz_order"][0]["note"] = "saved"
with engine.begin() as c:
ORDER.save(c, rows, now=NOW, snapshot=got, expected_version=1)
with count_statements(engine) as n:
again = load(engine)
assert n.statements == 2 and again["zz_order"][0]["note"] == "saved" and again.snapshot.version == 2
def test_any_write_to_one_of_its_tables_drops_it_text_sql_included(engine):
with request_scope({"sub": "ann@co.example"}):
load(engine), load(engine, "o2")
with engine.begin() as c:
c.execute(text("UPDATE zz_line SET qty = '5' WHERE order_id = 'o1'"))
with count_statements(engine) as n:
got = load(engine)
with count_statements(engine) as m:
load(engine, "o2")
with engine.begin() as c: # a write elsewhere drops nothing
c.execute(text("CREATE TABLE IF NOT EXISTS zz_other (k TEXT PRIMARY KEY)"))
c.execute(text("INSERT INTO zz_other (k) VALUES ('x')"))
with count_statements(engine) as k:
load(engine)
assert n.statements == 2 and got["zz_line"][0]["qty"] == "5"
assert m.statements == 2 # the same table: every order on it goes, not only the one written
assert k.statements == 0
def test_a_rollback_drops_everything_and_a_read_after_an_own_write_is_not_kept(engine):
with request_scope({"sub": "ann@co.example"}):
with engine.connect() as c:
c.begin()
c.execute(text("UPDATE zz_order SET note = 'uncommitted' WHERE order_id = 'o2'"))
seen = ORDER.load(c, {"order_id": "o2"}) # SQLite shows a transaction its own write
c.rollback()
with count_statements(engine) as n:
after = load(engine, "o2")
assert seen["zz_order"][0]["note"] == "uncommitted"
assert n.statements == 2 and after["zz_order"][0]["note"] == "new"
def test_a_fresh_attempt_starts_empty_and_a_closed_tree_is_never_used(engine):
with request_scope({"sub": "ann@co.example"}) as tree:
load(engine)
with request_scope({"sub": "ann@co.example"}, fresh=True):
with count_statements(engine) as n:
load(engine)
assert n.statements == 2
assert tree.closed and repository.current_tree() is None
repository.adopt_tree(tree) # an after-response job holding the old tree
assert repository.current_tree() is None
def test_a_concurrent_save_still_loses_with_409(engine):
"""Two requests open version 1; one saves; the other's save — from its cached snapshot — is a Conflict."""
with request_scope({"sub": "ann@co.example"}):
a = load(engine)
load(engine) # served from the map
with engine.begin() as c: # a racer, outside this request
c.execute(ORDER.versions.insert().values(order_id="o1", version=2, saved_at=NOW))
c.execute(text("UPDATE zz_order SET version = 2, note = 'racer' WHERE order_id = 'o1'"))
rows = {k: [dict(r) for r in v] for k, v in a.items()}
rows["zz_order"][0]["note"] = "mine"
with pytest.raises(Conflict):
with repository.conflicts(), engine.begin() as c:
ORDER.save(c, rows, now=NOW, snapshot=a, expected_version=1)
assert load(engine)["zz_order"][0]["note"] == "racer"
def test_a_stale_expected_version_is_refused_without_a_statement(engine):
got = load(engine)
with engine.begin() as c, count_statements(engine) as n:
with pytest.raises(Conflict):
ORDER.save(c, {k: [dict(r) for r in v] for k, v in got.items()}, now=NOW, snapshot=got, expected_version=7)
assert n.statements == 0
def test_upsert_new_rows_skips_the_key_lookup(engine):
rows = [{"order_id": "o9", "line_no": 1, "qty": "2", "created_at": NOW}]
with engine.begin() as c, count_statements(engine) as n:
repository.upsert(c, lines, rows, ("order_id", "line_no"), new=True)
with engine.begin() as c, count_statements(engine) as m:
repository.upsert(c, lines, [dict(rows[0], qty="3")], ("order_id", "line_no"))
assert n.statements == 1 and m.statements == 2 # the default still looks the keys up first
# ── through the doors and a tier-1 seam: the tree is shared per caller, never across callers ─────────────────
ORD = Brick("zzord", instructions="test")
CALL = Brick("zzcall", instructions="test")
ORD_SEAM = Seam("zzord", "ZZORD_BASE_URL", roles=["service"], permissions=["zz:order:read"], owner="zzcall")
@ORD.use_case(kind="query", route="GET /orders/{order_id}", permission=None)
def show_order(caller: dict, order_id: str) -> dict:
"""An order its owner (or a brick's service grant) may see."""
with STATE["engine"].connect() as c:
got = ORDER.load(c, {"order_id": order_id})
if got is None or not (caller.get("service") or got["zz_order"][0]["owner"] == caller.get("sub")):
raise Refused(404, "no such order")
return {"order_id": order_id, "owner": got["zz_order"][0]["owner"]}
@CALL.use_case(kind="query", route="GET /twice/{order_id}", permission=None)
def ask_twice(caller: dict, order_id: str) -> dict:
"""Ask the order brick twice on this brick's own grant, then once as the person."""
own = [ORD_SEAM.get(f"/orders/{order_id}", actor="zzcall") for _ in range(2)]
as_person = ORD_SEAM.get(f"/orders/{order_id}", caller=caller)
return {"own": [r.status for r in own], "as_person": as_person.status}
def _ask(app, path, who):
c = {"sub": who, "email": who, "roles": [], "permissions": []}
with httpx.Client(transport=_InProcessTransport(app), base_url="http://in-process") as client:
return client.get(path, headers=gateway_headers(c))
def test_through_a_tier_1_seam_the_tree_is_shared_by_caller_and_rows_stay_scoped(engine):
register_local("zzord", ORD.app())
try:
app = CALL.app()
with count_statements(engine) as n:
r = _ask(app, "/twice/o2", "ann@co.example") # ann may not see bob's order
assert r.status_code == 200, r.text
assert r.json() == {"own": [200, 200], "as_person": 404}
# the brick's own grant read o2 ONCE for both asks; ann's ask read it again, as ann, and was refused
assert n.statements == 4, n.sql
with count_statements(engine) as m:
r2 = _ask(app, "/twice/o2", "bob@co.example") # a new request: nothing carried over
assert r2.json() == {"own": [200, 200], "as_person": 200} and m.statements == 4
finally:
unregister_local("zzord")
def test_memo_reads_once_per_caller_and_a_write_to_its_table_drops_it(engine):
calls = []
def read():
calls.append(1)
with engine.connect() as c:
return [dict(r._mapping) for r in c.execute(text("SELECT order_id, note FROM zz_order ORDER BY order_id"))]
m = lambda: repository.memo(engine.connect(), "all", ("zz_order",), read) # noqa: E731
assert m() and len(calls) == 1 # outside a request: no memo
m()
assert len(calls) == 2
with request_scope({"sub": "ann@co.example"}):
first = m()
first[0]["note"] = "edited" # a copy: never what the next answer says
assert m()[0]["note"] == "new" and len(calls) == 3
with request_scope({"sub": "bob@co.example"}):
m()
assert len(calls) == 4
with engine.begin() as c:
c.execute(text("UPDATE zz_order SET note = 'x' WHERE order_id = 'o2'"))
assert m()[1]["note"] == "x" and len(calls) == 5
def test_memo_is_per_engine_and_never_keeps_a_read_after_an_own_write(engine, tmp_path):
other = create_engine(f"sqlite:///{tmp_path / 'other.db'}")
asked = []
def read(name):
asked.append(name)
return name
with request_scope({"sub": "ann@co.example"}):
with engine.connect() as a, other.connect() as b:
assert repository.memo(a, "settings", ("x",), lambda: read("a")) == "a"
assert repository.memo(b, "settings", ("x",), lambda: read("b")) == "b" # same caller, same key, another brick
with engine.connect() as c:
c.begin()
c.execute(text("UPDATE zz_order SET note = 'n' WHERE order_id = 'o1'"))
repository.memo(c, "mine", ("zz_order",), lambda: read("dirty"))
c.rollback()
with engine.connect() as c:
repository.memo(c, "mine", ("zz_order",), lambda: read("clean"))
assert asked == ["a", "b", "dirty", "clean"]
other.dispose()
@pytest.mark.parametrize("sql", [
"UPDATE zz_line SET qty = '5'",
"-- a comment first\nDELETE FROM zz_line WHERE order_id = 'o1'",
"/* a block comment */ INSERT INTO zz_line (order_id, line_no, qty) VALUES ('o1', 9, '1')",
"COPY zz_line FROM STDIN",
"SELECT * INTO zz_line FROM zz_other",
"UPDATE public.zz_line SET qty = '5'",
'UPDATE "zz_line" SET qty = \'5\'',
])
def test_every_write_form_drops_the_tables_it_names(engine, sql):
with request_scope({"sub": "ann@co.example"}) as tree:
load(engine)
assert tree.entries
repository.note_write(sql)
assert not tree.entries, sql
def test_reads_and_other_tables_drop_nothing(engine):
assert not repository.is_write("SELECT * FROM zz_line") and not repository.is_write("-- UPDATE x\nSELECT 1")
with request_scope({"sub": "ann@co.example"}) as tree:
load(engine)
repository.note_write("UPDATE zz_line_archive SET qty = '1'")
repository.note_write("UPDATE other_zz_line SET qty = '1'")
assert tree.entries
def test_upsert_new_on_a_stored_key_is_a_conflict_and_overwrites_nothing(engine):
with engine.begin() as c:
with pytest.raises(Conflict):
repository.upsert(c, lines, [{"order_id": "o1", "line_no": 1, "qty": "9", "created_at": NOW}],
("order_id", "line_no"), new=True)
assert load(engine)["zz_line"][0]["qty"] == "1"
def test_an_after_response_job_does_not_read_through_the_request_map(engine):
from erp_platform import after
seen = {}
def job():
seen["tree"] = repository.current_tree()
with request_scope({"sub": "ann@co.example"}):
t = after.start([after._Job(job, (), {})])
t.join(5)
assert "tree" in seen and seen["tree"] is None
def test_a_tier_2_command_forgets_the_map_and_a_query_keeps_it(engine):
far = Seam("zzfar", "ZZFAR_BASE_URL", roles=["service"], permissions=[], owner="zz")
far.base = "http://127.0.0.1:9" # nothing listens: unreachable is a reply
with request_scope({"sub": "ann@co.example"}) as tree:
load(engine)
assert not far.get("/x", actor="zz") and tree.entries
assert not far.post("/x", {}, actor="zz") and not tree.entries
FLAKY = Brick("zzflaky", instructions="test")
TRIES: list = []
@FLAKY.use_case(kind="query", route="GET /flaky/{order_id}", permission=None)
def flaky(caller: dict, order_id: str) -> dict:
"""Load an order; the first attempt then fails with a transient error."""
with STATE["engine"].connect() as c:
ORDER.load(c, {"order_id": order_id})
TRIES.append(len(repository.current_tree().entries))
if len(TRIES) == 1:
class Busy(Exception):
sqlstate = "40001"
raise Busy("could not serialize access")
return {"ok": True}
def test_a_retried_attempt_through_the_door_starts_from_an_empty_map(engine, monkeypatch):
from erp_platform import retry
monkeypatch.setattr(retry, "BACKOFF", (0.0,))
TRIES.clear()
with count_statements(engine) as n:
r = _ask(FLAKY.app(), "/flaky/o1", "ann@co.example")
assert r.status_code == 200, r.text
assert TRIES == [1, 1] and n.statements == 4 # read again on the retry, not answered from attempt 1