"""Unit tests for auth service: backend auth API calls and server-owned auth."""

from unittest.mock import AsyncMock, MagicMock, patch

import httpx
import pytest

from src.models.cloud_access import AuthResponse, S3Access
from src.services import auth
from src.services.auth import (
    _auth_request,
    _fetch_auth,
    bootstrap_server_auth,
    get_delta_credentials,
    get_gcp_access_token,
    get_server_access_token,
    get_server_auth0_token,
    get_valid_server_auth,
)
from src.services.prometheus import get_prometheus_client_for_cluster
from src.utils.errors import AuthServiceError, NetworkError


def _mock_httpx_client(
    response: httpx.Response | None = None, side_effect: Exception | None = None
):
    mock_client = MagicMock()
    if side_effect is not None:
        mock_client.request = AsyncMock(side_effect=side_effect)
    else:
        mock_client.request = AsyncMock(return_value=response)
    mock_cm = MagicMock()
    mock_cm.__aenter__ = AsyncMock(return_value=mock_client)
    mock_cm.__aexit__ = AsyncMock(return_value=None)
    return mock_cm


class TestAuthRequest:
    @pytest.mark.asyncio
    async def test_returns_json_on_success(self):
        resp = httpx.Response(200, json={"access_token": "jwt", "expires_in": 3600})
        with patch.object(auth, "get_auth_service_url", return_value="https://auth.example.com"):
            with patch(
                "src.services.auth.httpx.AsyncClient",
                return_value=_mock_httpx_client(resp),
            ):
                data = await _auth_request("GET", "/auth/me", "bearer-token")
        assert data["access_token"] == "jwt"
        assert data["expires_in"] == 3600

    @pytest.mark.asyncio
    async def test_does_not_set_authorization_when_token_missing(self):
        resp = httpx.Response(200, json={"ok": True})
        with patch.object(auth, "get_auth_service_url", return_value="https://auth.example.com"):
            with patch(
                "src.services.auth.httpx.AsyncClient",
                return_value=_mock_httpx_client(resp),
            ) as mock_client_cls:
                await _auth_request(
                    "POST", "/client/auth", None, json_body={"github_token": "ghp_x"}
                )
        request_call = mock_client_cls.return_value.__aenter__.return_value.request
        _, kwargs = request_call.await_args
        assert "Authorization" not in kwargs["headers"]

    @pytest.mark.asyncio
    async def test_401_raises_auth_service_error(self):
        resp = httpx.Response(401, json={})
        with patch.object(auth, "get_auth_service_url", return_value="https://auth.example.com"):
            with patch(
                "src.services.auth.httpx.AsyncClient",
                return_value=_mock_httpx_client(resp),
            ):
                with pytest.raises(AuthServiceError) as exc_info:
                    await _auth_request("GET", "/x", "token")
                assert exc_info.value.status_code == 401
                assert "Invalid or expired" in exc_info.value.message

    @pytest.mark.asyncio
    async def test_403_raises_auth_service_error(self):
        resp = httpx.Response(403, json={})
        with patch.object(auth, "get_auth_service_url", return_value="https://auth.example.com"):
            with patch(
                "src.services.auth.httpx.AsyncClient",
                return_value=_mock_httpx_client(resp),
            ):
                with pytest.raises(AuthServiceError) as exc_info:
                    await _auth_request("GET", "/x", "token")
                assert exc_info.value.status_code == 403

    @pytest.mark.asyncio
    async def test_4xx_raises_auth_service_error(self):
        resp = httpx.Response(404, json={})
        with patch.object(auth, "get_auth_service_url", return_value="https://auth.example.com"):
            with patch(
                "src.services.auth.httpx.AsyncClient",
                return_value=_mock_httpx_client(resp),
            ):
                with pytest.raises(AuthServiceError) as exc_info:
                    await _auth_request("GET", "/x", "token")
                assert exc_info.value.status_code == 404
                assert "404" in exc_info.value.message

    @pytest.mark.asyncio
    async def test_network_error_raises_network_error(self):
        with patch.object(auth, "get_auth_service_url", return_value="https://auth.example.com"):
            with patch(
                "src.services.auth.httpx.AsyncClient",
                return_value=_mock_httpx_client(side_effect=httpx.ConnectError("unreachable")),
            ):
                with pytest.raises(NetworkError) as exc_info:
                    await _auth_request("GET", "/x", "token")
                assert (
                    "unreachable" in exc_info.value.message
                    or "Auth service" in exc_info.value.message
                )

    @pytest.mark.asyncio
    async def test_timeout_raises_auth_service_error(self):
        with patch.object(auth, "get_auth_service_url", return_value="https://auth.example.com"):
            with patch(
                "src.services.auth.httpx.AsyncClient",
                return_value=_mock_httpx_client(side_effect=httpx.TimeoutException("timed out")),
            ):
                with pytest.raises(AuthServiceError) as exc_info:
                    await _auth_request("GET", "/x", "token")
                assert (
                    "did not respond" in exc_info.value.message
                    or "timeout" in exc_info.value.message.lower()
                )

    @pytest.mark.asyncio
    async def test_invalid_json_raises_auth_service_error(self):
        resp = httpx.Response(200, text="not json")
        with patch.object(auth, "get_auth_service_url", return_value="https://auth.example.com"):
            with patch(
                "src.services.auth.httpx.AsyncClient",
                return_value=_mock_httpx_client(resp),
            ):
                with pytest.raises(AuthServiceError) as exc_info:
                    await _auth_request("GET", "/x", "token")
                assert "invalid JSON" in exc_info.value.message


