dodil/erp-platform

tests/test_hardening.py

"""0.5 — what the CRM audit (2026-10-17) found every brick needed, written once here.

* 64-bit ids leave as strings (opt-in) and come in as a string or a number (types.BigId);
* bounded input types: a caller's mistake is a 422 naming the field, never the engine's 500;
* control characters refused in every string, through both doors;
* a body past the limit is 413;
* If-Match fills an optimistic version, and a query can send its version as the ETag;
* Idempotency-Key (header) / idempotency_key (MCP argument) on the commands that make something new;
* a route that is no agent's tool (tool=False);
* a crash through MCP says 500 and carries the request id;
* a seam that asks AS THE CALLER, in process and across apps;
* repository.retry_on_conflict: a change sent with no version is re-run on the latest, never lost.

Each property has a test that fails when it is broken (checked by putting the defect back).
"""
import base64
import json
import socket
import threading
import time
from concurrent.futures import ThreadPoolExecutor
from decimal import Decimal
from typing import Optional

import anyio
import httpx
import pytest
import uvicorn
from fastapi import FastAPI, Request
from mcp import ClientSession
from mcp.client.streamable_http import create_mcp_http_client, streamable_http_client
from sqlalchemy import Column, Integer, MetaData, String, Table, create_engine, event, select

from erp_platform import auth, idempotency, repository, seam
from erp_platform.bricks import Brick, BodyLimit, Refused, ids_as_text
from erp_platform.repository import Aggregate, Part
from erp_platform.types import Amount, BigId, IsoDate, Name, Quantity

# ── a small brick that uses every 0.5 feature ──────────────────────────────────────────────

NOTES: dict = {}
WRITES: list = []

md = MetaData()
cards = Table("demo_cards", md, Column("card_id", String, primary_key=True), Column("title", String),
              Column("body", String), Column("version", Integer))
CARD = Aggregate(Part(cards, ("card_id",)), version="version")

DEMO = Brick("demo", ids_as_strings=True, id_keys=("parents",))
PLAIN = Brick("plain")


@DEMO.use_case(kind="query", route="GET /partners/{bp_id}")
def get_partner(caller: dict, bp_id: BigId) -> dict:
    """One partner by its 64-bit id."""
    return {"bp_id": bp_id, "parents": [bp_id - 1], "family": [{"bp_id": bp_id + 1, "rank": 2}],
            "count": 3, "is_id": True}


@DEMO.use_case(kind="command", route="POST /bounded", idempotent=True)
def bounded(caller: dict, name: Name, when: IsoDate, amount: Amount, quantity: Quantity = 1,
            bp_id: Optional[BigId] = None) -> dict:
    """Echo bounded inputs."""
    return {"name": name, "when": when, "amount": amount, "quantity": quantity, "bp_id": bp_id}


@DEMO.use_case(kind="command", route="POST /notes", permission="demo:note:write")
def add_note(caller: dict, text: str, tags: Optional[list[str]] = None) -> dict:
    """Make a new note — a command that makes something new on every call."""
    if text == "refuse":
        raise Refused(409, "not now")
    n = f"N-{len(NOTES) + 1}"
    NOTES[n] = {"text": text, "by": caller["email"]}
    WRITES.append(n)
    time.sleep(0.05)
    return {"note_id": n, "text": text}


@DEMO.use_case(kind="query", route="GET /search")
def search(caller: dict, q: Optional[Name] = None, limit: int = 10) -> dict:
    """Search with query parameters."""
    return {"q": q, "limit": limit}


@DEMO.use_case(kind="query", route="GET /cards/{card_id}", etag="version")
def get_card(caller: dict, card_id: str) -> dict:
    """One card and its version."""
    return {"card_id": card_id, "version": 7}


@DEMO.use_case(kind="command", route="PATCH /cards/{card_id}", idempotent=True, if_match="version")
def edit_card(caller: dict, card_id: str, title: str, version: Optional[int] = None) -> dict:
    """Edit a card at the version the caller loaded."""
    return {"card_id": card_id, "title": title, "version_seen": version}


