approval-engine/tests/test_outbox.py

import json

import pytest

from approval_engine.errors import DuplicateApprover, 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_a_positive_claim_per_declared_class(engine):
    """One assertion per class, because silence is class-shaped.

    A single per-source heartbeat is discharged by whichever class is busy.
    `revocation` is the class whose silence matters here, and it is the one a
    per-source beat would hide behind `issuance`.
    """
    from approval_engine.store import HEARTBEAT_CLASSES

    approve(engine)
    beat = engine.emit_heartbeat()
    assert beat["assertion"] == "nothing-to-report"
    assert beat["counts"]["issuance"] == 1
    assert set(beat["classes"]) == set(HEARTBEAT_CLASSES)

    pending = [p for p in engine.undrained() if p["class"] == "heartbeat"]
    assert len(pending) == len(HEARTBEAT_CLASSES)
    asserted = {p["payload"]["details"]["class"] for p in pending}
    assert asserted == set(HEARTBEAT_CLASSES)
    assert "revocation" in asserted
    for p in pending:
        assert p["payload"]["details"]["assertion"] == "nothing-to-report"
        assert p["payload"]["action"] == "audit-core.heartbeat"


def test_heartbeat_classes_are_emitted_atomically(engine):
    """Partial emission would report some classes healthy and others stalled."""
    from approval_engine.store import HEARTBEAT_CLASSES

    engine.fail_outbox = True
    try:
        with pytest.raises(Exception):
            engine.emit_heartbeat()
    finally:
        engine.fail_outbox = False
    assert [p for p in engine.undrained() if p["class"] == "heartbeat"] == []
    engine.emit_heartbeat()
    assert len(
        [p for p in engine.undrained() if p["class"] == "heartbeat"]
    ) == len(HEARTBEAT_CLASSES)


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())


def test_drain_failure_records_bounded_attempt_state(engine):
    approve(engine)

    class SensitiveFailure(Exception):
        pass

    result = engine.drain(lambda _payload: (_ for _ in ()).throw(SensitiveFailure("secret")))
    assert result["failed"] == 1
    pending = engine.undrained()
    assert pending[0]["attempts"] == 1
    assert pending[0]["last_error"] == "SensitiveFailure"
    stats = engine.outbox_stats()
    assert stats["failed_pending"] == 1
    assert stats["attempts"] == 1


def _event(engine, cls):
    return next(p["payload"] for p in engine.undrained() if p["class"] == cls)


def test_issuance_carries_the_threshold_evaluation(engine):
    """GH-DEC-2026-005 §9.6: the PEP no longer counts approvers, so the
    evaluation must be recoverable from what this engine emitted."""
    approve(engine, required=2)
    threshold = _event(engine, "issuance")["details"]["threshold"]
    assert threshold["required_count"] == 2
    assert threshold["distinct_approver_count"] == 2
    assert threshold["threshold_met"] is True
    assert [a["subject_id"] for a in threshold["approvers"]] == [
        "user:approver-0",
        "user:approver-1",
    ]
    assert all(a["approved_at"] for a in threshold["approvers"])


def test_use_event_reconstructs_the_threshold_without_live_rows(engine):
    """An auditor holding only the use row must be able to recompute it."""
    obj = approve(engine, required=2)
    engine.consume(obj.id, obj.binding_digest)
    threshold = _event(engine, "use")["details"]["threshold"]
    assert threshold["required_count"] == 2
    assert threshold["distinct_approver_count"] == 2
    assert threshold["threshold_met"] is True
    approvers = [a["subject_id"] for a in threshold["approvers"]]
    assert approvers == ["user:approver-0", "user:approver-1"]
    # the recomputation an auditor performs
    assert len(set(approvers)) >= threshold["required_count"]


def test_duplicate_approver_is_refused_so_distinctness_is_an_invariant(engine):
    """Dual control is enforced at storage, not recomputed from evidence."""
    obj = engine.create(binding(), validity(), required_count=2)
    engine.add_entry(obj.id, "user:alice")
    with pytest.raises(DuplicateApprover):
        engine.add_entry(obj.id, "user:alice")
    assert engine.get(obj.id).status == "requested"
    engine.add_entry(obj.id, "user:bob")
    threshold = _event(engine, "issuance")["details"]["threshold"]
    assert threshold["distinct_approver_count"] == 2
    assert threshold["threshold_met"] is True
    assert [a["subject_id"] for a in threshold["approvers"]] == [
        "user:alice",
        "user:bob",
    ]


def test_claim_still_discloses_no_approver_identities(engine):
    """Identities are evidence for audit-core, never consumer-facing."""
    obj = approve(engine, required=2)
    claim = engine.claim(obj.id)
    assert "threshold" not in claim
    assert "approvers" not in claim
    assert "user:approver-0" not in json.dumps(claim)
    assert claim["valid_now"] is True