class TestFetchAuth:
    @pytest.mark.asyncio
    async def test_raises_when_token_empty(self):
        with pytest.raises(AuthServiceError) as exc_info:
            await _fetch_auth("")
        assert "required" in exc_info.value.message or "empty" in exc_info.value.message

    @pytest.mark.asyncio
    async def test_raises_when_token_whitespace_only(self):
        with pytest.raises(AuthServiceError):
            await _fetch_auth("   \t  ")

    @pytest.mark.asyncio
    async def test_success_parses_full_response(self):
        resp = httpx.Response(
            200,
            json={
                "token": "jwt123",
                "expiration": "2099-03-17T09:50:27+00:00",
                "auth0_token": "auth0-jwt",
                "observabilityGlobalService": "https://obs.example.com",
                "access_key_id": "ak",
                "secret_access_key": "sk",
                "session_token": "st",
            },
        )
        with patch.object(auth, "get_auth_service_url", return_value="https://auth.example.com"):
            with patch(
                "src.services.auth.httpx.AsyncClient",
                return_value=_mock_httpx_client(resp),
            ) as mock_client_cls:
                result = await _fetch_auth("github-token")
        request_call = mock_client_cls.return_value.__aenter__.return_value.request
        request_call.assert_awaited_once()
        args, kwargs = request_call.await_args
        assert args[0] == "POST"
        assert args[1] == "https://auth.example.com/client/auth"
        assert kwargs["json"] == {"github_token": "github-token", "provider": "aws"}
        assert "Authorization" not in kwargs["headers"]
        assert result.access_token == "jwt123"
        assert result.expires_in is not None
        assert result.expires_in > 0
        assert result.auth0_token == "auth0-jwt"
        assert result.observability_global_service == "https://obs.example.com"
        assert result.s3_access is not None
        assert result.s3_access.provider == "aws"
        assert result.s3_access.access_key_id == "ak"
        assert result.s3_access.secret_access_key == "sk"
        assert result.s3_access.session_token == "st"

    @pytest.mark.asyncio
    async def test_success_with_alicloud_provider(self):
        resp = httpx.Response(
            200,
            json={
                "token": "jwt123",
                "expires_in": 3600,
                "auth0_token": "auth0-jwt",
                "observabilityGlobalService": "https://obs.example.com",
                "access_key_id": "ak",
                "secret_access_key": "sk",
                "session_token": "st",
            },
        )
        with patch.object(auth, "get_auth_service_url", return_value="https://auth.example.com"):
            with patch(
                "src.services.auth.httpx.AsyncClient",
                return_value=_mock_httpx_client(resp),
            ) as mock_client_cls:
                result = await _fetch_auth("github-token", provider="alicloud")
        request_call = mock_client_cls.return_value.__aenter__.return_value.request
        args, kwargs = request_call.await_args
        assert args[0] == "POST"
        assert args[1] == "https://auth.example.com/client/auth"
        assert kwargs["json"] == {
            "github_token": "github-token",
            "provider": "alicloud",
        }
        assert result.s3_access is not None
        assert result.s3_access.provider == "alicloud"

    @pytest.mark.asyncio
    async def test_success_parses_backend_expiration_and_provider(self):
        # Matches current /client/auth response shape: expiration + provider + token.
        resp = httpx.Response(
            200,
            json={
                "token": "jwt123",
                "expiration": "2099-03-17T09:50:27+00:00",
                "auth0_token": "auth0-jwt",
                "observability_global_service": "https://www.gs.us-west-2.aws.observability.tidbcloud.com",
                "provider": "aws",
                "access_key_id": "ak",
                "secret_access_key": "sk",
                "session_token": "st",
            },
        )
        with patch.object(auth, "get_auth_service_url", return_value="https://auth.example.com"):
            with patch(
                "src.services.auth.httpx.AsyncClient",
                return_value=_mock_httpx_client(resp),
            ):
                result = await _fetch_auth("github-token", provider="alicloud")
        assert result.expires_in is not None
        assert result.expires_in > 0
        assert result.observability_global_service == (
            "https://www.gs.us-west-2.aws.observability.tidbcloud.com"
        )
        assert result.s3_access is not None
        # Backend provider takes precedence when present.
        assert result.s3_access.provider == "aws"

    @pytest.mark.asyncio
    async def test_raises_when_provider_unsupported(self):
        with pytest.raises(AuthServiceError) as exc_info:
            await _fetch_auth("github-token", provider="azure")
        assert "provider must be one of" in exc_info.value.message

    @pytest.mark.asyncio
    async def test_success_with_gcp_provider_and_token_only(self):
        resp = httpx.Response(
            200,
            json={
                "token": "jwt123",
                "expiration": "2099-03-17T09:50:27+00:00",
                "auth0_token": "auth0-jwt",
                "observability_global_service": "https://www.gs.us-central1.gcp.observability.tidbcloud.com",
                "provider": "gcp",
            },
        )
        with patch.object(auth, "get_auth_service_url", return_value="https://auth.example.com"):
            with patch(
                "src.services.auth.httpx.AsyncClient",
                return_value=_mock_httpx_client(resp),
            ) as mock_client_cls:
                result = await _fetch_auth("github-token", provider="gcp")
        request_call = mock_client_cls.return_value.__aenter__.return_value.request
        args, kwargs = request_call.await_args
        assert args[0] == "POST"
        assert args[1] == "https://auth.example.com/client/auth"
        assert kwargs["json"] == {
            "github_token": "github-token",
            "provider": "gcp",
        }
        assert result.access_token == "jwt123"
        assert result.s3_access is None

    @pytest.mark.asyncio
    async def test_success_with_gcp_provider_keeps_gcp_access_token(self):
        resp = httpx.Response(
            200,
            json={
                "token": "backend-jwt",
                "access_token": "ya29.gcp-token",
                "expiration": "2099-03-17T09:50:27+00:00",
                "auth0_token": "auth0-jwt",
                "observability_global_service": "https://www.gs.us-central1.gcp.observability.tidbcloud.com",
                "provider": "gcp",
            },
        )
        with patch.object(auth, "get_auth_service_url", return_value="https://auth.example.com"):
            with patch(
                "src.services.auth.httpx.AsyncClient",
                return_value=_mock_httpx_client(resp),
            ):
                result = await _fetch_auth("github-token", provider="gcp")
        assert result.access_token == "backend-jwt"
        assert result.gcp_access_token == "ya29.gcp-token"
        assert result.s3_access is None

    @pytest.mark.asyncio
    async def test_invalid_response_raises_auth_service_error(self):
        resp = httpx.Response(200, json={"token": 12345, "expires_in": 3600})
        with patch.object(auth, "get_auth_service_url", return_value="https://auth.example.com"):
            with patch(
                "src.services.auth.httpx.AsyncClient",
                return_value=_mock_httpx_client(resp),
            ):
                with pytest.raises(AuthServiceError) as exc_info:
                    await _fetch_auth("github-token")
                assert "invalid" in exc_info.value.message.lower()