@DEMO.use_case(kind="command", route="PATCH /cards/{card_id}/archive", idempotent=True, if_match="version")
def archive_card(caller: dict, card_id: str, reason: str = "tidy", version: Optional[int] = None) -> dict:
    """Archive a card; every body field is optional."""
    return {"card_id": card_id, "reason": reason, "version_seen": version}


@DEMO.use_case(kind="command", route="POST /machine/link", permission="demo:link:machine", tool=False)
def machine_link(caller: dict, ref: str) -> dict:
    """A route only a neighbouring brick calls."""
    return {"ref": ref}


@DEMO.use_case(kind="query", route="GET /boom")
def boom(caller: dict) -> dict:
    """Crash, to show what an agent is told."""
    raise RuntimeError("secret value 4111-1111 in the message")


@PLAIN.use_case(kind="query", route="GET /partners/{bp_id}")
def plain_partner(caller: dict, bp_id: int) -> dict:
    """Not opted in: ids stay numbers."""
    return {"bp_id": bp_id}


def as_user(email, roles, permissions):
    payload = base64.urlsafe_b64encode(json.dumps({"permissions": permissions}).encode()).decode()
    return {"X-Dodil-User": json.dumps({"sub": email, "email": email, "app_roles": roles}),
            "X-Dodil-User-JWT": f"h.{payload.rstrip('=')}.s"}


ANNA = as_user("anna@co.example", ["sales"], ["demo:note:write"])
BEN = as_user("ben@co.example", ["sales"], ["demo:note:write"])
BIG = 45665172796086509                                    # past 2**53: a double reads it as …512


def _serve(app):
    with socket.socket() as s:
        s.bind(("127.0.0.1", 0))
        port = s.getsockname()[1]
    server = uvicorn.Server(uvicorn.Config(app, host="127.0.0.1", port=port, log_level="critical"))
    t = threading.Thread(target=server.run, daemon=True)
    t.start()
    deadline = time.time() + 10
    while not server.started and time.time() < deadline:
        time.sleep(0.02)
    assert server.started
    return server, t, f"http://127.0.0.1:{port}"


@pytest.fixture(scope="module")
def base():
    server, t, url = _serve(DEMO.app())
    yield url
    server.should_exit = True
    t.join(5)


