approval-engine/tests/test_t06_replay.py

"""Canon T-06 — Approval Replay.

Reuse a previously valid approval artifact for a different target, parameter
set, or later time. Pass: parameter binding, expiry, or replay protection
rejects the request.

Consume-side replay follows GH-DEC-2026-003: same-digest retries are
idempotent; a different digest against a consumed object conflicts.
"""

from datetime import datetime, timezone

from approval_engine.binding import binding_digest
from approval_engine.errors import Conflict
from approval_engine.store import Engine
from tests.conftest import approve, binding, validity


def _consumer_accepts(claim, proposed_binding) -> bool:
    if claim["issuer"] != "approval-engine":
        return False
    if not claim["valid_now"]:
        return False
    if claim["consumed"]:
        return False
    if claim["reason_code"] != "ok":
        return False
    return claim["binding"]["digest"] == binding_digest(proposed_binding)


def test_t06_wrong_target_rejected(engine):
    obj = approve(engine)
    claim = engine.claim(obj.id)
    assert _consumer_accepts(claim, binding()) is True
    other = binding(target={"id": "some-other-lane", "stage": "prod"})
    assert _consumer_accepts(claim, other) is False


def test_t06_wrong_action_rejected(engine):
    obj = approve(engine)
    claim = engine.claim(obj.id)
    other = binding(action="secrets.kv.read")
    assert _consumer_accepts(claim, other) is False


def test_t06_later_time_expired():
    jumping = {"t": datetime(2026, 8, 29, 12, 0, tzinfo=timezone.utc)}
    eng = Engine(":memory:", clock=lambda: jumping["t"])
    obj = approve(eng)
    claim = eng.claim(obj.id)
    assert _consumer_accepts(claim, binding()) is True
    jumping["t"] = datetime(2026, 8, 29, 16, 0, tzinfo=timezone.utc)
    claim = eng.claim(obj.id)
    assert claim["state"] == "expired"
    assert _consumer_accepts(claim, binding()) is False
    eng.close()


def test_t06_revoked_rejected(engine):
    obj = approve(engine)
    engine.revoke(obj.id)
    assert _consumer_accepts(engine.claim(obj.id), binding()) is False


def test_t06_superseded_rejected(engine):
    obj = approve(engine)
    engine.supersede(obj.id)
    assert _consumer_accepts(engine.claim(obj.id), binding()) is False


def test_t06_consumed_approval_rejects_different_request_digest(engine):
    obj = approve(engine)
    first = "sha256:" + "56" * 32
    engine.consume(obj.id, first)
    try:
        engine.consume(obj.id, "sha256:" + "78" * 32)
        raise AssertionError("consume-side replay must conflict")
    except Conflict:
        pass


def test_t06_consumed_approval_allows_same_request_retry(engine):
    obj = approve(engine)
    digest = "sha256:" + "9a" * 32
    engine.consume(obj.id, digest)
    retry = engine.consume(obj.id, digest)
    assert retry["idempotent"] is True