217 lines
7.8 KiB
Python
217 lines
7.8 KiB
Python
import sqlite3
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from threading import Barrier
|
|
|
|
import pytest
|
|
|
|
from src.idempotency import IdempotencyLedger
|
|
from src.state_encryption import PrivateStateEncryptionError
|
|
|
|
|
|
PRIVATE_KEY = b"i" * 32
|
|
|
|
|
|
def test_authored_payloads_are_encrypted_at_rest_and_replay_after_restart(tmp_path):
|
|
database = tmp_path / "idempotency.sqlite3"
|
|
fingerprint_canary = "private-issue-body-canary"
|
|
response_canary = "private-gitea-response-canary"
|
|
ledger = IdempotencyLedger(
|
|
database,
|
|
ttl_seconds=600,
|
|
max_entries=256,
|
|
encryption_key=PRIVATE_KEY,
|
|
)
|
|
|
|
assert ledger.reserve("create-201", ("issue-create", fingerprint_canary)).state == "reserved"
|
|
ledger.complete("create-201", {"title": response_canary})
|
|
|
|
persisted = database.read_bytes()
|
|
assert fingerprint_canary.encode() not in persisted
|
|
assert response_canary.encode() not in persisted
|
|
with sqlite3.connect(database) as connection:
|
|
fingerprint, response = connection.execute(
|
|
"SELECT fingerprint, response_json FROM idempotency_operations WHERE key = ?",
|
|
("create-201",),
|
|
).fetchone()
|
|
assert fingerprint.startswith("v1:")
|
|
assert response.startswith("v1:")
|
|
|
|
reopened = IdempotencyLedger(
|
|
database,
|
|
ttl_seconds=600,
|
|
max_entries=256,
|
|
encryption_key=PRIVATE_KEY,
|
|
)
|
|
replay = reopened.reserve("create-201", ("issue-create", fingerprint_canary))
|
|
assert replay.state == "completed"
|
|
assert replay.response == {"title": response_canary}
|
|
|
|
|
|
def test_legacy_plaintext_completed_row_is_migrated_without_losing_replay(tmp_path):
|
|
database = tmp_path / "legacy.sqlite3"
|
|
ledger = IdempotencyLedger(
|
|
database,
|
|
ttl_seconds=600,
|
|
max_entries=256,
|
|
encryption_key=PRIVATE_KEY,
|
|
clock=lambda: 100.0,
|
|
)
|
|
with sqlite3.connect(database) as connection:
|
|
connection.execute(
|
|
"INSERT INTO idempotency_operations "
|
|
"(key, fingerprint, status, response_json, created_at, completed_at) "
|
|
"VALUES (?, ?, 'completed', ?, ?, ?)",
|
|
(
|
|
"legacy-comment",
|
|
'["issue-comment","stackchain/api",7,"legacy secret"]',
|
|
'{"id":42,"body":"legacy response"}',
|
|
90.0,
|
|
95.0,
|
|
),
|
|
)
|
|
|
|
replay = ledger.reserve(
|
|
"legacy-comment",
|
|
("issue-comment", "stackchain/api", 7, "legacy secret"),
|
|
)
|
|
|
|
assert replay.state == "completed"
|
|
assert replay.response == {"id": 42, "body": "legacy response"}
|
|
with sqlite3.connect(database) as connection:
|
|
fingerprint, response = connection.execute(
|
|
"SELECT fingerprint, response_json FROM idempotency_operations WHERE key = ?",
|
|
("legacy-comment",),
|
|
).fetchone()
|
|
assert fingerprint.startswith("v1:")
|
|
assert response.startswith("v1:")
|
|
|
|
|
|
def test_wrong_key_fails_closed_before_a_completed_result_can_replay(tmp_path):
|
|
database = tmp_path / "wrong-key.sqlite3"
|
|
ledger = IdempotencyLedger(
|
|
database, ttl_seconds=600, max_entries=256, encryption_key=PRIVATE_KEY
|
|
)
|
|
fingerprint = ("issue-comment", "stackchain/api", 7, "secret")
|
|
assert ledger.reserve("comment-7", fingerprint).state == "reserved"
|
|
ledger.complete("comment-7", {"id": 42})
|
|
|
|
wrong_key = IdempotencyLedger(
|
|
database, ttl_seconds=600, max_entries=256, encryption_key=b"x" * 32
|
|
)
|
|
with pytest.raises(
|
|
PrivateStateEncryptionError, match="private state could not be decrypted"
|
|
):
|
|
wrong_key.reserve("comment-7", fingerprint)
|
|
|
|
|
|
def test_completed_ciphertext_cannot_be_substituted_between_operation_keys(tmp_path):
|
|
database = tmp_path / "substitution.sqlite3"
|
|
ledger = IdempotencyLedger(
|
|
database, ttl_seconds=600, max_entries=256, encryption_key=PRIVATE_KEY
|
|
)
|
|
fingerprint = ("issue-create", "stackchain/api", "same authored request")
|
|
for key, number in (("create-a", 41), ("create-b", 42)):
|
|
assert ledger.reserve(key, fingerprint).state == "reserved"
|
|
ledger.complete(key, {"number": number})
|
|
with sqlite3.connect(database) as connection:
|
|
responses = dict(
|
|
connection.execute(
|
|
"SELECT key, response_json FROM idempotency_operations"
|
|
).fetchall()
|
|
)
|
|
connection.execute(
|
|
"UPDATE idempotency_operations SET response_json = ? WHERE key = ?",
|
|
(responses["create-b"], "create-a"),
|
|
)
|
|
|
|
with pytest.raises(
|
|
PrivateStateEncryptionError, match="private state could not be decrypted"
|
|
):
|
|
ledger.reserve("create-a", fingerprint)
|
|
|
|
|
|
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"]
|
|
|
|
|
|
def test_stale_pending_reservation_becomes_uncertain_without_consuming_capacity(tmp_path):
|
|
now = [1_000.0]
|
|
ledger = IdempotencyLedger(
|
|
tmp_path / "orphaned.sqlite3",
|
|
ttl_seconds=10,
|
|
max_entries=1,
|
|
clock=lambda: now[0],
|
|
)
|
|
fingerprint = ("issue-comment", "stackchain/api", 7, "Verify first")
|
|
|
|
assert ledger.reserve("orphan", fingerprint).state == "reserved"
|
|
now[0] += 11
|
|
|
|
assert ledger.reserve("orphan", fingerprint).state == "uncertain"
|
|
assert ledger.reserve("unrelated", ("issue-comment", "stackchain/api", 8, "Continue")).state == "reserved"
|
|
assert ledger.reserve("orphan", ("issue-comment", "stackchain/api", 7, "Changed")).state == "conflict"
|