@pytest.fixture
def store(tmp_path):
    e = create_engine(f"sqlite:///{tmp_path / 'idem.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):
        conn.exec_driver_sql("BEGIN IMMEDIATE")
    idempotency.md.create_all(e)
    DEMO.idempotency_connection = e.begin
    NOTES.clear()
    WRITES.clear()
    yield e
    DEMO.idempotency_connection = None


async def _mcp(base, headers, tool, args, list_tools=False):
    async with create_mcp_http_client(headers=headers) as http:
        async with streamable_http_client(f"{base}/mcp", http_client=http) as (r, w):
            async with ClientSession(r, w) as s:
                await s.initialize()
                if list_tools:
                    return (await s.list_tools()).tools
                res = await s.call_tool(tool, args)
    text = res.content[0].text if res.content else ""
    if res.is_error:
        return "refused", text.split(f"Error executing tool {tool}: ", 1)[-1]
    return "ok", json.loads(text)


def mcp(base, headers, tool, args=None):
    return anyio.run(_mcp, base, headers, tool, args or {})


def tools(base):
    return {t.name: t for t in anyio.run(_mcp, base, ANNA, "", {}, True)}


# ── 64-bit ids ─────────────────────────────────────────────────────────────────────────────

def test_ids_leave_as_strings_through_both_doors(base):
    r = httpx.get(f"{base}/partners/{BIG}", headers=ANNA)
    assert r.status_code == 200 and r.json() == {
        "bp_id": str(BIG), "parents": [str(BIG - 1)], "family": [{"bp_id": str(BIG + 1), "rank": 2}],
        "count": 3, "is_id": True}
    assert f'"bp_id":"{BIG}"' in r.text.replace(" ", "")        # on the wire, not just after parsing
    assert mcp(base, ANNA, "get_partner", {"bp_id": str(BIG)}) == ("ok", r.json())


def test_an_id_is_accepted_as_digits_or_as_a_number(base):
    for given in (str(BIG), BIG):
        r = httpx.post(f"{base}/bounded", headers=ANNA,
                       json={"name": "x", "when": "2026-10-17", "amount": "1.00", "bp_id": given})
        assert r.status_code == 200 and r.json()["bp_id"] == str(BIG)
        assert mcp(base, ANNA, "bounded", {"name": "x", "when": "2026-10-17", "amount": "1.00",
                                           "bp_id": given})[1]["bp_id"] == str(BIG)


def test_an_id_that_is_not_a_64_bit_integer_is_422_not_500(base):
    for bad in ("99999999999999999999999", "12a", "1.5", True):
        r = httpx.post(f"{base}/bounded", headers=ANNA,
                       json={"name": "x", "when": "2026-10-17", "amount": "1", "bp_id": bad})
        assert r.status_code == 422, (bad, r.status_code)
        status, reason = mcp(base, ANNA, "bounded", {"name": "x", "when": "2026-10-17", "amount": "1",
                                                     "bp_id": bad})
        assert status == "refused" and "bp_id" in reason, (bad, reason)
    assert httpx.get(f"{base}/partners/99999999999999999999999", headers=ANNA).status_code == 422


def test_the_published_schemas_say_an_id_is_a_string_of_digits(base):
    spec = httpx.get(f"{base}/openapi.json").json()
    param = spec["paths"]["/partners/{bp_id}"]["get"]["parameters"][0]["schema"]
    assert param["type"] == "string" and param["format"] == "int64"
    schema = tools(base)["get_partner"].input_schema["properties"]["bp_id"]
    assert schema["type"] == "string" and schema["format"] == "int64"


async def _asgi(app, method, path, **kw):
    async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://b") as c:
        return await c.request(method, path, **kw)


def asgi(app, method, path, **kw):
    return anyio.run(lambda: _asgi(app, method, path, **kw))


def test_a_brick_that_did_not_opt_in_keeps_numbers():
    assert asgi(PLAIN.routes_app(), "GET", "/partners/12", headers=ANNA).json() == {"bp_id": 12}


def test_ids_as_text_leaves_everything_else_alone():
    assert ids_as_text({"line_no": 3, "version": 2, "quote_id": "Q-1", "contact_id": None,
                        "score": 80, "amount": "1.00", "ok": True}) == {
        "line_no": 3, "version": 2, "quote_id": "Q-1", "contact_id": None, "score": 80, "amount": "1.00",
        "ok": True}


# ── bounded inputs ─────────────────────────────────────────────────────────────────────────

@pytest.mark.parametrize("field,value", [
    ("name", "L" * 201), ("when", "not-a-date"), ("when", "2026-02-30"), ("when", "1066-10-14"),
    ("amount", "1e20"), ("amount", "99999999999999999.99"), ("amount", "NaN"), ("amount", "1.0001"),
    ("quantity", 2 ** 31), ("quantity", 0)])
def test_an_out_of_bounds_input_is_422_through_both_doors(base, field, value):
    body = {"name": "x", "when": "2026-10-17", "amount": "1.00", field: value}
    r = httpx.post(f"{base}/bounded", headers=ANNA, json=body)
    assert r.status_code == 422 and field in json.dumps(r.json()), r.text
    status, reason = mcp(base, ANNA, "bounded", body)
    assert status == "refused" and field in reason, reason


def test_bounded_inputs_in_range_are_served(base):
    r = httpx.post(f"{base}/bounded", headers=ANNA,
                   json={"name": "Ünïcødé شركة", "when": "2026-10-17T10:00:00Z", "amount": "12.345", "quantity": 1000000})
    assert r.status_code == 200 and r.json()["when"] == "2026-10-17" and r.json()["amount"] == "12.345"


# ── control characters ─────────────────────────────────────────────────────────────────────

def test_a_control_character_is_refused_with_the_field_through_both_doors(base, store):
    for text, tags in (("nul\x00byte", None), ("bell\x07", None), ("ok", ["fine", "esc\x1b[2J"])):
        r = httpx.post(f"{base}/notes", headers=ANNA, json={"text": text, "tags": tags})
        assert r.status_code == 422 and "control characters" in r.json()["detail"], r.text
        assert r.json()["detail"].startswith("tags.1:" if tags else "text:")
        assert mcp(base, ANNA, "add_note", {"text": text, "tags": tags})[0] == "refused"
    assert WRITES == []


def test_tabs_and_newlines_are_text(base, store):
    r = httpx.post(f"{base}/notes", headers=ANNA, json={"text": "line one\r\n\tline two"})
    assert r.status_code == 200


# ── body size ──────────────────────────────────────────────────────────────────────────────

def test_a_body_past_the_limit_is_413_with_a_reason():
    app = FastAPI()

    @app.post("/x")
    async def x(request: Request):
        return {"bytes": len(await request.body())}
    limited = BodyLimit(app, limit=1024)
    assert asgi(limited, "POST", "/x", content=b"a" * 1000).json() == {"bytes": 1000}
    r = asgi(limited, "POST", "/x", content=b"a" * 5000)
    assert r.status_code == 413 and "1 KiB" in r.json()["detail"]

    async def chunks():                                # no content-length: counted as it arrives
        for _ in range(10):
            yield b"a" * 500
    assert asgi(limited, "POST", "/x", content=chunks()).status_code == 413


def test_the_served_app_has_the_limit(base):
    r = httpx.post(f"{base}/notes", headers={**ANNA, "content-type": "application/json"},
                   content=b'{"text":"' + b"a" * (6 * 1024 * 1024) + b'"}')
    assert r.status_code == 413


# ── If-Match and ETag ──────────────────────────────────────────────────────────────────────

def test_if_match_fills_the_version_and_the_etag_carries_it(base):
    r = httpx.get(f"{base}/cards/c1", headers=ANNA)
    assert r.headers["etag"] == '"7"'
    for header in ('"7"', 'W/"7"', "7"):
        r = httpx.patch(f"{base}/cards/c1", headers={**ANNA, "If-Match": header}, json={"title": "t"})
        assert r.status_code == 200 and r.json()["version_seen"] == 7
    assert httpx.patch(f"{base}/cards/c1", headers=ANNA, json={"title": "t", "version": 3}).json()["version_seen"] == 3
    assert httpx.patch(f"{base}/cards/c1", headers=ANNA, json={"title": "t"}).json()["version_seen"] is None
    r = httpx.patch(f"{base}/cards/c1", headers={**ANNA, "If-Match": '"7"'}, json={"title": "t", "version": 3})
    assert r.status_code == 422 and "If-Match" in r.json()["detail"]
    r = httpx.patch(f"{base}/cards/c1", headers={**ANNA, "If-Match": "*"}, json={"title": "t"})
    assert r.status_code == 422


def test_a_body_of_optional_fields_may_be_left_out(base):
    """DELETE /partners/{id}/roles/{role} with an optional version: no body is a fine request, not a 422."""
    r = httpx.patch(f"{base}/cards/c1/archive", headers=ANNA)
    assert r.status_code == 200 and r.json() == {"card_id": "c1", "reason": "tidy", "version_seen": None}
    r = httpx.patch(f"{base}/cards/c1/archive", headers={**ANNA, "If-Match": '"4"'})
    assert r.json()["version_seen"] == 4
    assert httpx.patch(f"{base}/cards/c1", headers=ANNA).status_code == 422     # title is required: still 422


def test_query_parameters_reach_the_use_case(base):
    assert httpx.get(f"{base}/search", headers=ANNA, params={"q": "acme", "limit": 3}).json() == {"q": "acme", "limit": 3}
    assert httpx.get(f"{base}/search", headers=ANNA).json() == {"q": None, "limit": 10}


def test_if_match_must_name_an_input():
    def edit(caller: dict, y_id: str) -> dict:
        """Edit a y."""
    with pytest.raises(TypeError, match="if_match"):
        Brick("x").use_case(kind="command", route="PATCH /y/{y_id}", if_match="version")(edit)


# ── Idempotency-Key ────────────────────────────────────────────────────────────────────────

def test_a_repeated_key_returns_the_first_answer_and_writes_once(base, store):
    h = {**ANNA, "Idempotency-Key": "k-1"}
    first = httpx.post(f"{base}/notes", headers=h, json={"text": "hello"})
    again = httpx.post(f"{base}/notes", headers=h, json={"text": "hello"})
    assert first.status_code == again.status_code == 200 and first.json() == again.json()
    assert again.headers.get("idempotent-replayed") == "true" and "idempotent-replayed" not in first.headers
    assert WRITES == ["N-1"]
    # through the agent door, with the argument: the same rule
    one = mcp(base, ANNA, "add_note", {"text": "via mcp", "idempotency_key": "k-2"})
    two = mcp(base, ANNA, "add_note", {"text": "via mcp", "idempotency_key": "k-2"})
    assert one == two and WRITES == ["N-1", "N-2"]
    # without a key, nothing changes: each call is a new note
    httpx.post(f"{base}/notes", headers=ANNA, json={"text": "hello"})
    assert len(WRITES) == 3


def test_the_same_key_with_another_request_is_422(base, store):
    h = {**ANNA, "Idempotency-Key": "k-3"}
    assert httpx.post(f"{base}/notes", headers=h, json={"text": "a"}).status_code == 200
    r = httpx.post(f"{base}/notes", headers=h, json={"text": "b"})
    assert r.status_code == 422 and "different request" in r.json()["detail"]
    assert WRITES == ["N-1"]


def test_a_key_is_per_caller(base, store):
    httpx.post(f"{base}/notes", headers={**ANNA, "Idempotency-Key": "shared"}, json={"text": "a"})
    r = httpx.post(f"{base}/notes", headers={**BEN, "Idempotency-Key": "shared"}, json={"text": "a"})
    assert r.status_code == 200 and r.json()["note_id"] == "N-2"


def test_a_refusal_is_not_stored_so_the_key_can_try_again(base, store):
    h = {**ANNA, "Idempotency-Key": "k-4"}
    assert httpx.post(f"{base}/notes", headers=h, json={"text": "refuse"}).status_code == 409
    with store.connect() as c:
        assert c.execute(select(idempotency.claims)).fetchall() == []
    assert httpx.post(f"{base}/notes", headers=h, json={"text": "refuse"}).status_code == 409


def test_ten_concurrent_repeats_of_one_key_write_once(base, store):
    h = {**ANNA, "Idempotency-Key": "k-race"}
    with ThreadPoolExecutor(10) as ex:
        rs = list(ex.map(lambda _: httpx.post(f"{base}/notes", headers=h, json={"text": "race"}, timeout=30),
                         range(10)))
    codes = sorted(r.status_code for r in rs)
    assert WRITES == ["N-1"], WRITES
    assert set(codes) <= {200, 409} and codes.count(200) >= 1
    assert all(r.json()["note_id"] == "N-1" for r in rs if r.status_code == 200)
    assert all("still being processed" in r.json()["detail"] for r in rs if r.status_code == 409)


def test_the_insert_alone_lets_one_racer_through_when_none_saw_the_others(base, store, monkeypatch):
    """On DataK3 every racer can read "no claim yet" from its snapshot: the plain INSERT is then the only
    guard. Simulated by making every read see nothing — exactly one request may run."""
    monkeypatch.setattr(idempotency.Store, "rows", lambda self, *a: [])
    h = {**ANNA, "Idempotency-Key": "k-blind"}
    with ThreadPoolExecutor(8) as ex:
        rs = list(ex.map(lambda _: httpx.post(f"{base}/notes", headers=h, json={"text": "blind"}, timeout=30),
                         range(8)))
    assert WRITES == ["N-1"], WRITES
    assert sorted(r.status_code for r in rs) == [200] + [409] * 7


def test_a_malformed_key_is_422(base, store):
    r = httpx.post(f"{base}/notes", headers={**ANNA, "Idempotency-Key": "has spaces"}, json={"text": "a"})
    assert r.status_code == 422 and WRITES == []


def test_only_commands_that_make_something_new_take_a_key(base):
    t = tools(base)
    assert "idempotency_key" in t["add_note"].input_schema["properties"]
    assert "idempotency_key" not in t["edit_card"].input_schema["properties"]      # idempotent already
    assert "idempotency_key" not in t["get_card"].input_schema["properties"]
    spec = httpx.get(f"{base}/openapi.json").json()
    names = [p["name"] for p in spec["paths"]["/notes"]["post"]["parameters"]]
    assert "Idempotency-Key" in names


# ── a route that is no agent's tool ────────────────────────────────────────────────────────

def test_tool_false_is_a_route_and_not_a_tool(base):
    assert "machine_link" not in tools(base)
    r = httpx.post(f"{base}/machine/link", headers=ANNA, json={"ref": "x"})
    assert r.status_code == 403 and r.json()["detail"] == "missing permission: demo:link:machine"
    ok = httpx.post(f"{base}/machine/link", headers=as_user("svc", ["service"], ["demo:link:machine"]),
                    json={"ref": "x"})
    assert ok.json() == {"ref": "x"}


# ── a crash through MCP ────────────────────────────────────────────────────────────────────

def test_a_crash_through_mcp_says_500_and_the_request_id_not_the_message(base):
    status, reason = mcp(base, ANNA, "boom")
    assert status == "refused" and reason.startswith("500: internal error (request id ")
    assert "4111" not in reason
    r = httpx.get(f"{base}/boom", headers=ANNA)
    assert r.status_code == 500 and r.json()["detail"] == "internal error"


# ── a seam that asks as the caller ─────────────────────────────────────────────────────────

deals = FastAPI()
OWNED = {"O-1": "anna@co.example", "O-2": "ben@co.example"}


@deals.get("/opportunities/{oid}")
def read_deal(oid: str, request: Request):
    who = auth.identity_from_headers(request.headers)
    scope = auth.owner_scope(who)
    if oid not in OWNED or (scope is not None and OWNED[oid] != scope):
        from fastapi import HTTPException
        raise HTTPException(404, f"no opportunity {oid}")
    return {"opportunity_id": oid, "as": who["email"], "permissions": who["permissions"],
            "acting_service": who.get("acting_service")}


ANNA_ID = auth.identity_from_headers(ANNA)
MANAGER_ID = auth.identity_from_headers(as_user("max@co.example", ["manager"], ["demo:note:write"]))


@pytest.fixture
def pipeline_seam(monkeypatch):
    yield Seam_("pipeline")
    seam.unregister_local("pipeline")


def Seam_(name):
    return seam.Seam(name, "PIPELINE_BASE_URL", roles=["service"], permissions=["quote:opportunity:read"])


def test_in_process_the_neighbour_sees_exactly_the_caller(monkeypatch, pipeline_seam):
    monkeypatch.delenv("PIPELINE_BASE_URL", raising=False)
    seam.register_local("pipeline", deals)
    assert pipeline_seam.get("/opportunities/O-1", caller=ANNA_ID).data["as"] == "anna@co.example"
    assert pipeline_seam.get("/opportunities/O-2", caller=ANNA_ID).status == 404       # not hers
    mgr = pipeline_seam.get("/opportunities/O-2", caller=MANAGER_ID)
    assert mgr.ok and mgr.data["permissions"] == ["demo:note:write"]                    # the person's, not the seam's
    # without caller=, the seam's own service identity (scoped to the actor's email, the old behaviour)
    assert pipeline_seam.get("/opportunities/O-2", actor="max@co.example").status == 404


def test_across_apps_acting_for_is_honoured_only_from_a_service_that_may(monkeypatch, pipeline_seam):
    server, t, url = _serve(deals)
    try:
        monkeypatch.setenv("PIPELINE_BASE_URL", url)
        s = Seam_("pipeline")
        # the local stand-in for the gateway: the seam's fabricated headers carry its own permissions
        s.permissions = ["quote:opportunity:read", seam.ACT_FOR]
        r = s.get("/opportunities/O-2", caller=MANAGER_ID)
        assert r.ok and r.data["as"] == "max@co.example" and r.data["permissions"] == []
        assert r.data["acting_service"]                         # who really asked stays on the identity
        assert s.get("/opportunities/O-2", caller=ANNA_ID).status == 404
        s.permissions = ["quote:opportunity:read"]               # no act_for: the header is ignored
        assert s.get("/opportunities/O-1", caller=MANAGER_ID).status == 404
        # a PERSON sending the header borrows nobody's eyes
        forged = {**as_user("ben@co.example", ["sales"], []),
                  seam.ACTING_FOR_HEADER: json.dumps({"sub": "max", "email": "max@co.example", "app_roles": ["manager"]})}
        assert httpx.get(f"{url}/opportunities/O-1", headers=forged).status_code == 404
    finally:
        server.should_exit = True
        t.join(5)


# ── a change with no version is never lost ─────────────────────────────────────────────────

@pytest.fixture
def cards_db(tmp_path):
    e = create_engine(f"sqlite:///{tmp_path / 'cards.db'}")
    md.create_all(e)
    with e.begin() as c:
        CARD.save(c, {"demo_cards": [{"card_id": "c1", "title": "t", "body": ""}]}, expected_version=0)
    return e


def test_a_stale_save_with_no_version_is_rerun_on_the_latest(cards_db):
    """Two people change one card at once, neither naming a version: both changes land. The stale load is
    what DataK3's snapshot gives the second of two concurrent requests."""
    with cards_db.connect() as c:
        stale = CARD.load(c, {"card_id": "c1"})
    with cards_db.begin() as c:                                         # the other person's edit commits
        now = CARD.load(c, {"card_id": "c1"})
        CARD.save(c, {"demo_cards": [{**now["demo_cards"][0], "title": "theirs"}]}, expected_version=1)
    loads = iter([stale])

    def edit_body():
        with cards_db.begin() as c:
            got = next(loads, None) or CARD.load(c, {"card_id": "c1"})
            row = got["demo_cards"][0]
            CARD.save(c, {"demo_cards": [{**row, "body": "mine"}]}, expected_version=CARD.version_of(got))
    repository.retry_on_conflict(edit_body, sleep=lambda _s: None)
    with cards_db.connect() as c:
        row = CARD.load(c, {"card_id": "c1"})["demo_cards"][0]
    assert (row["title"], row["body"], row["version"]) == ("theirs", "mine", 3)


def test_retry_on_conflict_gives_up_and_never_retries_anything_else(cards_db):
    calls = []

    def always_stale():
        calls.append(1)
        raise repository.Conflict()
    with pytest.raises(repository.Conflict):
        repository.retry_on_conflict(always_stale, attempts=3, sleep=lambda _s: None)
    assert len(calls) == 3
    calls.clear()

    def other():
        calls.append(1)
        raise ValueError("a bug")
    with pytest.raises(ValueError):
        repository.retry_on_conflict(other, sleep=lambda _s: None)
    assert len(calls) == 1