social-app/backend/tests/unit/v1/auth/test_auth_gateway.py

from __future__ import annotations

from types import SimpleNamespace
from unittest.mock import MagicMock

import pytest
from fastapi import HTTPException

from v1.auth.gateway import SupabaseAuthGateway
from v1.auth.schemas import (
    PasswordResetConfirmRequest,
    PasswordResetRequest,
    SessionCreateRequest,
    SessionRefreshRequest,
    VerificationCreateRequest,
    VerificationVerifyRequest,
    VerificationResendRequest,
)


class TestSupabaseAuthGateway:
    @pytest.fixture
    def gateway(
        self, monkeypatch: pytest.MonkeyPatch
    ) -> tuple[SupabaseAuthGateway, MagicMock, MagicMock]:
        mock_client = MagicMock()
        mock_admin_client = MagicMock()
        monkeypatch.setattr(
            "v1.auth.gateway.supabase_service.get_client", lambda: mock_client
        )
        monkeypatch.setattr(
            "v1.auth.gateway.supabase_service.get_admin_client",
            lambda: mock_admin_client,
        )
        return SupabaseAuthGateway(), mock_client, mock_admin_client

    @pytest.mark.asyncio
    async def test_request_password_reset_calls_email_with_string(
        self, gateway: tuple[SupabaseAuthGateway, MagicMock, MagicMock]
    ) -> None:
        sut, mock_client, _ = gateway
        mock_reset_email = MagicMock()
        mock_client.auth.reset_password_email = mock_reset_email

        request = PasswordResetRequest(email="test@example.com")
        await sut.request_password_reset(request)

        mock_reset_email.assert_called_once_with("test@example.com")

    @pytest.mark.asyncio
    async def test_create_verification_maps_timeout_error_to_503(
        self, gateway: tuple[SupabaseAuthGateway, MagicMock, MagicMock]
    ) -> None:
        sut, mock_client, _ = gateway
        from supabase import AuthError

        mock_client.auth.sign_up = MagicMock(
            side_effect=AuthError("request_timeout", None)
        )

        with pytest.raises(HTTPException) as exc_info:
            await sut.create_verification(
                VerificationCreateRequest(
                    username="tester",
                    email="test@example.com",
                    password="secret123",
                )
            )

        assert exc_info.value.status_code == 503
        assert exc_info.value.detail == "Auth service temporarily unavailable"

    @pytest.mark.asyncio
    async def test_request_password_reset_with_redirect(
        self, gateway: tuple[SupabaseAuthGateway, MagicMock, MagicMock]
    ) -> None:
        sut, mock_client, _ = gateway
        mock_reset_email = MagicMock()
        mock_client.auth.reset_password_email = mock_reset_email

        request = PasswordResetRequest(
            email="test@example.com",
            redirect_to="http://localhost:3000/reset-password",
        )
        await sut.request_password_reset(request)

        mock_reset_email.assert_called_once_with(
            "test@example.com",
            options={"redirect_to": "http://localhost:3000/reset-password"},
        )

    @pytest.mark.asyncio
    async def test_create_verification_rejects_untrusted_redirect_url(
        self, gateway: tuple[SupabaseAuthGateway, MagicMock, MagicMock]
    ) -> None:
        sut, _, _ = gateway

        with pytest.raises(HTTPException) as exc_info:
            await sut.create_verification(
                VerificationCreateRequest(
                    username="tester",
                    email="test@example.com",
                    password="secret123",
                    redirect_to="https://evil.example.com/callback",
                )
            )

        assert exc_info.value.status_code == 422
        assert exc_info.value.detail == "Invalid redirect URL"

    @pytest.mark.asyncio
    async def test_request_password_reset_rejects_untrusted_redirect_url(
        self, gateway: tuple[SupabaseAuthGateway, MagicMock, MagicMock]
    ) -> None:
        sut, _, _ = gateway

        with pytest.raises(HTTPException) as exc_info:
            await sut.request_password_reset(
                PasswordResetRequest(
                    email="test@example.com",
                    redirect_to="https://evil.example.com/reset",
                )
            )

        assert exc_info.value.status_code == 422
        assert exc_info.value.detail == "Invalid redirect URL"

    @pytest.mark.asyncio
    async def test_request_password_reset_swallows_auth_error(
        self, gateway: tuple[SupabaseAuthGateway, MagicMock, MagicMock]
    ) -> None:
        sut, mock_client, _ = gateway
        from supabase import AuthError

        mock_reset_email = MagicMock(side_effect=AuthError("rate limit exceeded", None))
        mock_client.auth.reset_password_email = mock_reset_email

        request = PasswordResetRequest(email="test@example.com")

        result = await sut.request_password_reset(request)

        mock_reset_email.assert_called_once()
        assert result is None

    @pytest.mark.asyncio
    async def test_request_password_reset_extracts_email_from_mapping(
        self, gateway: tuple[SupabaseAuthGateway, MagicMock, MagicMock]
    ) -> None:
        sut, mock_client, _ = gateway
        mock_reset_email = MagicMock()
        mock_client.auth.reset_password_email = mock_reset_email

        request = PasswordResetRequest.model_construct(
            email={"email": "test@example.com"},
            redirect_to=None,
        )

        await sut.request_password_reset(request)

        mock_reset_email.assert_called_once_with("test@example.com")

    @pytest.mark.asyncio
    async def test_request_password_reset_rejects_invalid_email_shape(
        self, gateway: tuple[SupabaseAuthGateway, MagicMock, MagicMock]
    ) -> None:
        sut, _, _ = gateway
        request = PasswordResetRequest.model_construct(
            email={"unexpected": "value"},
            redirect_to=None,
        )

        with pytest.raises(HTTPException) as exc_info:
            await sut.request_password_reset(request)

        assert exc_info.value.status_code == 422
        assert exc_info.value.detail == "Invalid email"

    @pytest.mark.asyncio
    async def test_confirm_password_reset_updates_password_by_user_id(
        self, gateway: tuple[SupabaseAuthGateway, MagicMock, MagicMock]
    ) -> None:
        sut, mock_client, mock_admin_client = gateway
        verify_response = SimpleNamespace(
            session=SimpleNamespace(access_token="access"),
            user=SimpleNamespace(id="user-1"),
        )
        mock_verify_otp = MagicMock(return_value=verify_response)
        mock_client.auth.verify_otp = mock_verify_otp

        mock_update_user_by_id = MagicMock()
        mock_admin_client.auth.admin = SimpleNamespace(
            update_user_by_id=mock_update_user_by_id
        )

        request = PasswordResetConfirmRequest(
            email="test@example.com",
            token="123456",
            new_password="newpassword123",
        )

        await sut.confirm_password_reset(request)

        mock_verify_otp.assert_called_once_with(
            {
                "type": "recovery",
                "email": "test@example.com",
                "token": "123456",
            }
        )
        mock_update_user_by_id.assert_called_once_with(
            "user-1",
            {"password": "newpassword123"},
        )

    @pytest.mark.asyncio
    async def test_confirm_password_reset_raises_when_user_id_missing(
        self, gateway: tuple[SupabaseAuthGateway, MagicMock, MagicMock]
    ) -> None:
        sut, mock_client, _ = gateway
        verify_response = SimpleNamespace(
            session=SimpleNamespace(access_token="access"),
            user=SimpleNamespace(id=""),
        )
        mock_client.auth.verify_otp = MagicMock(return_value=verify_response)

        request = PasswordResetConfirmRequest(
            email="test@example.com",
            token="123456",
            new_password="newpassword123",
        )

        with pytest.raises(HTTPException) as exc_info:
            await sut.confirm_password_reset(request)

        assert exc_info.value.status_code == 401
        assert exc_info.value.detail == "Invalid or expired verification code"

    @pytest.mark.asyncio
    async def test_recovery_resend_calls_reset_password_email(
        self, gateway: tuple[SupabaseAuthGateway, MagicMock, MagicMock]
    ) -> None:
        sut, mock_client, _ = gateway
        mock_reset_email = MagicMock()
        mock_client.auth.reset_password_email = mock_reset_email

        await sut.resend_verification(
            VerificationResendRequest(
                type="recovery",
                email="test@example.com",
                redirect_to="http://localhost:3000/reset-password",
            )
        )

        mock_reset_email.assert_called_once_with(
            "test@example.com",
            options={"redirect_to": "http://localhost:3000/reset-password"},
        )

    @pytest.mark.asyncio
    async def test_verify_verification_maps_internal_error_to_503(
        self, gateway: tuple[SupabaseAuthGateway, MagicMock, MagicMock]
    ) -> None:
        sut, mock_client, _ = gateway
        from supabase import AuthError

        mock_client.auth.verify_otp = MagicMock(
            side_effect=AuthError("internal_server_error", None)
        )

        with pytest.raises(HTTPException) as exc_info:
            await sut.verify_verification(
                VerificationVerifyRequest(
                    type="signup",
                    email="test@example.com",
                    token="123456",
                )
            )

        assert exc_info.value.status_code == 503
        assert exc_info.value.detail == "Auth service temporarily unavailable"

    @pytest.mark.asyncio
    async def test_create_session_maps_internal_error_to_503(
        self, gateway: tuple[SupabaseAuthGateway, MagicMock, MagicMock]
    ) -> None:
        sut, mock_client, _ = gateway
        from supabase import AuthError

        mock_client.auth.sign_in_with_password = MagicMock(
            side_effect=AuthError("internal_server_error", None)
        )

        with pytest.raises(HTTPException) as exc_info:
            await sut.create_session(
                SessionCreateRequest(
                    email="test@example.com",
                    password="secret123",
                )
            )

        assert exc_info.value.status_code == 503
        assert exc_info.value.detail == "Auth service temporarily unavailable"

    @pytest.mark.asyncio
    async def test_refresh_session_maps_bad_gateway_to_503(
        self, gateway: tuple[SupabaseAuthGateway, MagicMock, MagicMock]
    ) -> None:
        sut, mock_client, _ = gateway
        from supabase import AuthError

        mock_client.auth.refresh_session = MagicMock(
            side_effect=AuthError("bad_gateway", None)
        )

        with pytest.raises(HTTPException) as exc_info:
            await sut.refresh_session(SessionRefreshRequest(refresh_token="rt"))

        assert exc_info.value.status_code == 503
        assert exc_info.value.detail == "Auth service temporarily unavailable"

    @pytest.mark.asyncio
    async def test_confirm_password_reset_maps_service_unavailable_to_503(
        self, gateway: tuple[SupabaseAuthGateway, MagicMock, MagicMock]
    ) -> None:
        sut, mock_client, _ = gateway
        from supabase import AuthError

        mock_client.auth.verify_otp = MagicMock(
            side_effect=AuthError("service_unavailable", None)
        )

        with pytest.raises(HTTPException) as exc_info:
            await sut.confirm_password_reset(
                PasswordResetConfirmRequest(
                    email="test@example.com",
                    token="123456",
                    new_password="newpassword123",
                )
            )

        assert exc_info.value.status_code == 503
        assert exc_info.value.detail == "Auth service temporarily unavailable"

    @pytest.mark.asyncio
    async def test_get_user_by_email_uses_in_memory_cache(
        self,
        gateway: tuple[SupabaseAuthGateway, MagicMock, MagicMock],
        monkeypatch: pytest.MonkeyPatch,
    ) -> None:
        sut, _, _ = gateway
        user = SimpleNamespace(
            id="user-1",
            email="cached@example.com",
            created_at="2026-03-16T00:00:00Z",
            email_confirmed_at=None,
        )
        list_calls = {"count": 0}

        def _fake_list_auth_users(_client: object) -> list[SimpleNamespace]:
            list_calls["count"] += 1
            return [user]

        monkeypatch.setattr("v1.auth.gateway._list_auth_users", _fake_list_auth_users)

        first = await sut.get_user_by_email("cached@example.com")
        second = await sut.get_user_by_email("CACHED@example.com")

        assert first.id == "user-1"
        assert second.email == "cached@example.com"
        assert list_calls["count"] == 1