import json
import threading
from io import BytesIO
from urllib.error import HTTPError, URLError
from wsgiref.simple_server import make_server

import pytest

from approval_engine.api import App, call
from approval_engine.auth import Identity, StaticTokenAuthenticator
from approval_engine.pep import (
    ApprovalHTTPClient,
    ApprovalProtocolError,
    ProtectedActionHarness,
)
from approval_engine.store import Engine
from tests.conftest import FROZEN, binding, validity

DIGEST = "sha256:" + "ab" * 32
OTHER = "sha256:" + "cd" * 32


class Client:
    def __init__(self, claim=None, consume=None, failure=None):
        self.claim_result = claim or {"valid_now": True, "consumed": False}
        self.consume_result = consume
        self.failure = failure
        self.calls = []

    def claim(self, approval_id):
        self.calls.append("claim")
        if self.failure == "claim":
            raise ApprovalProtocolError("down")
        return self.claim_result

    def consume(self, approval_id, digest, decision_id):
        self.calls.append("consume")
        if self.failure == "consume":
            raise ApprovalProtocolError("conflict")
        return self.consume_result or {"status": "consumed", "request_digest": digest}


class Response:
    def __init__(self, body, status=200):
        self.body = json.dumps(body).encode()
        self.status = status

    def getcode(self):
        return self.status

    def read(self, _size):
        return self.body

    def close(self):
        pass


def _token(tmp_path, value="test-token"):
    path = tmp_path / "token"
    path.write_text(value)
    return path


def allow(_claim):
    return {"effect": "ALLOW", "decision_id": "decision:1", "request_digest": DIGEST}


def test_http_client_rereads_mounted_token(tmp_path):
    token = _token(tmp_path, "first")
    seen = []

    def opener(request, timeout):
        seen.append((request.get_header("Authorization"), timeout))
        return Response({"valid_now": True, "consumed": False})

    client = ApprovalHTTPClient("http://approval-engine:8080", token, opener=opener)
    client.claim("approval-1")
    token.write_text("second")
    client.claim("approval-1")
    assert [item[0] for item in seen] == ["Bearer first", "Bearer second"]


def test_side_effect_occurs_only_after_claim_decision_and_consume():
    client = Client()
    order = client.calls
    result = ProtectedActionHarness(client).execute(
        "approval:1",
        DIGEST,
        lambda claim: (order.append("decision"), allow(claim))[1],
        lambda: (order.append("side-effect"), "dry-run-only")[1],
    )
    assert result == "dry-run-only"
    assert order == ["claim", "decision", "consume", "side-effect"]


@pytest.mark.parametrize("failure", ["claim", "consume"])
def test_unavailable_or_conflicting_engine_prevents_side_effect(failure):
    client = Client(failure=failure)
    effects = []
    with pytest.raises(ApprovalProtocolError):
        ProtectedActionHarness(client).execute(
            "approval:1", DIGEST, allow, lambda: effects.append("called")
        )
    assert effects == []


def test_deny_or_digest_mismatch_prevents_consume_and_side_effect():
    for decision in (
        {"effect": "DENY", "decision_id": "decision:1", "request_digest": DIGEST},
        {"effect": "ALLOW", "decision_id": "decision:1", "request_digest": OTHER},
    ):
        client = Client()
        effects = []
        with pytest.raises(ApprovalProtocolError):
            ProtectedActionHarness(client).execute(
                "approval:1",
                DIGEST,
                lambda _claim, value=decision: value,
                lambda: effects.append("called"),
            )
        assert client.calls == ["claim"]
        assert effects == []


def test_consume_http_errors_fail_closed(tmp_path):
    def opener(request, timeout):
        raise HTTPError(
            request.full_url,
            409,
            "conflict",
            hdrs=None,
            fp=BytesIO(b'{"error":"conflict"}'),
        )

    client = ApprovalHTTPClient("http://approval.test", _token(tmp_path), opener=opener)
    with pytest.raises(ApprovalProtocolError, match="conflict"):
        client.consume("approval-1", DIGEST, "decision:1")


def test_unreachable_engine_fails_closed(tmp_path):
    def opener(request, timeout):
        raise URLError("down")

    client = ApprovalHTTPClient("http://approval.test", _token(tmp_path), opener=opener)
    with pytest.raises(ApprovalProtocolError, match="unreachable"):
        client.claim("approval-1")


