319 lines
15 KiB
Python
319 lines
15 KiB
Python
|
from concurrent.futures import ThreadPoolExecutor
|
||
|
|
from dataclasses import replace
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
import sqlite3
|
||
|
|
import subprocess
|
||
|
|
import sys
|
||
|
|
import time
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
from informed_decision.approval_client import EntryResult
|
||
|
|
from informed_decision.disposition import Actor, ActorKind, DispositionRefused, Verb, record
|
||
|
|
from informed_decision.evidence import EventClass
|
||
|
|
from informed_decision.memo import Highlight, PacketItem
|
||
|
|
from informed_decision.provenance import Claim, HumanControlNotDischargeable, Route
|
||
|
|
from informed_decision.store import Conflict, EvidenceUnavailable, Store, StoreError
|
||
|
|
from test_skeleton import HUMAN, make_memo
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.fixture
|
||
|
|
def storage(tmp_path):
|
||
|
|
private = tmp_path / "private"
|
||
|
|
private.mkdir(mode=0o700)
|
||
|
|
store = Store(private / "evidence.sqlite")
|
||
|
|
digest = store.put_document(b"fixture packet material")
|
||
|
|
memo = make_memo(packet=(PacketItem("doc-1", "Fixture document", digest),),
|
||
|
|
brief="private brief sentinel", approval_binding_digest="sha256:" + "a" * 64)
|
||
|
|
store.save_memo(memo)
|
||
|
|
return store, memo
|
||
|
|
|
||
|
|
|
||
|
|
def present(store, memo, *, route=Route.AUTHENTICATION):
|
||
|
|
return store.present(memo.id, principal_sub=HUMAN.sub,
|
||
|
|
tenant=Claim("tenant:platform", Route.REGISTRATION), principal_type=Claim("human", route))
|
||
|
|
|
||
|
|
|
||
|
|
def prepare(storage, operation_id="click-1"):
|
||
|
|
store, memo = storage
|
||
|
|
p = present(store, memo)
|
||
|
|
store.acknowledge(p.id, HUMAN, ["h-1"])
|
||
|
|
d = store.record_disposition(p.id, HUMAN, Verb.ACCEPT, operation_id=operation_id)
|
||
|
|
return p, d
|
||
|
|
|
||
|
|
|
||
|
|
def abort_outbox(store):
|
||
|
|
with sqlite3.connect(store.path) as db:
|
||
|
|
db.execute("CREATE TRIGGER injected_failure BEFORE INSERT ON outbox BEGIN SELECT RAISE(ABORT,'injected disk failure'); END")
|
||
|
|
|
||
|
|
|
||
|
|
def test_packet_and_presentation_content_survive_reopen(storage):
|
||
|
|
store, memo = storage
|
||
|
|
p = present(store, memo)
|
||
|
|
reopened = Store(store.path)
|
||
|
|
saved, original, documents = reopened.retrieve_presentation(p.id)
|
||
|
|
assert saved == memo and original == p
|
||
|
|
assert documents == {"doc-1": b"fixture packet material"}
|
||
|
|
row = reopened.evidence()[0]
|
||
|
|
assert json.loads(row["content"])["memo"]["brief"] == "private brief sentinel"
|
||
|
|
assert "private brief sentinel" not in row["envelope"]
|
||
|
|
assert "fixture packet material" not in row["envelope"]
|
||
|
|
assert json.loads(row["envelope"])["data"]["content_exists"] is True
|
||
|
|
assert reopened.outbox()[0]["state"] == "pending"
|
||
|
|
|
||
|
|
|
||
|
|
def test_missing_packet_content_prevents_false_custody_assertion(storage):
|
||
|
|
store, memo = storage
|
||
|
|
altered = replace(memo, id="missing", packet=(PacketItem("doc-1", "Missing", "sha256:" + "b" * 64),))
|
||
|
|
with pytest.raises(EvidenceUnavailable): store.save_memo(altered)
|
||
|
|
with pytest.raises(EvidenceUnavailable): store.memo("missing")
|
||
|
|
assert store.evidence() == []
|
||
|
|
|
||
|
|
|
||
|
|
def test_versions_are_immutable_and_old_presentation_cannot_bind(storage):
|
||
|
|
store, memo = storage
|
||
|
|
p = present(store, memo)
|
||
|
|
store.save_memo(memo) # Exact import is harmless.
|
||
|
|
with pytest.raises(Conflict): store.save_memo(replace(memo, brief="silently changed"))
|
||
|
|
store.save_memo(memo.next_version(brief="revised"))
|
||
|
|
with pytest.raises(DispositionRefused, match="G_PRES"):
|
||
|
|
store.acknowledge(p.id, HUMAN, ["h-1"])
|
||
|
|
with pytest.raises(DispositionRefused, match="G_PRES"):
|
||
|
|
store.record_disposition(p.id, HUMAN, Verb.ACCEPT, operation_id="stale-click")
|
||
|
|
saved, _, _ = store.retrieve_presentation(p.id)
|
||
|
|
assert saved.brief == memo.brief # Retrieval uses the historic version.
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.parametrize("verb", [Verb.ACCEPT, Verb.DECLINE, Verb.ACKNOWLEDGE, Verb.RETURN, Verb.DISCUSS])
|
||
|
|
def test_actor_cannot_use_someone_elses_presentation(storage, verb):
|
||
|
|
store, memo = storage
|
||
|
|
p = present(store, memo)
|
||
|
|
other = Actor("other-person", ActorKind.PERSON)
|
||
|
|
with pytest.raises(DispositionRefused, match="G_ACTOR"):
|
||
|
|
record(memo, p.with_ack("h-1"), verb, other, reasons=("wrong_scope",))
|
||
|
|
before = store.evidence()
|
||
|
|
with pytest.raises(DispositionRefused, match="G_ACTOR"):
|
||
|
|
store.record_disposition(p.id, other, verb, operation_id="stolen-click", reasons=("wrong_scope",))
|
||
|
|
assert store.evidence() == before
|
||
|
|
|
||
|
|
|
||
|
|
def test_acknowledgments_are_explicit_append_only_and_idempotent(storage):
|
||
|
|
store, memo = storage
|
||
|
|
p = present(store, memo)
|
||
|
|
with pytest.raises(DispositionRefused, match="G_ACK"):
|
||
|
|
store.record_disposition(p.id, HUMAN, Verb.ACCEPT, operation_id="before-ack")
|
||
|
|
with pytest.raises(DispositionRefused, match="G_ACK"):
|
||
|
|
store.acknowledge(p.id, HUMAN, ["does-not-exist"])
|
||
|
|
with pytest.raises(DispositionRefused, match="G_ACTOR"):
|
||
|
|
store.acknowledge(p.id, Actor("other", ActorKind.PERSON), ["h-1"])
|
||
|
|
after = store.acknowledge(p.id, HUMAN, ["h-1"])
|
||
|
|
assert after.view_hash == p.view_hash and after.acked_highlight_ids == {"h-1"}
|
||
|
|
assert store.acknowledge(p.id, HUMAN, ["h-1"]) == after
|
||
|
|
assert len(store.evidence()) == 2
|
||
|
|
with sqlite3.connect(store.path) as db:
|
||
|
|
original = json.loads(db.execute("SELECT body FROM presentations WHERE id=?", (p.id,)).fetchone()[0])
|
||
|
|
assert original["acked_highlight_ids"] == []
|
||
|
|
|
||
|
|
|
||
|
|
def test_registration_human_assertion_cannot_bind(storage):
|
||
|
|
store, memo = storage
|
||
|
|
p = present(store, memo, route=Route.REGISTRATION)
|
||
|
|
store.acknowledge(p.id, HUMAN, ["h-1"])
|
||
|
|
with pytest.raises(HumanControlNotDischargeable):
|
||
|
|
store.record_disposition(p.id, HUMAN, Verb.ACCEPT, operation_id="unverified")
|
||
|
|
|
||
|
|
|
||
|
|
def test_return_discuss_and_decline_remain_distinct_without_engine_jobs(storage):
|
||
|
|
store, memo = storage
|
||
|
|
p = present(store, memo)
|
||
|
|
with pytest.raises(DispositionRefused, match="G_REASONS"):
|
||
|
|
store.record_disposition(p.id, HUMAN, Verb.RETURN, operation_id="empty-return")
|
||
|
|
returned = store.record_disposition(p.id, HUMAN, Verb.RETURN, operation_id="return",
|
||
|
|
reasons=("wrong_scope",), note="private note sentinel")
|
||
|
|
discussed = store.record_disposition(p.id, HUMAN, Verb.DISCUSS, operation_id="discuss", note="question")
|
||
|
|
store.acknowledge(p.id, HUMAN, ["h-1"])
|
||
|
|
declined = store.record_disposition(p.id, HUMAN, Verb.DECLINE, operation_id="decline")
|
||
|
|
assert [returned.verb, discussed.verb, declined.verb] == [Verb.RETURN, Verb.DISCUSS, Verb.DECLINE]
|
||
|
|
assert all(store.submission(d.id) is None for d in [returned, discussed, declined])
|
||
|
|
assert store.retrieve_disposition(returned.id)[0].note == "private note sentinel"
|
||
|
|
assert all("private note sentinel" not in r["envelope"] for r in store.evidence())
|
||
|
|
|
||
|
|
|
||
|
|
def test_click_idempotency_and_parallel_submission_reservation(storage):
|
||
|
|
store, memo = storage
|
||
|
|
p = present(store, memo)
|
||
|
|
store.acknowledge(p.id, HUMAN, ["h-1"])
|
||
|
|
def click(_):
|
||
|
|
return Store(store.path).record_disposition(p.id, HUMAN, Verb.ACCEPT, operation_id="same-click")
|
||
|
|
with ThreadPoolExecutor(max_workers=2) as pool:
|
||
|
|
results = list(pool.map(click, range(2)))
|
||
|
|
assert results[0] == results[1]
|
||
|
|
assert len(store.evidence()) == 3
|
||
|
|
with pytest.raises(Conflict):
|
||
|
|
store.record_disposition(p.id, HUMAN, Verb.ACCEPT, operation_id="same-click", note="changed")
|
||
|
|
def reserve(_):
|
||
|
|
try: return Store(store.path).begin_submission(results[0].id)
|
||
|
|
except Conflict: return None
|
||
|
|
with ThreadPoolExecutor(max_workers=2) as pool:
|
||
|
|
attempts = list(pool.map(reserve, range(2)))
|
||
|
|
assert sum(a is not None for a in attempts) == 1
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.parametrize("operation", ["present", "acknowledge", "disposition", "completion"])
|
||
|
|
def test_state_and_outbox_rollback_together(storage, operation):
|
||
|
|
store, memo = storage
|
||
|
|
p = present(store, memo)
|
||
|
|
if operation in ("disposition", "completion"):
|
||
|
|
store.acknowledge(p.id, HUMAN, ["h-1"])
|
||
|
|
if operation == "completion":
|
||
|
|
d = store.record_disposition(p.id, HUMAN, Verb.ACCEPT, operation_id="intent")
|
||
|
|
attempt = store.begin_submission(d.id)
|
||
|
|
before = store.evidence()
|
||
|
|
abort_outbox(store)
|
||
|
|
with pytest.raises(sqlite3.IntegrityError, match="injected"):
|
||
|
|
if operation == "present": present(store, memo)
|
||
|
|
elif operation == "acknowledge": store.acknowledge(p.id, HUMAN, ["h-1"])
|
||
|
|
elif operation == "disposition": store.record_disposition(p.id, HUMAN, Verb.ACCEPT, operation_id="intent")
|
||
|
|
else: store.finish_submission(d.id, attempt, EntryResult("appr-1", HUMAN.sub, "2026-09-10T20:00:00Z", "approved"))
|
||
|
|
assert store.evidence() == before
|
||
|
|
assert len(store.outbox()) == len(before)
|
||
|
|
if operation == "acknowledge": assert not store.presentation(p.id).acked_highlight_ids
|
||
|
|
if operation == "completion": assert store.submission(d.id)["state"] == "in_flight"
|
||
|
|
with sqlite3.connect(store.path) as db:
|
||
|
|
assert db.execute("SELECT COUNT(*) FROM presentations").fetchone()[0] == 1
|
||
|
|
assert db.execute("SELECT COUNT(*) FROM dispositions").fetchone()[0] == (1 if operation == "completion" else 0)
|
||
|
|
|
||
|
|
|
||
|
|
def test_process_death_before_commit_loses_neither_half(storage):
|
||
|
|
store, memo = storage
|
||
|
|
code = '''import os, sys
|
||
|
|
from informed_decision.store import Store
|
||
|
|
from informed_decision.provenance import Claim, Route
|
||
|
|
s=Store(sys.argv[1]); original=s._event
|
||
|
|
def die(*args, **kwargs):
|
||
|
|
original(*args, **kwargs)
|
||
|
|
os._exit(91)
|
||
|
|
s._event=die
|
||
|
|
s.present("memo-1", principal_sub="bernd", tenant=Claim("tenant:platform", Route.REGISTRATION), principal_type=Claim("human", Route.AUTHENTICATION))
|
||
|
|
'''
|
||
|
|
result = subprocess.run([sys.executable, "-c", code, str(store.path)])
|
||
|
|
assert result.returncode == 91
|
||
|
|
reopened = Store(store.path)
|
||
|
|
assert reopened.memo(memo.id) == memo and reopened.evidence() == [] and reopened.outbox() == []
|
||
|
|
with sqlite3.connect(store.path) as db:
|
||
|
|
assert db.execute("SELECT COUNT(*) FROM presentations").fetchone()[0] == 0
|
||
|
|
|
||
|
|
|
||
|
|
def test_confirmation_points_to_original_ack_snapshot(storage):
|
||
|
|
store, memo = storage
|
||
|
|
extra = Highlight("h-2", "doc-1", "optional")
|
||
|
|
store.save_memo(memo.next_version(highlights=(*memo.highlights, extra)))
|
||
|
|
p, d = prepare((store, store.memo(memo.id)))
|
||
|
|
attempt = store.begin_submission(d.id)
|
||
|
|
result = EntryResult("appr-1", HUMAN.sub, "2026-09-10T20:00:00Z", "approved")
|
||
|
|
assert store.finish_submission(d.id, attempt, result) == "confirmed"
|
||
|
|
store.acknowledge(p.id, HUMAN, ["h-2"])
|
||
|
|
recovered, snapshot, _, _ = Store(store.path).presentation_for_entry(*result.correlation)
|
||
|
|
assert recovered.id == d.id and snapshot.acked_highlight_ids == {"h-1"}
|
||
|
|
assert store.presentation(p.id).acked_highlight_ids == {"h-1", "h-2"}
|
||
|
|
p2 = present(store, store.memo(memo.id))
|
||
|
|
store.acknowledge(p2.id, HUMAN, ["h-1"])
|
||
|
|
with pytest.raises(Conflict): store.record_disposition(p2.id, HUMAN, Verb.ACCEPT, operation_id="new-click")
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.parametrize("kind", ["lost-reply", "old-duplicate", "crash-after-reserve"])
|
||
|
|
def test_uncertain_entry_never_gets_a_new_presentation_or_automatic_retry(storage, kind):
|
||
|
|
store, memo = storage
|
||
|
|
p, d = prepare(storage)
|
||
|
|
attempt = store.begin_submission(d.id)
|
||
|
|
if kind != "crash-after-reserve":
|
||
|
|
result = EntryResult("appr-1", HUMAN.sub, "2020-01-01T00:00:00Z", "approved", True) if kind == "old-duplicate" else None
|
||
|
|
assert store.finish_submission(d.id, attempt, result) == "unresolved"
|
||
|
|
reopened = Store(store.path)
|
||
|
|
with pytest.raises(Conflict): reopened.begin_submission(d.id)
|
||
|
|
with pytest.raises(EvidenceUnavailable): reopened.presentation_for_entry("appr-1", HUMAN.sub, "2020-01-01T00:00:00Z")
|
||
|
|
assert reopened.submission(d.id)["approved_at"] is None
|
||
|
|
assert not any(json.loads(r["envelope"])["data"].get("entry_correlation") for r in reopened.evidence())
|
||
|
|
|
||
|
|
|
||
|
|
def test_updated_memo_stops_already_prepared_submission(storage):
|
||
|
|
store, memo = storage
|
||
|
|
_, d = prepare(storage)
|
||
|
|
store.save_memo(memo.next_version(brief="new question"))
|
||
|
|
with pytest.raises(DispositionRefused, match="G_PRES"): store.begin_submission(d.id)
|
||
|
|
assert store.submission(d.id)["state"] == "prepared"
|
||
|
|
|
||
|
|
|
||
|
|
def test_revision_cannot_race_an_in_flight_or_uncertain_entry(storage):
|
||
|
|
store, memo = storage
|
||
|
|
_, d = prepare(storage)
|
||
|
|
attempt = store.begin_submission(d.id)
|
||
|
|
with pytest.raises(Conflict): store.save_memo(memo.next_version(brief="racing edit"))
|
||
|
|
store.finish_submission(d.id, attempt)
|
||
|
|
with pytest.raises(Conflict): store.save_memo(memo.next_version(brief="unresolved edit"))
|
||
|
|
|
||
|
|
|
||
|
|
def test_wrong_entry_or_attempt_cannot_confirm(storage):
|
||
|
|
store, _ = storage
|
||
|
|
_, d = prepare(storage)
|
||
|
|
attempt = store.begin_submission(d.id)
|
||
|
|
with pytest.raises(Conflict): store.finish_submission(d.id, "wrong-attempt")
|
||
|
|
with pytest.raises(Conflict):
|
||
|
|
store.finish_submission(d.id, attempt, EntryResult("other-approval", HUMAN.sub, "2026-09-10T20:00:00Z", "approved"))
|
||
|
|
assert store.submission(d.id)["state"] == "in_flight"
|
||
|
|
|
||
|
|
|
||
|
|
def test_evidence_is_append_only_and_backup_preserves_pending_and_confirmed(storage):
|
||
|
|
store, _ = storage
|
||
|
|
p, d = prepare(storage)
|
||
|
|
attempt = store.begin_submission(d.id)
|
||
|
|
result = EntryResult("appr-1", HUMAN.sub, "2026-09-10T20:00:00Z", "approved")
|
||
|
|
store.finish_submission(d.id, attempt, result)
|
||
|
|
with sqlite3.connect(store.path) as db:
|
||
|
|
for statement in ["DELETE FROM evidence", "UPDATE presentations SET body='{}'", "DELETE FROM documents", "DELETE FROM dispositions"]:
|
||
|
|
with pytest.raises(sqlite3.IntegrityError, match="immutable"): db.execute(statement)
|
||
|
|
destination = store.path.parent / "backup.sqlite"
|
||
|
|
store.backup(destination)
|
||
|
|
backup = Store(destination)
|
||
|
|
assert backup.outbox() == store.outbox()
|
||
|
|
assert backup.presentation_for_entry(*result.correlation)[0].id == d.id
|
||
|
|
with pytest.raises(Conflict): store.backup(destination)
|
||
|
|
|
||
|
|
|
||
|
|
def test_store_refuses_unsafe_paths_and_unknown_schema(tmp_path):
|
||
|
|
private = tmp_path / "private"
|
||
|
|
private.mkdir(mode=0o755)
|
||
|
|
with pytest.raises(StoreError): Store(private / "db")
|
||
|
|
private.chmod(0o700)
|
||
|
|
db = private / "db"
|
||
|
|
db.touch(mode=0o644)
|
||
|
|
with pytest.raises(StoreError): Store(db)
|
||
|
|
db.chmod(0o600)
|
||
|
|
with sqlite3.connect(db) as conn: conn.execute("PRAGMA user_version=99")
|
||
|
|
with pytest.raises(StoreError, match="schema"): Store(db)
|
||
|
|
link = private / "linked"
|
||
|
|
link.symlink_to(db)
|
||
|
|
with pytest.raises(StoreError): Store(link)
|
||
|
|
|
||
|
|
|
||
|
|
def test_expired_delivery_lease_replays_same_id_and_bytes(storage):
|
||
|
|
store, memo = storage
|
||
|
|
present(store, memo)
|
||
|
|
first = store.claim_delivery(now=10)
|
||
|
|
assert store.claim_delivery(now=11) is None
|
||
|
|
again = Store(store.path).claim_delivery(now=41)
|
||
|
|
assert first[0] == again[0] and first[2] == again[2] and first[1] != again[1]
|
||
|
|
with pytest.raises(Conflict): store.finish_delivery(first[0], first[1], reference="old-receipt")
|
||
|
|
|
||
|
|
|
||
|
|
def test_heartbeats_do_not_mask_backlog_and_are_per_class(storage):
|
||
|
|
store, memo = storage
|
||
|
|
present(store, memo)
|
||
|
|
future = time.time() + 90000
|
||
|
|
assert store.queue_heartbeats(now=future) == 2
|
||
|
|
assert store.queue_heartbeats(now=future) == 0
|
||
|
|
beats = [json.loads(r["envelope"]) for r in store.evidence() if r["class"] == EventClass.HEARTBEAT.value]
|
||
|
|
assert {b["data"]["class"] for b in beats} == {EventClass.DISPOSITION.value, EventClass.STANCE_APPLICATION.value}
|
||
|
|
assert all(b["data"]["assertion"] == "nothing-to-report" for b in beats)
|