"""Tests for the authorization provider protocol, adapter, and configuration.

Phase 0 covers scaffolding only (no behavior change at ``enabled: false``).
These tests verify:
- Protocol conformance and ``@runtime_checkable`` isinstance checks.
- Principal / AuthzRequest / AuthzDecision dataclass construction.
- GuardrailAuthorizationAdapter request mapping and decision conversion.
- AuthorizationConfig defaults, singleton load/reset, and AppConfig wiring.
"""

from __future__ import annotations

import asyncio

import pytest

from deerflow.authz.adapter import GuardrailAuthorizationAdapter
from deerflow.authz.provider import (
    AuthorizationProvider,
    AuthzDecision,
    AuthzReason,
    AuthzRequest,
    Principal,
)
from deerflow.config.app_config import AppConfig
from deerflow.config.authorization_config import (
    AuthorizationConfig,
    get_authorization_config,
    load_authorization_config_from_dict,
    reset_authorization_config,
)
from deerflow.config.sandbox_config import SandboxConfig
from deerflow.guardrails.provider import GuardrailDecision, GuardrailProvider, GuardrailRequest

# --- Test providers ---


class _AllowAllProvider:
    """Provider that allows everything."""

    name = "allow-all"

    def authorize(self, request: AuthzRequest) -> AuthzDecision:
        return AuthzDecision(allow=True, reasons=[AuthzReason(code="test.allowed", message="allow-all")])

    async def aauthorize(self, request: AuthzRequest) -> AuthzDecision:
        return self.authorize(request)

    def filter_resources(self, principal: Principal, resource_type: str, candidates: list[str]) -> list[str]:
        return list(candidates)


class _DenyAllProvider:
    """Provider that denies everything."""

    name = "deny-all"

    def authorize(self, request: AuthzRequest) -> AuthzDecision:
        return AuthzDecision(allow=False, reasons=[AuthzReason(code="test.denied", message="deny-all")], policy_id="test.deny.v1")

    async def aauthorize(self, request: AuthzRequest) -> AuthzDecision:
        return self.authorize(request)

    def filter_resources(self, principal: Principal, resource_type: str, candidates: list[str]) -> list[str]:
        return []


class _FilterByDenylistProvider:
    """Provider whose filter_resources removes a denylist, regardless of authorize()."""

    name = "denylist-filter"

    def __init__(self, *, denied: list[str] | None = None):
        self._denied = set(denied) if denied else set()

    def authorize(self, request: AuthzRequest) -> AuthzDecision:
        if request.target in self._denied:
            return AuthzDecision(allow=False, reasons=[AuthzReason(code="test.denied", message=f"'{request.target}' is denied")])
        return AuthzDecision(allow=True, reasons=[AuthzReason(code="test.allowed")])

    async def aauthorize(self, request: AuthzRequest) -> AuthzDecision:
        return self.authorize(request)

    def filter_resources(self, principal: Principal, resource_type: str, candidates: list[str]) -> list[str]:
        return [c for c in candidates if c not in self._denied]


# --- Protocol conformance ---


class TestProtocolConformance:
    """Verify the @runtime_checkable Protocol recognizes concrete providers."""

    def test_allow_all_is_authorization_provider(self):
        assert isinstance(_AllowAllProvider(), AuthorizationProvider)

    def test_deny_all_is_authorization_provider(self):
        assert isinstance(_DenyAllProvider(), AuthorizationProvider)

    def test_plain_object_without_methods_is_not_provider(self):
        class _NotAProvider:
            pass

        assert not isinstance(_NotAProvider(), AuthorizationProvider)

    def test_provider_without_filter_resources_is_not_provider(self):
        """filter_resources is a required Protocol method.

        A provider that only implements name/authorize/aauthorize but omits
        filter_resources must NOT pass isinstance — otherwise it would be
        silently accepted as an AuthorizationProvider and Layer 1 would
        get None (fail-open) when calling filter_resources.
        """

        class _NoFilterMethod:
            name = "no-filter"

            def authorize(self, request: AuthzRequest) -> AuthzDecision:
                return AuthzDecision(allow=True)

            async def aauthorize(self, request: AuthzRequest) -> AuthzDecision:
                return self.authorize(request)

        assert not isinstance(_NoFilterMethod(), AuthorizationProvider)


