Kubectl MCP Tool

"""Tests for the MCP Authorization module."""

import os
import pytest
from unittest.mock import patch, MagicMock


class TestAuthConfig:
    """Tests for authentication configuration."""

    def test_auth_disabled_by_default(self):
        """Test that auth is disabled by default."""
        with patch.dict(os.environ, {}, clear=True):
            from kubectl_mcp_tool.auth.config import get_auth_config
            config = get_auth_config()
            assert config.enabled is False

    def test_auth_enabled_via_env(self):
        """Test enabling auth via environment variable."""
        with patch.dict(os.environ, {"MCP_AUTH_ENABLED": "true"}, clear=True):
            from kubectl_mcp_tool.auth.config import get_auth_config
            config = get_auth_config()
            assert config.enabled is True

    def test_auth_issuer_from_env(self):
        """Test issuer URL from environment."""
        with patch.dict(os.environ, {
            "MCP_AUTH_ENABLED": "true",
            "MCP_AUTH_ISSUER": "https://auth.example.com"
        }, clear=True):
            from kubectl_mcp_tool.auth.config import get_auth_config
            config = get_auth_config()
            assert config.issuer_url == "https://auth.example.com"

    def test_auth_audience_default(self):
        """Test default audience value."""
        with patch.dict(os.environ, {}, clear=True):
            from kubectl_mcp_tool.auth.config import get_auth_config
            config = get_auth_config()
            assert config.audience == "kubectl-mcp-server"

    def test_auth_audience_custom(self):
        """Test custom audience value."""
        with patch.dict(os.environ, {
            "MCP_AUTH_AUDIENCE": "custom-audience"
        }, clear=True):
            from kubectl_mcp_tool.auth.config import get_auth_config
            config = get_auth_config()
            assert config.audience == "custom-audience"

    def test_auth_required_scopes_default(self):
        """Test default required scopes."""
        with patch.dict(os.environ, {}, clear=True):
            from kubectl_mcp_tool.auth.config import get_auth_config
            config = get_auth_config()
            assert "mcp:tools" in config.required_scopes

    def test_auth_required_scopes_custom(self):
        """Test custom required scopes."""
        with patch.dict(os.environ, {
            "MCP_AUTH_REQUIRED_SCOPES": "mcp:read,mcp:write"
        }, clear=True):
            from kubectl_mcp_tool.auth.config import get_auth_config
            config = get_auth_config()
            assert "mcp:read" in config.required_scopes
            assert "mcp:write" in config.required_scopes

    def test_effective_jwks_uri_derived(self):
        """Test JWKS URI is derived from issuer."""
        with patch.dict(os.environ, {
            "MCP_AUTH_ISSUER": "https://auth.example.com"
        }, clear=True):
            from kubectl_mcp_tool.auth.config import get_auth_config
            config = get_auth_config()
            assert config.effective_jwks_uri == "https://auth.example.com/.well-known/jwks.json"

    def test_effective_jwks_uri_explicit(self):
        """Test explicit JWKS URI takes precedence."""
        with patch.dict(os.environ, {
            "MCP_AUTH_ISSUER": "https://auth.example.com",
            "MCP_AUTH_JWKS_URI": "https://custom.example.com/jwks"
        }, clear=True):
            from kubectl_mcp_tool.auth.config import get_auth_config
            config = get_auth_config()
            assert config.effective_jwks_uri == "https://custom.example.com/jwks"

    def test_config_validation_disabled(self):
        """Test validation passes when auth is disabled."""
        with patch.dict(os.environ, {}, clear=True):
            from kubectl_mcp_tool.auth.config import get_auth_config
            config = get_auth_config()
            assert config.validate() is True

    def test_config_validation_enabled_without_issuer(self):
        """Test validation fails when enabled without issuer."""
        with patch.dict(os.environ, {
            "MCP_AUTH_ENABLED": "true"
        }, clear=True):
            from kubectl_mcp_tool.auth.config import get_auth_config
            config = get_auth_config()
            assert config.validate() is False

    def test_config_validation_enabled_with_issuer(self):
        """Test validation passes when properly configured."""
        with patch.dict(os.environ, {
            "MCP_AUTH_ENABLED": "true",
            "MCP_AUTH_ISSUER": "https://auth.example.com"
        }, clear=True):
            from kubectl_mcp_tool.auth.config import get_auth_config
            config = get_auth_config()
            assert config.validate() is True


class TestMCPScopes:
    """Tests for MCP scope definitions."""

    def test_all_scopes_returns_list(self):
        """Test all_scopes returns a list."""
        from kubectl_mcp_tool.auth.scopes import MCPScopes
        scopes = MCPScopes.all_scopes()
        assert isinstance(scopes, list)
        assert len(scopes) > 0

    def test_scope_values(self):
        """Test scope enum values."""
        from kubectl_mcp_tool.auth.scopes import MCPScopes
        assert MCPScopes.READ.value == "mcp:read"
        assert MCPScopes.WRITE.value == "mcp:write"
        assert MCPScopes.ADMIN.value == "mcp:admin"
        assert MCPScopes.TOOLS.value == "mcp:tools"
        assert MCPScopes.HELM.value == "mcp:helm"

    def test_read_scopes(self):
        """Test read-only scopes."""
        from kubectl_mcp_tool.auth.scopes import MCPScopes
        scopes = MCPScopes.read_scopes()
        assert "mcp:read" in scopes
        assert "mcp:diagnostics" in scopes

    def test_admin_scopes_includes_all(self):
        """Test admin scopes include all scopes."""
        from kubectl_mcp_tool.auth.scopes import MCPScopes
        admin_scopes = MCPScopes.admin_scopes()
        all_scopes = MCPScopes.all_scopes()
        assert set(admin_scopes) == set(all_scopes)


