"""Optimistic versions on the repository (erp_platform.repository, `Aggregate(..., version=...)`).
SC-HR-08: two HR admins open the same person and both save — one commits, the other is told 409
"changed since you opened it" and nothing of theirs is written. On SQLite here; the race on DataK3 — where
there is no read-your-writes inside a transaction — is measured by scripts/live_version_check.py.
"""
import base64
import json
import threading
import anyio
import httpx
import httpx2
import pytest
from mcp import ClientSession
from mcp.client.streamable_http import streamable_http_client
from sqlalchemy import Column, Integer, MetaData, String, Table, TIMESTAMP, create_engine, event, select, text
from erp_platform import repository
from erp_platform.bricks import Brick
from erp_platform.repository import Aggregate, Conflict, Part
md = MetaData()
people = Table("hr_person", md,
Column("person_id", String, primary_key=True), Column("name", String),
Column("version", Integer), Column("created_at", TIMESTAMP), Column("updated_at", TIMESTAMP))
slices = Table("hr_pay_slice", md,
Column("person_id", String, primary_key=True), Column("slice_no", Integer, primary_key=True),
Column("amount", String))
PERSON = Aggregate(Part(people, ("person_id",)), Part(slices, ("person_id", "slice_no")), version="version")
PLAIN = Aggregate(Part(Table("crm_note", MetaData(), Column("note_id", String, primary_key=True),
Column("body", String)), ("note_id",)))
@pytest.fixture
def engine(tmp_path):
e = create_engine(f"sqlite:///{tmp_path / 'v.db'}", connect_args={"timeout": 30, "check_same_thread": False})
@event.listens_for(e, "connect")
def _no_pysqlite_begin(dbapi, _):
dbapi.isolation_level = None
@event.listens_for(e, "begin")
def _immediate(conn):
# SQLite serialises writers; IMMEDIATE takes the write lock at BEGIN so concurrent units of work queue
# instead of deadlocking on a lock upgrade (which SQLite answers "database is locked", not a result).
conn.exec_driver_sql("BEGIN IMMEDIATE")
md.create_all(e)
return e
def rows(name="Anna", slices_=((1, "3000.00"),)):
return {"hr_person": [{"person_id": "p1", "name": name}],
"hr_pay_slice": [{"person_id": "p1", "slice_no": n, "amount": a} for n, a in slices_]}
def load(engine):
with engine.connect() as c:
return PERSON.load(c, {"person_id": "p1"})
def save(engine, r, v):
with repository.conflicts(), engine.begin() as c:
return PERSON.save(c, r, expected_version=v)
def test_load_returns_the_version_and_a_save_writes_the_next_one(engine):
assert save(engine, rows(), 0) == 1
got = load(engine)
assert PERSON.version_of(got) == 1 and got["hr_person"][0]["version"] == 1
assert save(engine, rows("Anna B"), 1) == 2
assert PERSON.version_of(load(engine)) == 2
def test_sc_hr_08_a_stale_save_is_a_conflict_and_writes_nothing(engine):
save(engine, rows(), 0)
v = PERSON.version_of(load(engine)) # both admins open version 1
save(engine, rows("Anna B", ((1, "3100.00"),)), v)
with pytest.raises(Conflict, match="changed since you opened it") as e:
save(engine, rows("Anna C", ((1, "9999.00"), (2, "1.00"))), v)
assert (e.value.expected, e.value.stored) == (1, 2)
after = load(engine)
assert after["hr_person"][0]["name"] == "Anna B"
assert [r["amount"] for r in after["hr_pay_slice"]] == ["3100.00"]
assert save(engine, rows("Anna C"), PERSON.version_of(after)) == 3 # after a reload: a normal change
def test_the_claim_alone_refuses_a_racer_the_committed_check_could_not_see(engine):
"""The race DataK3 allows: the root still says 1, but a racer's claim on 2 is already in. The
committed-root check passes; the plain INSERT of (p1, 2) is what refuses — and takes the rows with it."""
save(engine, rows(), 0)
with engine.begin() as c:
c.execute(PERSON.versions.insert().values(person_id="p1", version=2, saved_at=text("CURRENT_TIMESTAMP")))
with pytest.raises(Conflict):
save(engine, rows("Racer loses"), 1)
assert load(engine)["hr_person"][0]["name"] == "Anna"
def test_ten_concurrent_saves_at_one_version_give_one_success(engine):
save(engine, rows(), 0)
outcomes, errors, gate = [], [], threading.Barrier(10)
def go(i):
try:
gate.wait()
save(engine, rows(f"writer {i}"), 1)
outcomes.append("ok")
except Conflict:
outcomes.append("conflict")
except Exception as exc: # noqa: BLE001 — reported below
errors.append(exc)
threads = [threading.Thread(target=go, args=(i,)) for i in range(10)]
[t.start() for t in threads]
[t.join(30) for t in threads]
assert not errors, errors
assert sorted(outcomes) == ["conflict"] * 9 + ["ok"]
with engine.connect() as c:
assert [r[0] for r in c.execute(select(PERSON.versions.c.version))] == [1, 2]
def test_a_claim_lost_at_commit_is_recognised_as_a_conflict():
"""DataK3 can refuse the claim only at COMMIT, outside save(): the error names the versions table."""
class UniqueViolation(Exception):
sqlstate = "23505"
lost = UniqueViolation("duplicate key value violates unique constraint \"hr_person_versions_pkey\"")
assert isinstance(repository.as_conflict(lost), Conflict)
other = UniqueViolation("duplicate key value violates unique constraint \"hr_person_pkey\"")
assert repository.as_conflict(other) is None # any other duplicate stays a bug
with pytest.raises(Conflict):
with repository.conflicts():
raise lost
def test_versioning_is_opt_in_and_a_versioned_save_needs_the_version(engine):
with engine.begin() as c:
with pytest.raises(TypeError, match="pass expected_version"):
PERSON.save(c, rows())
with pytest.raises(TypeError, match="not versioned"):
PLAIN.save(c, {"crm_note": [{"note_id": "n1", "body": "x"}]}, expected_version=0)
assert PLAIN.versions is None
with pytest.raises(ValueError, match="no column"):
Aggregate(Part(Table("zz_t", MetaData(), Column("k", String, primary_key=True)), ("k",)), version="version")
def test_remove_drops_the_claims_so_the_key_can_start_again(engine):
save(engine, rows(), 0)
with engine.begin() as c:
PERSON.remove(c, {"person_id": "p1"})
assert save(engine, rows(), 0) == 1
def test_the_ddl_for_schema_sql_creates_the_table_the_code_uses(tmp_path):
ddl = PERSON.version_ddl()
assert "CREATE TABLE IF NOT EXISTS hr_person_versions" in ddl and "PRIMARY KEY (person_id, version)" in ddl
e = create_engine(f"sqlite:///{tmp_path / 'ddl.db'}")
with e.begin() as c:
c.exec_driver_sql("\n".join(l for l in ddl.splitlines() if not l.startswith("--")))
cols = [r[1] for r in c.exec_driver_sql("PRAGMA table_info(hr_person_versions)")]
assert cols == [c.name for c in PERSON.versions.c]
# ── through the doors: 409 over HTTP, the same words over MCP ───────────────────────────────
HR = Brick("zzhr", instructions="test")
@HR.use_case(kind="command", route="PUT /people/{person_id}", permission=None)
def rename_person(caller: dict, person_id: str, name: str, version: int) -> dict:
"""Rename a person, from the version you opened."""
raise Conflict("hr_person", {"person_id": person_id}, version, version + 1)
@HR.use_case(kind="command", route="POST /people/{person_id}/race")
def lose_at_commit(caller: dict, person_id: str) -> dict:
"""A save whose claim is refused at COMMIT."""
class UniqueViolation(Exception):
sqlstate = "23505"
raise UniqueViolation("duplicate key 'p1\x1f2' in table 'hr_person_versions': a write already holds this key")
WHO = {"X-Dodil-User": json.dumps({"sub": "a@co.example", "email": "a@co.example", "app_roles": []}),
"X-Dodil-User-JWT": "h." + base64.urlsafe_b64encode(b'{"permissions":[]}').decode().rstrip("=") + ".s"}
def test_a_conflict_is_409_through_both_doors():
app = HR.app()
async def go():
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://b") as c:
r1 = await c.put("/people/p1", headers=WHO, json={"name": "B", "version": 1})
r2 = await c.post("/people/p1/race", headers=WHO)
async with app.router.lifespan_context(app):
async with httpx2.AsyncClient(transport=httpx2.ASGITransport(app=app), headers=WHO,
base_url="http://b") as http:
async with streamable_http_client("http://b/mcp", http_client=http) as (r, w):
async with ClientSession(r, w) as s:
await s.initialize()
m = await s.call_tool("rename_person", {"person_id": "p1", "name": "B", "version": 1})
return r1, r2, m
r1, r2, m = anyio.run(go)
assert r1.status_code == 409 and "changed since you opened it" in r1.json()["detail"]
assert r2.status_code == 409 and r2.json()["detail"] == r1.json()["detail"]
assert m.is_error and m.content[0].text.endswith(f"409: {r1.json()['detail']}")