tegwick
c3f1dfbc07
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a05e2e-805b-7042-a750-71f473bceea2
138 lines
4.6 KiB
Python
138 lines
4.6 KiB
Python
import tempfile
|
|
import threading
|
|
import sqlite3
|
|
from pathlib import Path
|
|
|
|
from approval_engine.errors import Conflict
|
|
from approval_engine.store import Engine
|
|
from tests.conftest import FROZEN, approve
|
|
|
|
|
|
def test_second_supersession_loses(engine):
|
|
obj = approve(engine)
|
|
first = engine.supersede(obj.id)
|
|
assert engine.get(obj.id).status == "superseded"
|
|
assert first["successor_id"]
|
|
try:
|
|
engine.supersede(obj.id)
|
|
raise AssertionError("second supersession must conflict")
|
|
except Conflict:
|
|
pass
|
|
claim = engine.claim(obj.id)
|
|
assert claim["valid_now"] is False
|
|
assert claim["reason_code"] == "superseded"
|
|
|
|
|
|
def test_concurrent_supersessions_one_winner():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
path = Path(tmp) / "a.sqlite"
|
|
setup = Engine(path, clock=lambda: FROZEN)
|
|
obj = approve(setup)
|
|
setup.close()
|
|
|
|
winners: list[str] = []
|
|
errors: list[str] = []
|
|
barrier = threading.Barrier(2)
|
|
|
|
def race():
|
|
eng = Engine(path, clock=lambda: FROZEN)
|
|
barrier.wait()
|
|
try:
|
|
result = eng.supersede(obj.id)
|
|
winners.append(result["successor_id"])
|
|
except Conflict as exc:
|
|
errors.append(str(exc))
|
|
finally:
|
|
eng.close()
|
|
|
|
threads = [threading.Thread(target=race) for _ in range(2)]
|
|
for t in threads:
|
|
t.start()
|
|
for t in threads:
|
|
t.join()
|
|
assert len(winners) == 1
|
|
assert len(errors) == 1
|
|
check = Engine(path, clock=lambda: FROZEN)
|
|
assert check.get(obj.id).status == "superseded"
|
|
check.close()
|
|
|
|
|
|
def test_consume_same_digest_is_idempotent(engine):
|
|
obj = approve(engine)
|
|
digest = "sha256:" + "ab" * 32
|
|
first = engine.consume(obj.id, digest, decision_id="decision:first")
|
|
second = engine.consume(obj.id, digest, decision_id="decision:retry")
|
|
assert first["idempotent"] is False
|
|
assert second["idempotent"] is True
|
|
assert second["decision_id"] == "decision:first"
|
|
assert [item["class"] for item in engine.undrained()].count("use") == 1
|
|
claim = engine.claim(obj.id)
|
|
assert claim["consumed"] is True
|
|
assert claim["valid_now"] is False
|
|
assert claim["reason_code"] == "consumed"
|
|
|
|
|
|
def test_consume_different_digest_conflicts(engine):
|
|
obj = approve(engine)
|
|
engine.consume(obj.id, "sha256:" + "ab" * 32)
|
|
try:
|
|
engine.consume(obj.id, "sha256:" + "cd" * 32)
|
|
raise AssertionError("different request digest must conflict")
|
|
except Conflict:
|
|
pass
|
|
|
|
|
|
def test_concurrent_same_digest_consume_is_one_use_event():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
path = Path(tmp) / "consume.sqlite"
|
|
setup = Engine(path, clock=lambda: FROZEN)
|
|
obj = approve(setup)
|
|
setup.close()
|
|
digest = "sha256:" + "ef" * 32
|
|
results: list[bool] = []
|
|
barrier = threading.Barrier(2)
|
|
|
|
def race():
|
|
eng = Engine(path, clock=lambda: FROZEN)
|
|
barrier.wait()
|
|
try:
|
|
results.append(eng.consume(obj.id, digest)["idempotent"])
|
|
finally:
|
|
eng.close()
|
|
|
|
threads = [threading.Thread(target=race) for _ in range(2)]
|
|
for thread in threads:
|
|
thread.start()
|
|
for thread in threads:
|
|
thread.join()
|
|
assert sorted(results) == [False, True]
|
|
check = Engine(path, clock=lambda: FROZEN)
|
|
assert [item["class"] for item in check.undrained()].count("use") == 1
|
|
check.close()
|
|
|
|
|
|
def test_existing_database_migrates_consumption_columns():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
path = Path(tmp) / "legacy.sqlite"
|
|
conn = sqlite3.connect(path)
|
|
conn.execute(
|
|
"""CREATE TABLE approvals (
|
|
id TEXT PRIMARY KEY, status TEXT NOT NULL,
|
|
binding_json TEXT NOT NULL, binding_digest TEXT NOT NULL,
|
|
pdp_digest TEXT, actor TEXT NOT NULL, principal TEXT NOT NULL,
|
|
action TEXT NOT NULL, purpose TEXT NOT NULL,
|
|
target_json TEXT NOT NULL, not_before TEXT NOT NULL,
|
|
expires_at TEXT NOT NULL, required_count INTEGER NOT NULL,
|
|
superseded_by TEXT, created_at TEXT NOT NULL,
|
|
updated_at TEXT NOT NULL
|
|
)"""
|
|
)
|
|
conn.commit()
|
|
conn.close()
|
|
eng = Engine(path, clock=lambda: FROZEN)
|
|
columns = {
|
|
row["name"]
|
|
for row in eng._conn().execute("PRAGMA table_info(approvals)").fetchall()
|
|
}
|
|
assert {"consumed_digest", "consumed_decision_id", "consumed_at"} <= columns
|
|
eng.close()
|