life-echo/api/tests/test_auth_flow_http.py

"""Auth HTTP 契约：注册 / 登录 / 刷新 / 受保护 /me（依赖注入 mock）。"""

from __future__ import annotations

from unittest.mock import AsyncMock, MagicMock

import pytest
from fastapi import FastAPI
from httpx import ASGITransport, AsyncClient

from app.core.dependencies import get_current_user
from app.features.auth.deps import get_auth_service
from app.features.auth.router import router as auth_router
from app.features.auth.service import AuthService


@pytest.fixture
def auth_app(make_test_user) -> FastAPI:
    app = FastAPI()
    app.include_router(auth_router)

    mock_service = MagicMock(spec=AuthService)
    mock_service.register = AsyncMock(
        return_value={"access_token": "access-reg", "refresh_token": "refresh-reg"}
    )
    mock_service.login = AsyncMock(
        return_value={"access_token": "access-login", "refresh_token": "refresh-login"}
    )
    mock_service.login_with_google = AsyncMock(
        return_value={
            "access_token": "access-google",
            "refresh_token": "refresh-google",
        }
    )
    mock_service.refresh_tokens = AsyncMock(
        return_value={"access_token": "access-new", "refresh_token": "refresh-new"}
    )

    app.dependency_overrides[get_auth_service] = lambda: mock_service
    app.dependency_overrides[get_current_user] = lambda: make_test_user()

    app.state._mock_auth_service = mock_service
    return app


@pytest.mark.asyncio
async def test_register_login_refresh_me(auth_app: FastAPI, unique_phone: str) -> None:
    transport = ASGITransport(app=auth_app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        reg = await ac.post(
            "/api/auth/register",
            json={
                "phone": unique_phone,
                "password": "secret12",
                "nickname": "T",
                "agreed_to_terms": True,
            },
        )
        assert reg.status_code == 201
        body = reg.json()
        assert body["access_token"] == "access-reg"

        login = await ac.post(
            "/api/auth/login",
            json={
                "phone": unique_phone,
                "password": "secret12",
                "agreed_to_terms": True,
            },
        )
        assert login.status_code == 200
        assert login.json()["access_token"] == "access-login"

        google = await ac.post(
            "/api/auth/login/google",
            json={
                "id_token": "header.payload.signature",
                "agreed_to_terms": True,
            },
        )
        assert google.status_code == 200
        assert google.json()["access_token"] == "access-google"

        ref = await ac.post(
            "/api/auth/refresh",
            json={"refresh_token": "refresh-login"},
        )
        assert ref.status_code == 200
        assert ref.json()["access_token"] == "access-new"

        me = await ac.get(
            "/api/auth/me",
            headers={"Authorization": "Bearer any"},
        )
        assert me.status_code == 200
        assert me.json()["nickname"] == "测试用户"

    svc: MagicMock = auth_app.state._mock_auth_service
    svc.register.assert_awaited_once()
    svc.login.assert_awaited_once()
    svc.login_with_google.assert_awaited_once_with(
        id_token="header.payload.signature",
        language=None,
    )
    svc.refresh_tokens.assert_awaited_once_with(refresh_token="refresh-login")


@pytest.mark.asyncio
async def test_me_without_auth_returns_401() -> None:
    app = FastAPI()
    app.include_router(auth_router)

    mock_service = MagicMock(spec=AuthService)
    app.dependency_overrides[get_auth_service] = lambda: mock_service
    # 不覆盖 get_current_user — 使用真实 OAuth2，无 header 时应 401

    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        r = await ac.get("/api/auth/me")
        assert r.status_code == 401