informed-decision/tests/test_approval_http.py

import copy
from dataclasses import replace
import time

import pytest

from informed_decision.approval_client import ApprovalEngineError
from informed_decision.approval_http import ApprovalHTTPClient
from informed_decision.http_transport import TransportError
from informed_decision.oidc import HumanSession
from informed_decision.provenance import Claim, Route

ORIGIN = "https://approval.test"
APPROVAL = {"id": "fixture", "binding": {"digest": "sha256:" + "a" * 64, "human_control": True},
            "status": "approved", "updated_at": "2099-01-01T00:00:00Z", "entries": [
                {"subject_id": "human-fixture", "principal_type": "human", "approved_at": "2026-09-10T00:00:00Z"}]}


def session():
    return HumanSession("human-fixture", Claim("tenant:platform", Route.REGISTRATION),
                        Claim("human", Route.AUTHENTICATION), {}, time.time() + 900, "synthetic-access-token")


class Responses:
    def __init__(self, *responses):
        self.responses = list(responses)
        self.calls = []

    def request(self, method, url, **kwargs):
        self.calls.append((method, url, kwargs))
        result = self.responses.pop(0)
        if isinstance(result, Exception): raise result
        return copy.deepcopy(result)


def test_entry_uses_access_token_empty_body_and_actual_entry_correlation():
    transport = Responses((200, APPROVAL), (200, APPROVAL))
    client = ApprovalHTTPClient(ORIGIN, session(), transport=transport)
    result = client.add_entry("fixture")
    assert result.correlation == ("fixture", "human-fixture", "2026-09-10T00:00:00Z")
    assert not result.duplicate
    assert [x[:2] for x in transport.calls] == [("GET", ORIGIN + "/v1/approvals/fixture"),
        ("POST", ORIGIN + "/v1/approvals/fixture/entries")]
    assert transport.calls[1][2]["body"] == b"{}"
    assert transport.calls[1][2]["headers"]["Authorization"] == "Bearer synthetic-access-token"
    assert not hasattr(client, "consume")


def test_duplicate_recovers_original_entry_never_invents_time_or_reposts():
    transport = Responses((200, APPROVAL), (409, {"error": "duplicate_approver"}), (200, APPROVAL))
    result = ApprovalHTTPClient(ORIGIN, session(), transport=transport).add_entry("fixture")
    assert result.duplicate and result.approved_at == APPROVAL["entries"][0]["approved_at"]
    assert [x[0] for x in transport.calls] == ["GET", "POST", "GET"]


@pytest.mark.parametrize("value", [False, None, "true", 1])
def test_undeclared_or_malformed_human_control_refused_before_mutation(value):
    data = copy.deepcopy(APPROVAL)
    data["binding"]["human_control"] = value
    transport = Responses((200, data))
    with pytest.raises(ApprovalEngineError, match="invalid_approval_response"):
        ApprovalHTTPClient(ORIGIN, session(), transport=transport).add_entry("fixture")
    assert len(transport.calls) == 1 and transport.calls[0][0] == "GET"


@pytest.mark.parametrize("value", [[], [{"subject_id": "other"}], [1], None,
    [{"subject_id": "human-fixture", "principal_type": "service", "approved_at": "2026-09-10T00:00:00Z"}],
    [{"subject_id": "human-fixture", "principal_type": "human", "approved_at": "2026-09-10"}],
    [{"subject_id": "human-fixture", "principal_type": "human", "approved_at": None}]])
def test_missing_corresponding_entry_cannot_be_reported_as_success(value):
    data = copy.deepcopy(APPROVAL)
    data["entries"] = value
    transport = Responses((200, APPROVAL), (200, data))
    with pytest.raises(ApprovalEngineError, match="entry_correlation_missing"):
        ApprovalHTTPClient(ORIGIN, session(), transport=transport).add_entry("fixture")


@pytest.mark.parametrize("response,reason", [
    ((403, {"error": "forbidden", "message": "secret-sentinel"}), "forbidden"),
    ((409, {"error": "conflict"}), "conflict"),
    ((503, {"error": "store_unavailable"}), "store_unavailable"),
    ((500, {"error": ["secret-sentinel"]}), "upstream_refused"),
    (TransportError("secret-sentinel"), "upstream_unavailable"),
])
def test_refusal_and_uncertain_post_are_not_retried(response, reason):
    transport = Responses((200, APPROVAL), response)
    with pytest.raises(ApprovalEngineError) as error:
        ApprovalHTTPClient(ORIGIN, session(), transport=transport).add_entry("fixture")
    assert error.value.reason == reason and "secret-sentinel" not in str(error.value)
    assert len(transport.calls) == 2


@pytest.mark.parametrize("identifier", ["../consume", "a/entries", "a?consume", "a%2Fentries", "", "a" * 129])
def test_identifier_cannot_inject_a_route(identifier):
    transport = Responses()
    with pytest.raises(ValueError):
        ApprovalHTTPClient(ORIGIN, session(), transport=transport).get_approval(identifier)
    assert not transport.calls


def test_session_expiry_refused_before_request():
    transport = Responses()
    with pytest.raises(ApprovalEngineError, match="session_expired"):
        ApprovalHTTPClient(ORIGIN, replace(session(), expires_at=1), transport=transport).get_approval("fixture")
    assert not transport.calls


def test_read_current_state_each_time():
    revoked = {**APPROVAL, "status": "revoked"}
    transport = Responses((200, APPROVAL), (200, revoked))
    client = ApprovalHTTPClient(ORIGIN, session(), transport=transport)
    assert client.get_approval("fixture")["status"] == "approved"
    assert client.get_approval("fixture")["status"] == "revoked"


def test_cluster_http_is_explicit_and_cannot_select_a_public_http_endpoint():
    with pytest.raises(ValueError): ApprovalHTTPClient("http://approval-engine.approval-engine.svc", session())
    client = ApprovalHTTPClient("http://approval-engine.approval-engine.svc", session(), allow_internal_http=True)
    assert client.origin.startswith("http:")
    with pytest.raises(ValueError): ApprovalHTTPClient("http://attacker.test", session(), allow_internal_http=True)