class TestServerOwnedAuth:
    @pytest.mark.asyncio
    async def test_bootstrap_server_auth_calls_fetch_and_caches(self):
        response = AuthResponse(
            access_token="jwt",
            expires_in=3600,
            auth0Token=None,
            observabilityGlobalService=None,
            s3_access=None,
        )
        with patch.object(auth, "get_github_token", return_value="gh-token"):
            with patch.object(auth, "get_auth_provider", return_value="aws"):
                with patch.object(
                    auth, "_fetch_auth", new_callable=AsyncMock, return_value=response
                ) as mock_fetch:
                    with patch.object(auth, "_set_cached") as mock_cache:
                        result = await bootstrap_server_auth()
        assert result is response
        mock_fetch.assert_awaited_once_with("gh-token", provider="aws")
        mock_cache.assert_called_once_with(response, provider="aws")

    @pytest.mark.asyncio
    async def test_bootstrap_server_auth_uses_configured_provider(self):
        response = AuthResponse(
            access_token="jwt",
            expires_in=3600,
            auth0Token=None,
            observabilityGlobalService=None,
            s3_access=None,
        )
        with patch.object(auth, "get_github_token", return_value="gh-token"):
            with patch.object(auth, "get_auth_provider", return_value="alicloud"):
                with patch.object(
                    auth, "_fetch_auth", new_callable=AsyncMock, return_value=response
                ) as mock_fetch:
                    with patch.object(auth, "_set_cached"):
                        await bootstrap_server_auth()
        mock_fetch.assert_awaited_once_with("gh-token", provider="alicloud")

    @pytest.mark.asyncio
    async def test_bootstrap_server_auth_raises_when_github_token_missing(self):
        with patch.object(auth, "get_github_token", return_value=None):
            with pytest.raises(AuthServiceError) as exc_info:
                await bootstrap_server_auth()
        assert "GITHUB_TOKEN" in exc_info.value.message

    @pytest.mark.asyncio
    async def test_get_valid_server_auth_returns_cached_value(self):
        cached = AuthResponse(
            access_token="jwt",
            expires_in=3600,
            auth0Token=None,
            observabilityGlobalService=None,
            s3_access=None,
        )
        with patch.object(auth, "_get_cached", return_value=cached):
            result = await get_valid_server_auth()
        assert result is cached

    @pytest.mark.asyncio
    async def test_get_valid_server_auth_bootstraps_on_cache_miss(self):
        fresh = AuthResponse(
            access_token="jwt",
            expires_in=3600,
            auth0Token=None,
            observabilityGlobalService=None,
            s3_access=None,
        )
        with patch.object(auth, "_get_cached", return_value=None):
            with patch.object(
                auth, "bootstrap_server_auth", new_callable=AsyncMock, return_value=fresh
            ) as mock_bootstrap:
                result = await get_valid_server_auth()
        assert result is fresh
        mock_bootstrap.assert_awaited_once_with("aws")

    @pytest.mark.asyncio
    async def test_get_valid_server_auth_uses_vendor_parameter(self):
        cached = AuthResponse(
            access_token="jwt",
            expires_in=3600,
            auth0Token=None,
            observabilityGlobalService=None,
            s3_access=None,
        )
        with patch.object(auth, "_get_cached", return_value=cached) as mock_cached:
            result = await get_valid_server_auth(vendor="alicloud")
        assert result is cached
        mock_cached.assert_called_once_with("alicloud")

    @pytest.mark.asyncio
    async def test_get_server_access_token_returns_token(self):
        cached = AuthResponse(
            access_token="jwt",
            expires_in=3600,
            auth0Token=None,
            observabilityGlobalService=None,
            s3_access=None,
        )
        with patch.object(auth, "_get_cached", return_value=cached):
            assert await get_server_access_token() == "jwt"

    @pytest.mark.asyncio
    async def test_get_server_access_token_refreshes_when_cache_empty(self):
        refreshed = AuthResponse(
            access_token="fresh-jwt",
            expires_in=3600,
            auth0Token=None,
            observabilityGlobalService=None,
            s3_access=None,
        )
        with patch.object(auth, "_get_cached", return_value=None):
            with patch.object(
                auth,
                "bootstrap_server_auth",
                new_callable=AsyncMock,
                return_value=refreshed,
            ):
                assert await get_server_access_token() == "fresh-jwt"

    @pytest.mark.asyncio
    async def test_get_server_auth0_token_returns_token(self):
        cached = AuthResponse(
            access_token="jwt",
            expires_in=3600,
            auth0Token="auth0-jwt",
            observabilityGlobalService=None,
            s3_access=None,
        )
        with patch.object(auth, "_get_cached", return_value=cached):
            assert await get_server_auth0_token() == "auth0-jwt"

    @pytest.mark.asyncio
    async def test_get_server_auth0_token_refreshes_when_cache_empty(self):
        refreshed = AuthResponse(
            access_token="jwt",
            expires_in=3600,
            auth0Token="fresh-auth0",
            observabilityGlobalService=None,
            s3_access=None,
        )
        with patch.object(auth, "_get_cached", return_value=None):
            with patch.object(
                auth,
                "bootstrap_server_auth",
                new_callable=AsyncMock,
                return_value=refreshed,
            ):
                assert await get_server_auth0_token() == "fresh-auth0"