def test_consume_rejects_non_canonical_digest_before_http(tmp_path):
    client = ApprovalHTTPClient(
        "http://approval.test",
        _token(tmp_path),
        opener=lambda *_args, **_kwargs: pytest.fail("must not HTTP"),
    )
    with pytest.raises(ApprovalProtocolError, match="canonical request digest"):
        client.consume("approval-1", "not-a-digest", "decision:1")


def test_consume_does_not_treat_decision_shaped_payload_as_permission(tmp_path):
    def opener(request, timeout):
        if request.get_method() == "GET":
            return Response({"valid_now": True, "consumed": False})
        return Response(
            {
                "status": "consumed",
                "request_digest": DIGEST,
                "effect": "ALLOW",
            }
        )

    client = ApprovalHTTPClient("http://approval.test", _token(tmp_path), opener=opener)
    effects = []
    with pytest.raises(ApprovalProtocolError, match="not a permission"):
        ProtectedActionHarness(client).execute(
            "approval-1", DIGEST, allow, lambda: effects.append("called")
        )
    assert effects == []


@pytest.fixture
def live_http(tmp_path):
    engine = Engine(tmp_path / "approval.sqlite", clock=lambda: FROZEN)
    identity = Identity(
        subject="agt-secrets-engine",
        issuer="https://keycape.example",
        audiences=("approval-engine",),
        principal_type="service",
        tenant="platform",
        roles=frozenset({"secrets-engine"}),
        scopes=frozenset(
            {
                "approval:create",
                "approval:read",
                "approval:approve",
                "approval:revoke",
                "approval:supersede",
                "approval:consume",
                "approval:observe",
                "approval:emit",
            }
        ),
        assurance={"level": "aal1", "methods": ["test"], "source": "test"},
        evidence_ref="test-identity:test-token",
    )
    app = App(engine, StaticTokenAuthenticator({"test-token": identity}))
    server = make_server("127.0.0.1", 0, app)
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    try:
        yield f"http://127.0.0.1:{server.server_port}", app
    finally:
        server.shutdown()
        thread.join(timeout=2)
        engine.close()


def _approved(app):
    _, created = call(
        app,
        "POST",
        "/v1/approvals",
        {"binding": binding(), "validity": validity()},
    )
    aid = created["id"]
    call(app, "POST", f"/v1/approvals/{aid}/entries", {})
    return aid


def test_live_http_pep_sequence_consumes_before_callback(live_http, tmp_path):
    base_url, app = live_http
    aid = _approved(app)
    client = ApprovalHTTPClient(base_url, _token(tmp_path))
    order = []
    result = ProtectedActionHarness(client).execute(
        aid,
        DIGEST,
        lambda claim: (
            order.append("decision"),
            {
                "effect": "ALLOW",
                "decision_id": "decision:live",
                "request_digest": DIGEST,
            },
        )[1],
        lambda: (order.append("side-effect"), "dry-run-only")[1],
    )
    assert result == "dry-run-only"
    assert order == ["decision", "side-effect"]
    retry = client.consume(aid, DIGEST, "decision:live")
    assert retry["status"] == "consumed"
    assert retry["idempotent"] is True
    assert retry["request_digest"] == DIGEST
    status, claim = call(app, "GET", f"/v1/approvals/{aid}/claim")
    assert status == 200
    assert claim["consumed"] is True
    assert claim["valid_now"] is False


def test_live_http_conflict_and_spent_approval_prevent_side_effect(live_http, tmp_path):
    base_url, app = live_http
    aid = _approved(app)
    client = ApprovalHTTPClient(base_url, _token(tmp_path))
    ProtectedActionHarness(client).execute(
        aid,
        DIGEST,
        lambda _claim: {
            "effect": "ALLOW",
            "decision_id": "decision:live",
            "request_digest": DIGEST,
        },
        lambda: "first",
    )
    effects = []
    with pytest.raises(ApprovalProtocolError, match="conflict"):
        client.consume(aid, OTHER, "decision:other")
    with pytest.raises(ApprovalProtocolError, match="not valid for use"):
        ProtectedActionHarness(client).execute(
            aid,
            DIGEST,
            lambda _claim: pytest.fail("must not decide"),
            lambda: effects.append("called"),
        )
    assert effects == []


def test_live_http_unavailability_prevents_side_effect(tmp_path):
    client = ApprovalHTTPClient("http://127.0.0.1:1", _token(tmp_path), timeout_seconds=0.2)
    effects = []
    with pytest.raises(ApprovalProtocolError, match="unreachable"):
        ProtectedActionHarness(client).execute(
            "approval-missing",
            DIGEST,
            allow,
            lambda: effects.append("called"),
        )
    assert effects == []