tegwick
c3f1dfbc07
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a05e2e-805b-7042-a750-71f473bceea2
84 lines
2.7 KiB
Python
84 lines
2.7 KiB
Python
from approval_engine.errors import StoreUnavailable
|
|
from approval_engine.store import Engine
|
|
from tests.conftest import FROZEN, approve, binding, validity
|
|
|
|
|
|
def test_issuance_queued_in_same_commit(engine):
|
|
obj = approve(engine)
|
|
pending = engine.undrained()
|
|
classes = [p["class"] for p in pending]
|
|
assert classes == ["issuance"]
|
|
assert pending[0]["approval_id"] == obj.id
|
|
payload = pending[0]["payload"]
|
|
assert payload["source"] == "approval-engine"
|
|
assert payload["action"] == "approval.issuance"
|
|
assert payload["event_id"] == pending[0]["event_id"]
|
|
|
|
|
|
def test_failed_outbox_rolls_back_mutation():
|
|
eng = Engine(":memory:", clock=lambda: FROZEN, fail_outbox=True)
|
|
obj = eng.create(binding(), validity(), required_count=1)
|
|
try:
|
|
eng.add_entry(obj.id, "user:alice")
|
|
raise AssertionError("must fail")
|
|
except StoreUnavailable:
|
|
pass
|
|
obj = eng.get(obj.id)
|
|
assert obj.status == "requested"
|
|
assert obj.entries == []
|
|
assert eng.undrained() == []
|
|
eng.close()
|
|
|
|
|
|
def test_revoke_succeeds_when_drain_sink_is_down(engine):
|
|
obj = approve(engine)
|
|
engine.revoke(obj.id)
|
|
assert engine.get(obj.id).status == "revoked"
|
|
|
|
def down(_payload):
|
|
raise ConnectionError("audit-core unreachable")
|
|
|
|
result = engine.drain(down)
|
|
assert result["failed"] >= 1
|
|
assert engine.get(obj.id).status == "revoked"
|
|
assert any(p["class"] == "revocation" for p in engine.undrained())
|
|
|
|
|
|
def test_drain_marks_delivered(engine):
|
|
approve(engine)
|
|
sink: list[dict] = []
|
|
result = engine.drain(sink.append)
|
|
assert result["delivered"] == 1
|
|
assert result["failed"] == 0
|
|
assert engine.undrained() == []
|
|
assert sink[0]["action"] == "approval.issuance"
|
|
|
|
|
|
def test_heartbeat_is_positive_claim(engine):
|
|
approve(engine)
|
|
beat = engine.emit_heartbeat()
|
|
assert beat["assertion"] == "nothing-to-report"
|
|
assert beat["counts"]["issuance"] == 1
|
|
assert beat["counts"]["heartbeat"] == 1
|
|
pending = [p for p in engine.undrained() if p["class"] == "heartbeat"]
|
|
assert len(pending) == 1
|
|
assert pending[0]["payload"]["details"]["assertion"] == "nothing-to-report"
|
|
|
|
|
|
def test_revocation_event_class(engine):
|
|
obj = approve(engine)
|
|
engine.revoke(obj.id)
|
|
classes = [p["class"] for p in engine.undrained()]
|
|
assert "revocation" in classes
|
|
|
|
|
|
def test_failed_outbox_rolls_back_consume(engine):
|
|
obj = approve(engine)
|
|
engine.fail_outbox = True
|
|
try:
|
|
engine.consume(obj.id, "sha256:" + "34" * 32)
|
|
raise AssertionError("must fail")
|
|
except StoreUnavailable:
|
|
pass
|
|
assert engine.get(obj.id).status == "approved"
|
|
assert all(item["class"] != "use" for item in engine.undrained())
|