approval-engine/tests/test_execution_confirmation.py

import json
from datetime import timedelta
from urllib.parse import urlsplit
from urllib.error import HTTPError

import pytest

from approval_engine.api import call
from approval_engine.errors import Conflict
from approval_engine.execution_confirmation import ExecutionConfirmationClient
from approval_engine.pep import ApprovalProtocolError
from tests.conftest import FROZEN, binding, validity
from tests.test_pep import Response

FULL = "sha256:" + "ab" * 32
FREE = "sha256:" + "cd" * 32


def client(tmp_path, opener):
    token = tmp_path / "test-token"
    token.write_text("test-token")
    return ExecutionConfirmationClient("http://approval-engine", token, opener=opener)


def consumed(engine):
    obj = engine.create(binding(), validity(), pdp_digest=FREE, pdp_path=True,
                        human_control=True)
    engine.add_entry(obj.id, "human:founder", principal_type="human")
    engine.consume(obj.id, FULL, decision_id="decision:original")
    return obj


def confirmer(c, obj):
    return c.confirm_consumption(obj.id, FULL, "decision:original",
                                 expected_binding=binding(), expected_pdp_digest=FREE)


def test_confirmation_reads_authenticated_current_object_without_mutation(engine, app, tmp_path):
    obj = consumed(engine)
    before = list(engine.undrained())
    requests = []

    def opener(req, timeout):
        requests.append(req)
        status, result = call(app, req.get_method(), urlsplit(req.full_url).path,
                              authorization=req.get_header("Authorization"))
        assert status == 200
        return Response(result)

    result = confirmer(client(tmp_path, opener), obj)
    assert result == {"approval_id": obj.id, "request_digest": FULL,
                      "decision_id": "decision:original", "consumed_at": FROZEN.isoformat()}
    assert requests[0].get_method() == "GET"
    assert engine.undrained() == before


@pytest.mark.parametrize("field,value", [
    ("id", "different"), ("status", "approved"),
    ("consumption", {"request_digest": FULL, "decision_id": "different", "consumed_at": FROZEN.isoformat()}),
    ("consumption", {"request_digest": FREE, "decision_id": "decision:original", "consumed_at": FROZEN.isoformat()}),
    ("consumption", {"request_digest": FULL, "decision_id": "decision:original", "consumed_at": "naive"}),
    ("consumption", None), ("effect", "ALLOW"),
])
def test_wrong_consumption_object_refuses(engine, tmp_path, field, value):
    obj = consumed(engine)
    result = engine.get(obj.id).as_dict()
    result[field] = value
    with pytest.raises(ApprovalProtocolError):
        confirmer(client(tmp_path, lambda *_a, **_k: Response(result)), obj)


@pytest.mark.parametrize("field,value", [
    ("action", "exec"), ("target", {"id": "sibling"}), ("actor", "forged"),
    ("principal", "forged"), ("purpose", "different"), ("pdp_digest", FULL),
    ("pdp_path", "true"), ("human_control", 1),
])
def test_wrong_act_or_digest_or_declaration_refuses(engine, tmp_path, field, value):
    obj = consumed(engine)
    result = engine.get(obj.id).as_dict()
    result["binding"][field] = value
    with pytest.raises(ApprovalProtocolError):
        confirmer(client(tmp_path, lambda *_a, **_k: Response(result)), obj)


def test_retry_is_historical_after_expiry_and_retains_original_decision(engine, tmp_path):
    obj = consumed(engine)
    engine.clock = lambda: FROZEN + timedelta(days=1)
    retry = engine.consume(obj.id, FULL, decision_id="decision:new")
    assert retry["idempotent"] is True
    assert retry["decision_id"] == "decision:original"
    with pytest.raises(ApprovalProtocolError):
        client(tmp_path, lambda *_a, **_k: Response(retry)).consume_exact(obj.id, FULL, "decision:new")
    assert engine.claim(obj.id)["valid_now"] is False
    assert confirmer(client(tmp_path, lambda *_a, **_k: Response(engine.get(obj.id).as_dict())), obj)
    with pytest.raises(Conflict):
        engine.revoke(obj.id)
    with pytest.raises(Conflict):
        engine.supersede(obj.id)


@pytest.mark.parametrize("field,value", [
    ("approval_id", "wrong"), ("decision_id", None), ("idempotent", "false"),
    ("consumed_at", "2026-10-11T12:00:00"), ("effect", "ALLOW"),
])
def test_exact_consume_response_refuses_loose_receipt(engine, tmp_path, field, value):
    obj = consumed(engine)
    result = engine.consume(obj.id, FULL, decision_id="decision:original")
    result[field] = value
    with pytest.raises(ApprovalProtocolError):
        client(tmp_path, lambda *_a, **_k: Response(result)).consume_exact(obj.id, FULL, "decision:original")


def test_exact_consume_filters_extraneous_payload(engine, tmp_path):
    obj = consumed(engine)
    result = engine.consume(obj.id, FULL, decision_id="decision:original")
    result["notes"] = "discard"
    confirmed = client(tmp_path, lambda *_a, **_k: Response(result)).consume_exact(obj.id, FULL, "decision:original")
    assert "notes" not in confirmed


def test_denied_read_never_represents_confirmation(engine, app, tmp_path):
    obj = consumed(engine)
    def opener(req, timeout):
        status, result = call(app, "GET", urlsplit(req.full_url).path, authorization=None)
        assert status == 401
        raise HTTPError(req.full_url, status, "refused", {}, None)
    with pytest.raises(ApprovalProtocolError, match="unauthorized"):
        confirmer(client(tmp_path, opener), obj)