# --- Dataclass construction ---


class TestDataclasses:
    """Verify Principal / AuthzRequest / AuthzDecision construction."""

    def test_principal_defaults(self):
        p = Principal()
        assert p.user_id is None
        assert p.role is None
        assert p.oauth_provider is None
        assert p.oauth_id is None
        assert p.channel_user_id is None
        assert p.is_internal is False
        assert p.attributes == {}

    def test_principal_with_fields(self):
        p = Principal(user_id="u1", role="admin", oauth_provider="github", oauth_id="gh-123", is_internal=True)
        assert p.user_id == "u1"
        assert p.role == "admin"
        assert p.oauth_provider == "github"
        assert p.oauth_id == "gh-123"
        assert p.is_internal is True

    def test_authz_request(self):
        p = Principal(user_id="u1", role="user")
        req = AuthzRequest(principal=p, resource="tool", action="call", target="bash")
        assert req.principal.user_id == "u1"
        assert req.resource == "tool"
        assert req.action == "call"
        assert req.target == "bash"
        assert req.context == {}

    def test_authz_request_with_context(self):
        p = Principal(user_id="u1")
        req = AuthzRequest(principal=p, resource="tool", action="call", target="write_file", context={"thread_id": "t1"})
        assert req.context["thread_id"] == "t1"

    def test_authz_decision_defaults(self):
        d = AuthzDecision(allow=True)
        assert d.allow is True
        assert d.reasons == []
        assert d.policy_id is None
        assert d.metadata == {}

    def test_authz_decision_with_reasons(self):
        d = AuthzDecision(allow=False, reasons=[AuthzReason(code="denied", message="no access")], policy_id="p1")
        assert d.allow is False
        assert len(d.reasons) == 1
        assert d.reasons[0].code == "denied"
        assert d.reasons[0].message == "no access"
        assert d.policy_id == "p1"


# --- filter_resources ---


class TestFilterResources:
    """Verify the Layer 1 batch filter."""

    def test_allow_all_returns_all(self):
        provider = _AllowAllProvider()
        result = provider.filter_resources(Principal(role="user"), "tool", ["bash", "web_search", "read_file"])
        assert result == ["bash", "web_search", "read_file"]

    def test_deny_all_returns_empty(self):
        provider = _DenyAllProvider()
        result = provider.filter_resources(Principal(role="user"), "tool", ["bash", "web_search"])
        assert result == []

    def test_denylist_filter_removes_denied(self):
        provider = _FilterByDenylistProvider(denied=["bash", "write_file"])
        result = provider.filter_resources(Principal(role="user"), "tool", ["bash", "web_search", "write_file", "read_file"])
        assert result == ["web_search", "read_file"]


# --- GuardrailAuthorizationAdapter ---


def _make_guardrail_request(
    *,
    tool_name: str = "bash",
    tool_input: dict | None = None,
    user_id: str | None = "u1",
    user_role: str | None = "user",
    oauth_provider: str | None = None,
    oauth_id: str | None = None,
    channel_user_id: str | None = None,
    thread_id: str | None = "t1",
    is_subagent: bool = False,
    agent_id: str | None = None,
    timestamp: str = "",
    is_internal: bool = False,
    authz_attributes: dict | None = None,
) -> GuardrailRequest:
    return GuardrailRequest(
        tool_name=tool_name,
        tool_input=tool_input or {},
        user_id=user_id,
        user_role=user_role,
        oauth_provider=oauth_provider,
        oauth_id=oauth_id,
        channel_user_id=channel_user_id,
        thread_id=thread_id,
        is_subagent=is_subagent,
        agent_id=agent_id,
        timestamp=timestamp,
        is_internal=is_internal,
        authz_attributes=authz_attributes if authz_attributes is not None else {},
    )