class TestGetDeltaCredentials:
    @pytest.mark.asyncio
    async def test_returns_none_when_auth_not_configured(self):
        with patch.object(auth, "_get_cached", return_value=None):
            result = await get_delta_credentials()
        assert result is None

    @pytest.mark.asyncio
    async def test_returns_none_when_cache_empty(self):
        with patch.object(auth, "_get_cached", return_value=None):
            result = await get_delta_credentials()
        assert result is None

    @pytest.mark.asyncio
    async def test_returns_none_when_no_s3_access_in_cache(self):
        cached = AuthResponse(
            access_token="jwt",
            expires_in=3600,
            auth0Token=None,
            observabilityGlobalService=None,
            s3_access=None,
        )
        with patch.object(auth, "_get_cached", return_value=cached):
            result = await get_delta_credentials()
        assert result is None

    @pytest.mark.asyncio
    async def test_returns_credentials_dict_when_s3_present(self):
        s3 = S3Access(
            provider="aws",
            access_key_id=None,
            secret_access_key="secret",
            region="us-east-1",
            endpoint=None,
            account_name=None,
            session_token=None,
        )
        cached = AuthResponse(
            access_token="jwt",
            expires_in=3600,
            auth0Token=None,
            observabilityGlobalService=None,
            s3_access=s3,
        )
        with patch.object(auth, "_get_cached", return_value=cached):
            result = await get_delta_credentials()
        assert result is not None
        assert result["provider"] == "aws"
        assert result["secret_access_key"] == "secret"

    @pytest.mark.asyncio
    async def test_get_delta_credentials_uses_vendor_parameter(self):
        s3 = S3Access(
            provider="alicloud",
            access_key_id="ak",
            secret_access_key="sk",
            region="cn-hangzhou",
            endpoint=None,
            account_name=None,
            session_token=None,
        )
        cached = AuthResponse(
            access_token="jwt",
            expires_in=3600,
            auth0Token=None,
            observabilityGlobalService=None,
            s3_access=s3,
        )
        with patch.object(auth, "_get_cached", return_value=cached) as mock_cached:
            result = await get_delta_credentials(vendor="alicloud")
        mock_cached.assert_called_once_with("alicloud")
        assert result is not None
        assert result["provider"] == "alicloud"


