"""0.9.6: a load is one round trip on psycopg, a save with the loaded snapshot writes only the diff.
Run on SQLite always, and on a real Postgres through psycopg 3 when ERP_TEST_PG_URL is set (e.g.
postgresql+psycopg://postgres@localhost:5432/postgres) — the pipeline path, with round trips counted on
the wire by a proxy. The benchmark is a 14-part aggregate, the shape of sales-order."""
import os
from datetime import datetime, timezone
from decimal import Decimal
from urllib.parse import urlparse
import pytest
from sqlalchemy import Column, Integer, MetaData, Numeric, String, TIMESTAMP, Table, create_engine, event, select
from erp_platform.repository import Aggregate, Conflict, Part
from erp_platform.testing import count_statements
PG_URL = os.environ.get("ERP_TEST_PG_URL", "")
NOW = datetime(2026, 10, 24, 12, 0, tzinfo=timezone.utc)
def _tables(md):
root = Table("so_order", md, Column("order_id", String, primary_key=True), Column("customer", String),
Column("version", Integer), Column("created_at", TIMESTAMP(timezone=True)),
Column("updated_at", TIMESTAMP(timezone=True)))
kids = [Table(f"so_part{i}", md, Column("order_id", String, primary_key=True),
Column("line_no", Integer, primary_key=True, autoincrement=False), Column("qty", Numeric(18, 2)),
Column("note", String), Column("created_at", TIMESTAMP(timezone=True)))
for i in range(13)]
return root, kids
md_plain, md_versioned = MetaData(), MetaData()
_r, _k = _tables(md_plain)
PLAIN = Aggregate(Part(_r, ("order_id",)), *(Part(t, ("order_id", "line_no")) for t in _k))
_rv, _kv = _tables(md_versioned)
VERSIONED = Aggregate(Part(_rv, ("order_id",)), *(Part(t, ("order_id", "line_no")) for t in _kv), version="version")
def memory(order_id="o1", lines=3, qty="1.00", customer="ACME"):
rows = {"so_order": [{"order_id": order_id, "customer": customer}]}
for i in range(13):
rows[f"so_part{i}"] = [{"order_id": order_id, "line_no": n, "qty": Decimal(qty), "note": f"p{i}l{n}"}
for n in range(1, lines + 1)]
return rows
def _utc(d):
return d.replace(tzinfo=timezone.utc) if d.tzinfo is None else d.astimezone(timezone.utc)
def _edit(loaded):
"""What a use case hands back to save: the loaded rows as plain dicts (the caller's working copy)."""
return {name: [dict(r) for r in part_rows] for name, part_rows in loaded.items()}
@pytest.fixture(params=["sqlite"] + (["pg"] if PG_URL else []))
def db(request, tmp_path):
proxy = None
if request.param == "sqlite":
e = create_engine(f"sqlite:///{tmp_path / 'diff.db'}")
else:
from pgwire_proxy import CountingProxy
u = urlparse(PG_URL.replace("postgresql+psycopg", "postgresql"))
proxy = CountingProxy(u.hostname, u.port or 5432)
e = create_engine(PG_URL.replace(f"{u.hostname}:{u.port or 5432}", f"127.0.0.1:{proxy.port}"))
for md in (md_plain, md_versioned):
md.drop_all(e)
md.create_all(e)
yield request.param, e, proxy
e.dispose()
if proxy:
proxy.close()
def _wire(proxy, fn):
"""Round trips `fn` made on the wire (None on SQLite)."""
if proxy is None:
fn()
return None
proxy.reset()
fn()
return proxy.round_trips
def test_a_load_carries_a_snapshot_and_is_one_round_trip_on_psycopg(db):
kind, e, proxy = db
with e.begin() as c:
PLAIN.save(c, memory(), now=NOW)
with e.connect() as c:
c.exec_driver_sql("SELECT 1") # the connection is up: count the load alone
out = {}
trips = _wire(proxy, lambda: out.setdefault("l", PLAIN.load(c, {"order_id": "o1"})))
loaded = out["l"]
assert len(loaded["so_part7"]) == 3 and loaded["so_order"][0]["customer"] == "ACME"
assert loaded.snapshot.root_key == ("o1",)
assert isinstance(loaded["so_part0"][0]["qty"], Decimal)
if kind == "pg":
assert trips == 1, trips # 14 SELECTs, one round trip (was 14)
def test_load_many_reads_n_aggregates_with_one_query_per_part(db):
kind, e, _ = db
with e.begin() as c:
for o in ("o1", "o2", "o3"):
PLAIN.save(c, memory(o, lines=o == "o2" and 1 or 2), now=NOW)
with e.connect() as c, count_statements(e) as n:
many = PLAIN.load_many(c, [{"order_id": "o3"}, {"order_id": "o1"}, {"order_id": "nope"}, {"order_id": "o2"}])
assert n.statements == 14 # one per part, not one per part per order
assert list(many) == [("o3",), ("o1",), ("nope",), ("o2",)]
assert many[("nope",)] is None and len(many[("o2",)]["so_part3"]) == 1 and len(many[("o1",)]["so_part3"]) == 2
if kind == "pg":
assert n.round_trips == 1
def _measure(e, proxy, agg, change):
"""(statements, wire round trips) for: load; save unchanged without a snapshot (0.9.5); with one; a one-line change."""
versioned = agg.version is not None
kw = (lambda l: {"expected_version": agg.version_of(l)}) if versioned else (lambda l: {})
with e.begin() as c:
agg.save(c, memory(), now=NOW, **({"expected_version": 0} if versioned else {}))
out = {}
def run(label, fn):
with e.begin() as c:
c.exec_driver_sql("SELECT 1")
with count_statements(e) as n:
trips = _wire(proxy, lambda: fn(c))
out[label] = (n.statements, trips if trips is not None else n.round_trips)
run("load", lambda c: out.setdefault("loaded", agg.load(c, {"order_id": "o1"})))
run("unchanged_old", lambda c: agg.save(c, memory(), now=NOW, # 0.9.5: no snapshot
**({"expected_version": 1} if versioned else {})))
with e.connect() as c:
loaded = agg.load(c, {"order_id": "o1"})
run("unchanged_new", lambda c: agg.save(c, _edit(loaded), now=NOW, snapshot=loaded, **kw(loaded)))
with e.connect() as c:
loaded = agg.load(c, {"order_id": "o1"})
edited = _edit(loaded)
change(edited)
run("one_line", lambda c: agg.save(c, edited, now=NOW, snapshot=loaded, **kw(loaded)))
return out
def _one_line(rows):
rows["so_part5"][1]["qty"] = Decimal("7.00")
def test_benchmark_unversioned_14_parts(db):
kind, e, proxy = db
m = _measure(e, proxy, PLAIN, _one_line)
print("\n14-part unversioned", kind, {k: v for k, v in m.items() if k != "loaded"})
assert m["load"][0] == 14
assert m["unchanged_old"][0] >= 28 # 0.9.5: re-read 14 keys + upsert 14 parts
assert m["unchanged_new"][0] == 0 # nothing changed: nothing sent
assert m["one_line"][0] == 1 # the changed row alone
if kind == "pg":
assert m["load"][1] == 1
with e.connect() as c:
after = PLAIN.load(c, {"order_id": "o1"})
assert after["so_part5"][1]["qty"] == Decimal("7.00") and after["so_part5"][0]["qty"] == Decimal("1.00")
def test_benchmark_versioned_14_parts(db):
kind, e, proxy = db
m = _measure(e, proxy, VERSIONED, _one_line)
print("\n14-part versioned", kind, {k: v for k, v in m.items() if k != "loaded"})
assert m["unchanged_old"][0] >= 30
assert m["unchanged_new"][0] == 3 # check + claim + the root's new version
assert m["one_line"][0] == 4 # ... + the changed row
if kind == "pg":
assert m["one_line"][1] <= 1 # 0.9.13: the check rides in the claim + writes pipeline
with e.connect() as c:
assert VERSIONED.load(c, {"order_id": "o1"})["so_order"][0]["version"] == 4
def test_a_diff_save_deletes_stale_rows_with_one_statement_per_table(db):
_, e, _ = db
with e.begin() as c:
PLAIN.save(c, memory(lines=5), now=NOW)
with e.connect() as c:
loaded = PLAIN.load(c, {"order_id": "o1"})
edited = _edit(loaded)
edited["so_part2"] = [r for r in edited["so_part2"] if r["line_no"] in (1, 4)]
edited["so_part9"] = []
with e.begin() as c, count_statements(e) as n:
PLAIN.save(c, edited, now=NOW, snapshot=loaded)
assert n.statements == 2 and all(s.lstrip().upper().startswith("DELETE") for s in n.sql), n.sql
with e.connect() as c:
after = PLAIN.load(c, {"order_id": "o1"})
assert [r["line_no"] for r in after["so_part2"]] == [1, 4] and after["so_part9"] == []
assert len(after["so_part1"]) == 5
def test_a_diff_save_keeps_creation_columns_and_writes_new_rows(db):
_, e, _ = db
with e.begin() as c:
PLAIN.save(c, memory(lines=1), now=NOW)
with e.connect() as c:
loaded = PLAIN.load(c, {"order_id": "o1"})
edited = _edit(loaded)
edited["so_part0"][0]["note"] = "changed"
edited["so_part0"][0]["created_at"] = datetime(2030, 1, 1, tzinfo=timezone.utc) # never rewritten
edited["so_part1"].append({"order_id": "o1", "line_no": 2, "qty": Decimal("2.00"), "note": "new"})
later = datetime(2026, 10, 25, tzinfo=timezone.utc)
with e.begin() as c:
PLAIN.save(c, edited, now=later, snapshot=loaded)
with e.connect() as c:
after = PLAIN.load(c, {"order_id": "o1"})
first = after["so_part0"][0]
assert first["note"] == "changed" and _utc(first["created_at"]) == NOW
assert after["so_part1"][1]["note"] == "new"
assert _utc(after["so_part1"][1]["created_at"]) == later
def test_editing_the_loaded_dict_in_place_still_diffs_against_what_was_read(db):
_, e, _ = db
with e.begin() as c:
PLAIN.save(c, memory(lines=1), now=NOW)
with e.connect() as c:
loaded = PLAIN.load(c, {"order_id": "o1"})
loaded["so_part3"][0]["note"] = "in place"
with e.begin() as c, count_statements(e) as n:
PLAIN.save(c, loaded, now=NOW, snapshot=loaded)
assert n.statements == 1
with e.connect() as c:
assert PLAIN.load(c, {"order_id": "o1"})["so_part3"][0]["note"] == "in place"
def test_a_stale_snapshot_save_is_a_conflict_and_writes_nothing(db):
_, e, _ = db
with e.begin() as c:
VERSIONED.save(c, memory(), now=NOW, expected_version=0)
with e.connect() as c:
a = VERSIONED.load(c, {"order_id": "o1"})
b = VERSIONED.load(c, {"order_id": "o1"})
ea, eb = _edit(a), _edit(b)
ea["so_part1"][0]["note"], eb["so_part1"][0]["note"] = "A", "B"
with e.begin() as c:
VERSIONED.save(c, ea, now=NOW, snapshot=a, expected_version=VERSIONED.version_of(a))
with pytest.raises(Conflict):
with e.begin() as c:
VERSIONED.save(c, eb, now=NOW, snapshot=b, expected_version=VERSIONED.version_of(b))
with e.connect() as c:
assert VERSIONED.load(c, {"order_id": "o1"})["so_part1"][0]["note"] == "A"
def test_a_lost_claim_inside_the_write_batch_is_a_conflict(db):
"""The race the check cannot see: the claim row exists already (a racer committed it) → Conflict."""
_, e, _ = db
with e.begin() as c:
VERSIONED.save(c, memory(), now=NOW, expected_version=0)
with e.connect() as c:
loaded = VERSIONED.load(c, {"order_id": "o1"})
with e.begin() as c:
c.execute(VERSIONED.versions.insert().values(order_id="o1", version=2, saved_at=NOW))
edited = _edit(loaded)
edited["so_part1"][0]["note"] = "late"
with pytest.raises(Conflict):
with e.begin() as c:
VERSIONED.save(c, edited, now=NOW, snapshot=loaded, expected_version=1)
with e.connect() as c:
assert VERSIONED.load(c, {"order_id": "o1"})["so_part1"][0]["note"] != "late"
def test_a_new_aggregate_saves_without_reading(db):
_, e, _ = db
with e.begin() as c, count_statements(e) as n:
PLAIN.save(c, memory("o9", lines=1), now=NOW, snapshot=PLAIN.nothing_stored({"order_id": "o9"}))
assert n.statements == 14 and not any(s.lstrip().upper().startswith("SELECT") for s in n.sql)
def test_a_snapshot_of_another_root_is_refused(db):
_, e, _ = db
with e.begin() as c:
PLAIN.save(c, memory("o1"), now=NOW)
with e.connect() as c:
loaded = PLAIN.load(c, {"order_id": "o1"})
with pytest.raises(ValueError, match="snapshot"):
with e.begin() as c:
PLAIN.save(c, memory("o2"), now=NOW, snapshot=loaded)
# ── 0.9.7: an unchanged root's version bump on a table with a NOT NULL column ─────────────────
md_strict = MetaData()
_rs = Table("so_strict", md_strict, Column("order_id", String, primary_key=True),
Column("person_key", String, nullable=False), Column("customer", String), Column("version", Integer),
Column("created_at", TIMESTAMP(timezone=True)), Column("updated_at", TIMESTAMP(timezone=True)))
_ks = Table("so_strict_line", md_strict, Column("order_id", String, primary_key=True),
Column("line_no", Integer, primary_key=True, autoincrement=False), Column("note", String, nullable=False))
STRICT = Aggregate(Part(_rs, ("order_id",)), Part(_ks, ("order_id", "line_no")), version="version")
def test_an_unchanged_root_bumps_its_version_on_a_table_with_a_not_null_column(db):
"""0.9.6 bumped an unchanged root with a row naming only the key and the version. Postgres and SQLite check
NOT NULL on the proposed INSERT row before ON CONFLICT turns it into an update, so the save failed. Still 3
statements; every other root column is kept."""
_, e, _ = db
md_strict.drop_all(e)
md_strict.create_all(e)
with e.begin() as c:
STRICT.save(c, {"so_strict": [{"order_id": "o1", "person_key": "pk-1", "customer": "ACME"}],
"so_strict_line": [{"order_id": "o1", "line_no": 1, "note": "a"}]}, now=NOW, expected_version=0)
with e.connect() as c:
loaded = STRICT.load(c, {"order_id": "o1"})
edited = _edit(loaded)
edited["so_strict_line"][0]["note"] = "b" # only a child changes
later = datetime(2026, 10, 25, 9, 0, tzinfo=timezone.utc)
with e.begin() as c, count_statements(e) as n:
assert STRICT.save(c, edited, now=later, snapshot=loaded, expected_version=1) == 2
assert n.statements == 4 # check + claim + root's version + the one line
with e.connect() as c:
again = STRICT.load(c, {"order_id": "o1"})
with e.begin() as c, count_statements(e) as n3:
assert STRICT.save(c, _edit(again), now=later, snapshot=again, expected_version=2) == 3
assert n3.statements == 3 # nothing changed: check + claim + version
with e.connect() as c:
root = STRICT.load(c, {"order_id": "o1"})["so_strict"][0]
assert (root["person_key"], root["customer"], root["version"]) == ("pk-1", "ACME", 3)
assert _utc(root["created_at"]) == NOW and _utc(root["updated_at"]) == later
# ── 0.9.13 review: a remove deletes the claims, so the claim alone cannot refuse a save after it ──────────────
def test_a_save_after_a_concurrent_remove_is_a_conflict_and_resurrects_nothing(db):
kind, e, proxy = db
with e.begin() as c:
VERSIONED.save(c, memory(), now=NOW, expected_version=0)
with e.connect() as c:
loaded = VERSIONED.load(c, {"order_id": "o1"})
with e.begin() as c:
VERSIONED.remove(c, {"order_id": "o1"}) # someone deletes it, claims and all
edited = _edit(loaded)
edited["so_part1"][0]["note"] = "late"
with pytest.raises(Conflict):
with e.begin() as c:
VERSIONED.save(c, edited, now=NOW, snapshot=loaded, expected_version=1)
with e.connect() as c:
assert VERSIONED.load(c, {"order_id": "o1"}) is None
assert c.execute(select(_kv[1])).first() is None # no orphan child came back either
def test_a_save_after_a_remove_and_a_re_create_below_its_version_is_a_conflict(db):
_, e, _ = db
with e.begin() as c:
VERSIONED.save(c, memory(), now=NOW, expected_version=0)
for v in (1, 2): # the order reaches version 3
with e.connect() as c:
got = VERSIONED.load(c, {"order_id": "o1"})
with e.begin() as c:
VERSIONED.save(c, _edit(got), now=NOW, snapshot=got, expected_version=v)
with e.connect() as c:
stale = VERSIONED.load(c, {"order_id": "o1"}) # read at version 3
with e.begin() as c:
VERSIONED.remove(c, {"order_id": "o1"})
VERSIONED.save(c, memory(customer="NEW"), now=NOW, expected_version=0) # re-created: version 1
edited = _edit(stale)
edited["so_order"][0]["customer"] = "STALE"
with pytest.raises(Conflict):
with e.begin() as c:
VERSIONED.save(c, edited, now=NOW, snapshot=stale, expected_version=3) # claim (o1, 4) is free
with e.connect() as c:
assert VERSIONED.load(c, {"order_id": "o1"})["so_order"][0]["customer"] == "NEW"
def test_the_committed_check_rides_in_the_write_pipeline(db):
"""What the fix costs: a versioned one-line save is check + claim + root + line, and ONE round trip on psycopg."""
kind, e, proxy = db
m = _measure(e, proxy, VERSIONED, _one_line)
print("\n0.9.13 versioned one-line save", kind, m["one_line"], "unchanged", m["unchanged_new"])
assert m["one_line"][0] == 4 and m["unchanged_new"][0] == 3
assert m["one_line"][1] == 1 and m["unchanged_new"][1] == 1 # SQLite reports the trips the pg wire would make
def test_a_new_versioned_aggregate_saves_from_nothing_stored(db):
"""Version 0 has no root to check: the claim alone decides (two creators of one key: one 409)."""
_, e, _ = db
with e.begin() as c:
assert VERSIONED.save(c, memory("o7", lines=1), now=NOW, expected_version=0,
snapshot=VERSIONED.nothing_stored({"order_id": "o7"})) == 1
with pytest.raises(Conflict):
with e.begin() as c:
VERSIONED.save(c, memory("o7", lines=1), now=NOW, expected_version=0,
snapshot=VERSIONED.nothing_stored({"order_id": "o7"}))
def test_a_conflict_caught_inside_the_unit_of_work_writes_nothing_when_it_commits(db):
"""The check fails after the claim and writes were queued: the save rolls the transaction back before raising."""
_, e, _ = db
with e.begin() as c:
VERSIONED.save(c, memory(), now=NOW, expected_version=0)
with e.connect() as c:
loaded = VERSIONED.load(c, {"order_id": "o1"})
with e.begin() as c:
VERSIONED.remove(c, {"order_id": "o1"})
edited = _edit(loaded)
edited["so_part1"][0]["note"] = "late"
with e.begin() as c:
try:
VERSIONED.save(c, edited, now=NOW, snapshot=loaded, expected_version=1)
except Conflict:
pass # caught; the block then commits
with e.connect() as c:
assert VERSIONED.load(c, {"order_id": "o1"}) is None
assert c.execute(select(VERSIONED.versions)).first() is None
assert all(c.execute(select(t)).first() is None for t in _kv)