class TestGuardrailAuthorizationAdapter:
    """Verify the adapter maps between Guardrail and Authz request/decision types."""

    def test_adapter_name(self):
        adapter = GuardrailAuthorizationAdapter(_AllowAllProvider())
        assert adapter.name == "authorization"

    def test_adapter_is_guardrail_provider(self):
        """The adapter must satisfy the GuardrailProvider Protocol so existing
        GuardrailMiddleware can enforce authz decisions without a new middleware."""
        adapter = GuardrailAuthorizationAdapter(_AllowAllProvider())
        assert isinstance(adapter, GuardrailProvider)

    def test_evaluate_allow(self):
        adapter = GuardrailAuthorizationAdapter(_AllowAllProvider())
        gr_req = _make_guardrail_request(tool_name="web_search")
        decision = adapter.evaluate(gr_req)
        assert decision.allow is True
        assert len(decision.reasons) == 1
        assert decision.reasons[0].code == "test.allowed"

    def test_evaluate_deny(self):
        adapter = GuardrailAuthorizationAdapter(_DenyAllProvider())
        gr_req = _make_guardrail_request(tool_name="bash")
        decision = adapter.evaluate(gr_req)
        assert decision.allow is False
        assert decision.policy_id == "test.deny.v1"

    def test_evaluate_maps_complete_principal_identity(self):
        """Every GuardrailRequest identity field flows into Principal."""
        captured: list[AuthzRequest] = []

        class _CapturingProvider:
            name = "capturing"

            def authorize(self, request: AuthzRequest) -> AuthzDecision:
                captured.append(request)
                return AuthzDecision(allow=True, reasons=[AuthzReason(code="ok")])

            async def aauthorize(self, request: AuthzRequest) -> AuthzDecision:
                return self.authorize(request)

            def filter_resources(self, principal: Principal, resource_type: str, candidates: list[str]) -> list[str]:
                return list(candidates)

        adapter = GuardrailAuthorizationAdapter(_CapturingProvider())
        gr_req = _make_guardrail_request(
            user_id="user-42",
            user_role="admin",
            oauth_provider="github",
            oauth_id="gh-42",
            channel_user_id="channel-42",
            is_internal=True,
            authz_attributes={"department": "engineering"},
            tool_name="write_file",
        )
        adapter.evaluate(gr_req)

        assert len(captured) == 1
        authz_req = captured[0]
        assert authz_req.principal.user_id == "user-42"
        assert authz_req.principal.role == "admin"
        assert authz_req.principal.oauth_provider == "github"
        assert authz_req.principal.oauth_id == "gh-42"
        assert authz_req.principal.channel_user_id == "channel-42"
        assert authz_req.principal.is_internal is True
        assert authz_req.principal.attributes == {"department": "engineering"}
        assert authz_req.resource == "tool"
        assert authz_req.action == "call"
        assert authz_req.target == "write_file"

    def test_evaluate_maps_is_internal_true(self):
        """is_internal=True on GuardrailRequest maps to Principal.is_internal=True."""
        captured: list[AuthzRequest] = []

        class _CapturingProvider:
            name = "capturing"

            def authorize(self, request: AuthzRequest) -> AuthzDecision:
                captured.append(request)
                return AuthzDecision(allow=True)

            async def aauthorize(self, request: AuthzRequest) -> AuthzDecision:
                return self.authorize(request)

            def filter_resources(self, principal: Principal, resource_type: str, candidates: list[str]) -> list[str]:
                return list(candidates)

        adapter = GuardrailAuthorizationAdapter(_CapturingProvider())
        gr_req = _make_guardrail_request(user_role="user", is_internal=True)
        adapter.evaluate(gr_req)
        assert captured[0].principal.is_internal is True

    def test_evaluate_maps_is_internal_false(self):
        """is_internal=False (default) maps to Principal.is_internal=False."""
        captured: list[AuthzRequest] = []

        class _CapturingProvider:
            name = "capturing"

            def authorize(self, request: AuthzRequest) -> AuthzDecision:
                captured.append(request)
                return AuthzDecision(allow=True)

            async def aauthorize(self, request: AuthzRequest) -> AuthzDecision:
                return self.authorize(request)

            def filter_resources(self, principal: Principal, resource_type: str, candidates: list[str]) -> list[str]:
                return list(candidates)

        adapter = GuardrailAuthorizationAdapter(_CapturingProvider())
        adapter.evaluate(_make_guardrail_request(user_role="user"))
        assert captured[0].principal.is_internal is False

    def test_evaluate_applies_default_role_when_missing(self):
        """Adapter uses default_role when user_role is absent."""
        captured: list[AuthzRequest] = []

        class _CapturingProvider:
            name = "capturing"

            def authorize(self, request: AuthzRequest) -> AuthzDecision:
                captured.append(request)
                return AuthzDecision(allow=True)

            async def aauthorize(self, request: AuthzRequest) -> AuthzDecision:
                return self.authorize(request)

            def filter_resources(self, principal: Principal, resource_type: str, candidates: list[str]) -> list[str]:
                return list(candidates)

        adapter = GuardrailAuthorizationAdapter(_CapturingProvider(), default_role="guest")
        gr_req = _make_guardrail_request(user_role=None)
        adapter.evaluate(gr_req)
        assert captured[0].principal.role == "guest"

    def test_evaluate_preserves_unknown_role(self):
        """Unknown non-empty role is NOT replaced by default_role."""
        captured: list[AuthzRequest] = []

        class _CapturingProvider:
            name = "capturing"

            def authorize(self, request: AuthzRequest) -> AuthzDecision:
                captured.append(request)
                return AuthzDecision(allow=True)

            async def aauthorize(self, request: AuthzRequest) -> AuthzDecision:
                return self.authorize(request)

            def filter_resources(self, principal: Principal, resource_type: str, candidates: list[str]) -> list[str]:
                return list(candidates)

        adapter = GuardrailAuthorizationAdapter(_CapturingProvider(), default_role="user")
        gr_req = _make_guardrail_request(user_role="editor")
        adapter.evaluate(gr_req)
        assert captured[0].principal.role == "editor"

    def test_evaluate_sync_async_same_principal(self):
        """Sync and async paths produce identical Principal fields."""
        captured_sync: list[AuthzRequest] = []
        captured_async: list[AuthzRequest] = []

        class _SyncCapture:
            name = "sync-capture"

            def authorize(self, request: AuthzRequest) -> AuthzDecision:
                captured_sync.append(request)
                return AuthzDecision(allow=True)

            async def aauthorize(self, request: AuthzRequest) -> AuthzDecision:
                return self.authorize(request)

            def filter_resources(self, principal: Principal, resource_type: str, candidates: list[str]) -> list[str]:
                return list(candidates)

        class _AsyncCapture:
            name = "async-capture"

            def authorize(self, request: AuthzRequest) -> AuthzDecision:
                captured_async.append(request)
                return AuthzDecision(allow=True)

            async def aauthorize(self, request: AuthzRequest) -> AuthzDecision:
                captured_async.append(request)
                return AuthzDecision(allow=True)

            def filter_resources(self, principal: Principal, resource_type: str, candidates: list[str]) -> list[str]:
                return list(candidates)

        gr_req = _make_guardrail_request(
            user_id="user-42",
            user_role="admin",
            oauth_provider="github",
            oauth_id="gh-42",
            channel_user_id="channel-42",
            is_internal=True,
            authz_attributes={"department": "engineering"},
        )
        GuardrailAuthorizationAdapter(_SyncCapture()).evaluate(gr_req)
        asyncio.run(GuardrailAuthorizationAdapter(_AsyncCapture()).aevaluate(gr_req))

        p1 = captured_sync[0].principal
        p2 = captured_async[0].principal
        assert p1 == p2

    def test_evaluate_maps_context_fields(self):
        """Verify thread_id, tool_input, and is_subagent flow into AuthzRequest.context."""
        captured: list[AuthzRequest] = []

        class _CapturingProvider:
            name = "capturing"

            def authorize(self, request: AuthzRequest) -> AuthzDecision:
                captured.append(request)
                return AuthzDecision(allow=True)

            async def aauthorize(self, request: AuthzRequest) -> AuthzDecision:
                return self.authorize(request)

            def filter_resources(self, principal: Principal, resource_type: str, candidates: list[str]) -> list[str]:
                return list(candidates)

        adapter = GuardrailAuthorizationAdapter(_CapturingProvider())
        gr_req = _make_guardrail_request(
            tool_name="write_file",
            tool_input={"path": "/tmp/test.txt"},
            thread_id="thread-99",
            is_subagent=True,
            agent_id="passport-42",
            timestamp="2026-07-13T00:00:00Z",
        )
        adapter.evaluate(gr_req)

        ctx = captured[0].context
        assert ctx["thread_id"] == "thread-99"
        assert ctx["tool_input"] == {"path": "/tmp/test.txt"}
        assert ctx["is_subagent"] is True
        assert ctx["agent_id"] == "passport-42"
        assert ctx["timestamp"] == "2026-07-13T00:00:00Z"

    def test_custom_resource_type_and_action(self):
        """The adapter can be configured for non-tool resource types."""
        captured: list[AuthzRequest] = []

        class _CapturingProvider:
            name = "capturing"

            def authorize(self, request: AuthzRequest) -> AuthzDecision:
                captured.append(request)
                return AuthzDecision(allow=True)

            async def aauthorize(self, request: AuthzRequest) -> AuthzDecision:
                return self.authorize(request)

            def filter_resources(self, principal: Principal, resource_type: str, candidates: list[str]) -> list[str]:
                return list(candidates)

        adapter = GuardrailAuthorizationAdapter(_CapturingProvider(), resource_type="model", action="use")
        adapter.evaluate(_make_guardrail_request(tool_name="claude-sonnet-4-6"))

        assert captured[0].resource == "model"
        assert captured[0].action == "use"

    def test_aevaluate_allow(self):
        adapter = GuardrailAuthorizationAdapter(_AllowAllProvider())
        gr_req = _make_guardrail_request(tool_name="web_search")
        decision = asyncio.run(adapter.aevaluate(gr_req))
        assert decision.allow is True

    def test_aevaluate_deny(self):
        adapter = GuardrailAuthorizationAdapter(_DenyAllProvider())
        gr_req = _make_guardrail_request(tool_name="bash")
        decision = asyncio.run(adapter.aevaluate(gr_req))
        assert decision.allow is False

    def test_decision_conversion_preserves_metadata(self):
        class _MetadataProvider:
            name = "metadata"

            def authorize(self, request: AuthzRequest) -> AuthzDecision:
                return AuthzDecision(
                    allow=True,
                    reasons=[AuthzReason(code="ok", message="allowed by policy X")],
                    policy_id="rbac.v1",
                    metadata={"rule_id": "rule-42"},
                )

            async def aauthorize(self, request: AuthzRequest) -> AuthzDecision:
                return self.authorize(request)

            def filter_resources(self, principal: Principal, resource_type: str, candidates: list[str]) -> list[str]:
                return list(candidates)

        adapter = GuardrailAuthorizationAdapter(_MetadataProvider())
        decision = adapter.evaluate(_make_guardrail_request())

        assert isinstance(decision, GuardrailDecision)
        assert decision.policy_id == "rbac.v1"
        assert decision.metadata == {"rule_id": "rule-42"}
        assert decision.reasons[0].message == "allowed by policy X"

    def test_evaluate_propagates_provider_exception(self):
        """Provider exceptions propagate to the caller (sync).

        The adapter intentionally does not catch provider exceptions.
        GuardrailMiddleware's wrap_tool_call applies fail_closed semantics
        (deny on error when fail_closed=True). Catching here would duplicate
        that logic and risk divergent behavior between layers.
        """

        class _ExplodingProvider:
            name = "exploding"

            def authorize(self, request: AuthzRequest) -> AuthzDecision:
                raise RuntimeError("provider crashed")

            async def aauthorize(self, request: AuthzRequest) -> AuthzDecision:
                raise RuntimeError("provider crashed")

            def filter_resources(self, principal: Principal, resource_type: str, candidates: list[str]) -> list[str]:
                return list(candidates)

        adapter = GuardrailAuthorizationAdapter(_ExplodingProvider())
        with pytest.raises(RuntimeError, match="provider crashed"):
            adapter.evaluate(_make_guardrail_request())

    def test_aevaluate_propagates_provider_exception(self):
        """Provider exceptions propagate to the caller (async).

        Same rationale as the sync variant: GuardrailMiddleware's
        awrap_tool_call handles fail_closed.
        """

        class _ExplodingProvider:
            name = "exploding"

            def authorize(self, request: AuthzRequest) -> AuthzDecision:
                raise RuntimeError("provider crashed")

            async def aauthorize(self, request: AuthzRequest) -> AuthzDecision:
                raise RuntimeError("provider crashed")

            def filter_resources(self, principal: Principal, resource_type: str, candidates: list[str]) -> list[str]:
                return list(candidates)

        adapter = GuardrailAuthorizationAdapter(_ExplodingProvider())
        with pytest.raises(RuntimeError, match="provider crashed"):
            asyncio.run(adapter.aevaluate(_make_guardrail_request()))