class TestToolScopes:
    """Tests for tool-to-scope mappings."""

    def test_get_required_scopes_read_tool(self):
        """Test read tools require read scope."""
        from kubectl_mcp_tool.auth.scopes import get_required_scopes, MCPScopes
        scopes = get_required_scopes("get_pods")
        assert MCPScopes.READ.value in scopes

    def test_get_required_scopes_write_tool(self):
        """Test write tools require write scope."""
        from kubectl_mcp_tool.auth.scopes import get_required_scopes, MCPScopes
        scopes = get_required_scopes("scale_deployment")
        assert MCPScopes.WRITE.value in scopes

    def test_get_required_scopes_admin_tool(self):
        """Test admin tools require admin scope."""
        from kubectl_mcp_tool.auth.scopes import get_required_scopes, MCPScopes
        scopes = get_required_scopes("drain_node")
        assert MCPScopes.ADMIN.value in scopes

    def test_get_required_scopes_helm_tool(self):
        """Test helm tools require helm scope."""
        from kubectl_mcp_tool.auth.scopes import get_required_scopes, MCPScopes
        scopes = get_required_scopes("helm_list_releases")
        assert MCPScopes.HELM.value in scopes

    def test_get_required_scopes_unknown_tool(self):
        """Test unknown tools require default scope."""
        from kubectl_mcp_tool.auth.scopes import get_required_scopes, MCPScopes
        scopes = get_required_scopes("unknown_tool_xyz")
        assert MCPScopes.TOOLS.value in scopes

    def test_has_required_scopes_with_tools_scope(self):
        """Test mcp:tools grants access to all tools."""
        from kubectl_mcp_tool.auth.scopes import has_required_scopes, MCPScopes
        token_scopes = {MCPScopes.TOOLS.value}
        assert has_required_scopes(token_scopes, "get_pods") is True
        assert has_required_scopes(token_scopes, "drain_node") is True

    def test_has_required_scopes_with_admin_scope(self):
        """Test mcp:admin grants access to all tools."""
        from kubectl_mcp_tool.auth.scopes import has_required_scopes, MCPScopes
        token_scopes = {MCPScopes.ADMIN.value}
        assert has_required_scopes(token_scopes, "get_pods") is True
        assert has_required_scopes(token_scopes, "drain_node") is True

    def test_has_required_scopes_read_only(self):
        """Test read scope grants access to read tools only."""
        from kubectl_mcp_tool.auth.scopes import has_required_scopes, MCPScopes
        token_scopes = {MCPScopes.READ.value}
        assert has_required_scopes(token_scopes, "get_pods") is True
        assert has_required_scopes(token_scopes, "scale_deployment") is False

    def test_has_required_scopes_write_not_admin(self):
        """Test write scope doesn't grant admin access."""
        from kubectl_mcp_tool.auth.scopes import has_required_scopes, MCPScopes
        token_scopes = {MCPScopes.WRITE.value}
        assert has_required_scopes(token_scopes, "scale_deployment") is True
        assert has_required_scopes(token_scopes, "drain_node") is False


class TestAuthVerifier:
    """Tests for authentication verifier creation."""

    def test_create_verifier_auth_disabled(self):
        """Test verifier is None when auth disabled."""
        with patch.dict(os.environ, {}, clear=True):
            from kubectl_mcp_tool.auth import get_auth_config, create_auth_verifier
            config = get_auth_config()
            verifier = create_auth_verifier(config)
            assert verifier is None

    def test_create_verifier_missing_fastmcp_auth(self):
        """Test graceful handling when FastMCP auth not available."""
        with patch.dict(os.environ, {
            "MCP_AUTH_ENABLED": "true",
            "MCP_AUTH_ISSUER": "https://auth.example.com"
        }, clear=True):
            from kubectl_mcp_tool.auth import get_auth_config, create_auth_verifier
            config = get_auth_config()

            # Mock the import to fail
            with patch.dict('sys.modules', {'fastmcp.server.auth': None}):
                verifier = create_auth_verifier(config)
                # Should return None gracefully when auth module not available
                assert verifier is None


class TestMCPServerAuth:
    """Tests for MCP server with authentication."""

    def test_server_initializes_without_auth(self):
        """Test server initializes with auth disabled."""
        with patch.dict(os.environ, {}, clear=True):
            from kubectl_mcp_tool.mcp_server import MCPServer
            server = MCPServer(name="test-server")
            assert server.auth_config.enabled is False

    def test_server_with_auth_enabled(self):
        """Test server with auth enabled loads config."""
        with patch.dict(os.environ, {
            "MCP_AUTH_ENABLED": "true",
            "MCP_AUTH_ISSUER": "https://auth.example.com"
        }, clear=True):
            from kubectl_mcp_tool.mcp_server import MCPServer
            server = MCPServer(name="test-server")
            assert server.auth_config.enabled is True
            assert server.auth_config.issuer_url == "https://auth.example.com"