life-echo/api/tests/test_avatar_preset_http.py

"""预设头像 HTTP 契约。"""

from __future__ import annotations

import uuid
from datetime import datetime, timezone
from pathlib import Path
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
from app.features.user.models import User


def _mock_current_user() -> User:
    u = MagicMock(spec=User)
    u.id = str(uuid.uuid4())
    u.phone = "13800000000"
    u.email = None
    u.nickname = "测试用户"
    u.avatar_url = None
    u.subscription_type = "free"
    u.created_at = datetime.now(timezone.utc)
    return u


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

    fixed_user = _mock_current_user()

    async def _fake_update_avatar(uid: str, url: str):
        fixed_user.avatar_url = url
        return fixed_user

    mock_service = MagicMock(spec=AuthService)
    mock_service.update_avatar_url = AsyncMock(side_effect=_fake_update_avatar)

    app.dependency_overrides[get_auth_service] = lambda: mock_service
    app.dependency_overrides[get_current_user] = lambda: fixed_user

    app.state._mock_auth_service = mock_service
    app.state._fixed_user = fixed_user
    return app


@pytest.mark.asyncio
async def test_list_avatar_presets(preset_auth_app: FastAPI) -> None:
    transport = ASGITransport(app=preset_auth_app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        r = await ac.get("/api/auth/avatar-presets")
    assert r.status_code == 200
    items = r.json()
    assert len(items) == 8
    assert items[0]["id"] == "01"
    assert items[0]["url"] == "/api/auth/avatar-presets/01.png"


@pytest.mark.asyncio
async def test_get_avatar_preset_ok(preset_auth_app: FastAPI) -> None:
    transport = ASGITransport(app=preset_auth_app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        r = await ac.get("/api/auth/avatar-presets/01.png")
    assert r.status_code == 200
    assert r.headers.get("content-type", "").startswith("image/png")


@pytest.mark.asyncio
async def test_get_avatar_preset_unknown(preset_auth_app: FastAPI) -> None:
    transport = ASGITransport(app=preset_auth_app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        r = await ac.get("/api/auth/avatar-presets/99.png")
    assert r.status_code == 404


@pytest.mark.asyncio
async def test_get_avatar_preset_path_traversal(preset_auth_app: FastAPI) -> None:
    transport = ASGITransport(app=preset_auth_app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        r = await ac.get("/api/auth/avatar-presets/../secrets.env")
    assert r.status_code == 404


@pytest.mark.asyncio
async def test_set_avatar_preset_ok(preset_auth_app: FastAPI) -> None:
    uid = preset_auth_app.state._fixed_user.id
    transport = ASGITransport(app=preset_auth_app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        r = await ac.put(
            "/api/auth/me/avatar/preset",
            json={"preset_id": "02"},
            headers={"Authorization": "Bearer x"},
        )
    assert r.status_code == 200
    body = r.json()
    assert body["avatar_url"] == "/api/auth/avatar-presets/02.png"
    svc: MagicMock = preset_auth_app.state._mock_auth_service
    svc.update_avatar_url.assert_awaited_once_with(
        uid, "/api/auth/avatar-presets/02.png"
    )


@pytest.mark.asyncio
async def test_set_avatar_preset_invalid(preset_auth_app: FastAPI) -> None:
    transport = ASGITransport(app=preset_auth_app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        r = await ac.put(
            "/api/auth/me/avatar/preset",
            json={"preset_id": "99"},
            headers={"Authorization": "Bearer x"},
        )
    assert r.status_code == 400


@pytest.mark.asyncio
async def test_get_uploaded_avatar_rejects_traversal() -> None:
    app = FastAPI()
    app.include_router(auth_router)
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        r = await ac.get("/api/auth/avatars/../../../etc/passwd")
    assert r.status_code == 404


@pytest.mark.asyncio
async def test_get_uploaded_avatar_ok_with_safe_file(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    import app.features.auth.router as auth_router_mod

    avatar_dir = tmp_path / "avatars"
    avatar_dir.mkdir()
    (avatar_dir / "abc-def-123.jpg").write_bytes(b"x")

    monkeypatch.setattr(auth_router_mod, "AVATAR_DIR", avatar_dir)

    app = FastAPI()
    app.include_router(auth_router)
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        r = await ac.get("/api/auth/avatars/abc-def-123.jpg")
    assert r.status_code == 200


@pytest.mark.asyncio
async def test_get_uploaded_avatar_ok_when_stem_has_underscore(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    import app.features.auth.router as auth_router_mod

    avatar_dir = tmp_path / "avatars"
    avatar_dir.mkdir()
    (avatar_dir / "user_abc_01.jpg").write_bytes(b"x")

    monkeypatch.setattr(auth_router_mod, "AVATAR_DIR", avatar_dir)

    app = FastAPI()
    app.include_router(auth_router)
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        r = await ac.get("/api/auth/avatars/user_abc_01.jpg")
    assert r.status_code == 200