74 lines
2.6 KiB
Python
74 lines
2.6 KiB
Python
import sqlite3
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from threading import Barrier
|
|
|
|
from src.idempotency import IdempotencyLedger
|
|
|
|
|
|
def test_completed_result_replays_after_ledger_reconstruction(tmp_path):
|
|
database = tmp_path / "idempotency.sqlite3"
|
|
first = IdempotencyLedger(database, ttl_seconds=600, max_entries=256)
|
|
|
|
reservation = first.reserve("create-201", ("issue-create", "stackchain/api", "Ship it"))
|
|
assert reservation.state == "reserved"
|
|
first.complete("create-201", {"number": 42, "title": "Ship it"})
|
|
|
|
reconstructed = IdempotencyLedger(database, ttl_seconds=600, max_entries=256)
|
|
replay = reconstructed.reserve(
|
|
"create-201", ("issue-create", "stackchain/api", "Ship it")
|
|
)
|
|
|
|
assert replay.state == "completed"
|
|
assert replay.response == {"number": 42, "title": "Ship it"}
|
|
assert sqlite3.connect(database).execute(
|
|
"SELECT status FROM idempotency_operations WHERE key = ?", ("create-201",)
|
|
).fetchone() == ("completed",)
|
|
|
|
|
|
def test_reservation_expires_completed_rows_and_enforces_capacity(tmp_path):
|
|
now = [1_000.0]
|
|
ledger = IdempotencyLedger(
|
|
tmp_path / "bounded.sqlite3",
|
|
ttl_seconds=10,
|
|
max_entries=2,
|
|
clock=lambda: now[0],
|
|
)
|
|
assert ledger.reserve("old", ("comment", "old")).state == "reserved"
|
|
ledger.complete("old", {"id": 1})
|
|
now[0] += 11
|
|
|
|
assert ledger.reserve("new-1", ("comment", "new-1")).state == "reserved"
|
|
ledger.complete("new-1", {"id": 2})
|
|
assert ledger.reserve("new-2", ("comment", "new-2")).state == "reserved"
|
|
ledger.complete("new-2", {"id": 3})
|
|
replacement = ledger.reserve("new-3", ("comment", "new-3"))
|
|
|
|
assert replacement.state == "reserved"
|
|
with sqlite3.connect(ledger.path) as connection:
|
|
keys = connection.execute(
|
|
"SELECT key FROM idempotency_operations ORDER BY key"
|
|
).fetchall()
|
|
indexes = connection.execute(
|
|
"PRAGMA index_list(idempotency_operations)"
|
|
).fetchall()
|
|
assert keys == [("new-2",), ("new-3",)]
|
|
assert any(index[1] == "idempotency_completed_at_idx" for index in indexes)
|
|
|
|
|
|
def test_independent_connections_atomically_reserve_one_operation(tmp_path):
|
|
database = tmp_path / "workers.sqlite3"
|
|
ledgers = [
|
|
IdempotencyLedger(database, ttl_seconds=600, max_entries=256)
|
|
for _ in range(2)
|
|
]
|
|
barrier = Barrier(2)
|
|
|
|
def reserve(ledger):
|
|
barrier.wait()
|
|
return ledger.reserve("worker-key-201", ("review", "stackchain/api", 7)).state
|
|
|
|
with ThreadPoolExecutor(max_workers=2) as executor:
|
|
states = list(executor.map(reserve, ledgers))
|
|
|
|
assert sorted(states) == ["pending", "reserved"]
|