# --- Configuration ---


class TestAuthorizationConfig:
    """Verify config defaults, singleton behavior, and AppConfig wiring."""

    def teardown_method(self):
        reset_authorization_config()

    def test_defaults(self):
        config = AuthorizationConfig()
        assert config.enabled is False
        assert config.fail_closed is True
        assert config.default_role == "user"
        assert config.provider is None

    def test_get_returns_defaults_when_not_loaded(self):
        reset_authorization_config()
        config = get_authorization_config()
        assert config.enabled is False

    def test_load_from_dict(self):
        config = load_authorization_config_from_dict(
            {
                "enabled": True,
                "default_role": "guest",
                "provider": {
                    "use": "my_package:MyProvider",
                    "config": {"roles": {"admin": {}}},
                },
            }
        )
        assert config.enabled is True
        assert config.default_role == "guest"
        assert config.provider is not None
        assert config.provider.use == "my_package:MyProvider"
        assert config.provider.config == {"roles": {"admin": {}}}

    def test_singleton_persistence(self):
        load_authorization_config_from_dict({"enabled": True})
        config2 = get_authorization_config()
        assert config2.enabled is True

    def test_reset_clears_singleton(self):
        load_authorization_config_from_dict({"enabled": True})
        reset_authorization_config()
        config = get_authorization_config()
        assert config.enabled is False

    def test_app_config_has_authorization_field(self):
        """Verify AuthorizationConfig is wired into AppConfig with correct defaults."""
        app_config = AppConfig(sandbox=SandboxConfig(use="test"))
        assert hasattr(app_config, "authorization")
        assert app_config.authorization.enabled is False
        assert app_config.authorization.fail_closed is True
        assert app_config.authorization.default_role == "user"

    def test_app_config_load_propagates_to_singleton(self):
        """Verify _apply_singleton_configs populates the authorization singleton.

        model_validate alone does NOT call _apply_singleton_configs (that runs
        only in from_file). We call it directly to verify the wiring line
        ``load_authorization_config_from_dict(config.authorization.model_dump())``
        actually populates the singleton — deleting that line should fail this test.
        """
        reset_authorization_config()
        validated = AppConfig.model_validate(
            {
                "sandbox": {"use": "test"},
                "authorization": {
                    "enabled": True,
                    "default_role": "operator",
                },
            }
        )
        # Drive the singleton wiring the same way from_file does.
        AppConfig._apply_singleton_configs(validated, acp_agents={})

        singleton = get_authorization_config()
        assert singleton.enabled is True
        assert singleton.default_role == "operator"