class TestGetGcpAccessToken:
    @pytest.mark.asyncio
    async def test_returns_none_when_not_available(self):
        cached = AuthResponse(
            access_token="backend-jwt",
            gcp_access_token=None,
            expires_in=3600,
            auth0Token=None,
            observabilityGlobalService=None,
            s3_access=None,
        )
        with patch.object(auth, "_get_cached", return_value=cached):
            result = await get_gcp_access_token()
        assert result is None

    @pytest.mark.asyncio
    async def test_returns_gcp_access_token(self):
        cached = AuthResponse(
            access_token="backend-jwt",
            gcp_access_token="ya29.real-token",
            expires_in=3600,
            auth0Token=None,
            observabilityGlobalService=None,
            s3_access=None,
        )
        with patch.object(auth, "_get_cached", return_value=cached):
            result = await get_gcp_access_token()
        assert result == "ya29.real-token"


class TestGetPrometheusClientForCluster:
    @pytest.mark.asyncio
    async def test_returns_client_with_resolved_url_and_token(self):
        from src.models.resolve_metrics_url import ResolveMetricsUrlResponse

        resolved = ResolveMetricsUrlResponse(
            url="https://cluster-prom.example.com", token="cluster-token"
        )
        with patch(
            "src.services.observability_resolve.get_or_resolve_metrics_url",
            new_callable=AsyncMock,
            return_value=resolved,
        ) as mock_resolve:
            client = await get_prometheus_client_for_cluster("cluster-1", "dev", "starter", 20.0)
        mock_resolve.assert_awaited_once_with(
            cluster_id="cluster-1",
            env="dev",
            deploy_type="starter",
        )
        assert client._base_url == "https://cluster-prom.example.com"
        assert client.timeout == 20.0
        assert client._token == "cluster-token"