f4e18f2b9b
Mock COS storage for auth router DI and skip Postgres integration tests when the database is unreachable. Co-authored-by: Cursor <cursoragent@cursor.com>
142 lines
4.9 KiB
Python
142 lines
4.9 KiB
Python
"""真实 feature router 的错误契约 HTTP 场景测试。"""
|
|
|
|
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_object_storage
|
|
from app.core.errors import BadRequestError, NotFoundError
|
|
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 AuthError
|
|
from app.features.conversation.deps import get_conversation_service
|
|
from app.features.conversation.router import router as conversation_router
|
|
from tests.conftest import install_test_error_handlers
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_auth_register_validation_returns_unified_422() -> None:
|
|
app = install_test_error_handlers(FastAPI())
|
|
app.include_router(auth_router)
|
|
app.dependency_overrides[get_object_storage] = lambda: MagicMock()
|
|
|
|
transport = ASGITransport(app=app, raise_app_exceptions=False)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
r = await client.post(
|
|
"/api/auth/register",
|
|
json={"phone": "123", "password": "x", "nickname": ""},
|
|
)
|
|
|
|
assert r.status_code == 422
|
|
body = r.json()
|
|
assert body["error_code"] == "VALIDATION_ERROR"
|
|
assert "request_id" in body
|
|
assert body["message"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_conversation_list_requires_auth_unified_401() -> None:
|
|
app = install_test_error_handlers(FastAPI())
|
|
app.include_router(conversation_router)
|
|
|
|
transport = ASGITransport(app=app, raise_app_exceptions=False)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
r = await client.get("/api/conversations")
|
|
|
|
assert r.status_code == 401
|
|
body = r.json()
|
|
assert body["error_code"] == "AUTHENTICATION_FAILED"
|
|
assert r.headers.get("www-authenticate") == "Bearer"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_conversation_detail_not_found_unified_404() -> None:
|
|
app = install_test_error_handlers(FastAPI())
|
|
app.include_router(conversation_router)
|
|
|
|
mock_service = MagicMock()
|
|
mock_service.get_one = AsyncMock(
|
|
side_effect=NotFoundError("Conversation not found")
|
|
)
|
|
app.dependency_overrides[get_conversation_service] = lambda: mock_service
|
|
|
|
fake_user = MagicMock()
|
|
fake_user.id = "user-1"
|
|
from app.core.dependencies import get_current_user
|
|
|
|
app.dependency_overrides[get_current_user] = lambda: fake_user
|
|
|
|
transport = ASGITransport(app=app, raise_app_exceptions=False)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
r = await client.get("/api/conversations/conv-missing")
|
|
|
|
assert r.status_code == 404
|
|
body = r.json()
|
|
assert body["error_code"] == "NOT_FOUND"
|
|
assert body["message"] == "Conversation not found"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_auth_login_invalid_credentials_unified_401() -> None:
|
|
app = install_test_error_handlers(FastAPI())
|
|
app.include_router(auth_router)
|
|
|
|
mock_service = MagicMock()
|
|
mock_service.login = AsyncMock(
|
|
side_effect=AuthError("手机号或密码错误", "INVALID_CREDENTIALS")
|
|
)
|
|
app.dependency_overrides[get_auth_service] = lambda: mock_service
|
|
|
|
transport = ASGITransport(app=app, raise_app_exceptions=False)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
r = await client.post(
|
|
"/api/auth/login",
|
|
json={
|
|
"phone": "13800138000",
|
|
"password": "wrong-password",
|
|
"agreed_to_terms": True,
|
|
},
|
|
)
|
|
|
|
assert r.status_code == 401
|
|
body = r.json()
|
|
assert body["error_code"] == "AUTHENTICATION_FAILED"
|
|
assert body["message"] == "手机号或密码错误"
|
|
assert "request_id" in body
|
|
assert r.headers.get("www-authenticate") == "Bearer"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_auth_avatar_invalid_image_unified_400() -> None:
|
|
app = install_test_error_handlers(FastAPI())
|
|
app.include_router(auth_router)
|
|
|
|
mock_service = MagicMock()
|
|
mock_service.upload_avatar = AsyncMock(
|
|
side_effect=BadRequestError("无效的图片文件格式。文件头: deadbeef")
|
|
)
|
|
app.dependency_overrides[get_auth_service] = lambda: mock_service
|
|
|
|
fake_user = MagicMock()
|
|
fake_user.id = "user-1"
|
|
fake_user.avatar_url = None
|
|
from app.core.dependencies import get_current_user
|
|
|
|
app.dependency_overrides[get_current_user] = lambda: fake_user
|
|
|
|
transport = ASGITransport(app=app, raise_app_exceptions=False)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
r = await client.post(
|
|
"/api/auth/me/avatar",
|
|
files={"file": ("x.png", b"not-an-image", "image/png")},
|
|
)
|
|
|
|
assert r.status_code == 400
|
|
body = r.json()
|
|
assert body["error_code"] == "BAD_REQUEST"
|
|
assert "无效的图片" in body["message"]
|