测试与质量保证

安装测试依赖

pip install pytest pytest-asyncio httpx pytest-cov

# pyproject.toml 配置
# [tool.pytest.ini_options]
# asyncio_mode = "auto"
# testpaths = ["tests"]

TestClient 同步测试

TestClient 封装了 HTTPX，提供类 requests 的同步接口，适合大多数路由测试场景：

# tests/test_basic.py
from fastapi.testclient import TestClient
from app.main import app

client = TestClient(app)

# ── 基本 GET 测试 ─────────────────────────────────────────
def test_health_check():
    response = client.get("/health")
    assert response.status_code == 200
    data = response.json()
    assert data["status"] == "ok"

# ── POST 测试（请求体）────────────────────────────────────
def test_create_user():
    response = client.post("/users/", json={
        "username": "testuser",
        "email": "test@example.com",
        "password": "secret123"
    })
    assert response.status_code == 201
    data = response.json()
    assert data["username"] == "testuser"
    assert "password" not in data  # 确认密码未泄露
    assert "id" in data

# ── 验证错误测试 ──────────────────────────────────────────
def test_create_user_invalid_data():
    response = client.post("/users/", json={
        "username": "",  # 空用户名应验证失败
        "email": "not-an-email"
    })
    assert response.status_code == 422  # Unprocessable Entity
    errors = response.json()["detail"]
    assert len(errors) > 0

# ── 404 测试 ─────────────────────────────────────────────
def test_get_nonexistent_user():
    response = client.get("/users/99999")
    assert response.status_code == 404
    assert "不存在" in response.json()["detail"]

Fixture：数据库测试隔离

# tests/conftest.py
import pytest
import pytest_asyncio
from httpx import AsyncClient, ASGITransport
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from app.main import app
from app.database import Base, get_db

# 使用独立的内存数据库进行测试
TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:"

@pytest.fixture(scope="session")
def event_loop_policy():
    import asyncio
    return asyncio.DefaultEventLoopPolicy()

@pytest_asyncio.fixture(scope="function")
async def test_db():
    """每个测试函数使用独立的内存数据库，测试完毕自动清空"""
    engine = create_async_engine(TEST_DATABASE_URL, echo=False)
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

    session_factory = async_sessionmaker(engine, expire_on_commit=False)

    async with session_factory() as session:
        yield session  # 注入测试使用的 Session

    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
    await engine.dispose()

@pytest_asyncio.fixture()
async def async_client(test_db: AsyncSession):
    """覆盖数据库依赖，使用测试数据库"""
    async def override_get_db():
        yield test_db

    # dependency_overrides：替换真实依赖为测试版本
    app.dependency_overrides[get_db] = override_get_db

    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test"
    ) as client:
        yield client

    app.dependency_overrides.clear()  # 清除覆盖，避免影响其他测试

AsyncClient 异步测试

# tests/test_users.py
import pytest
from httpx import AsyncClient

@pytest.mark.asyncio
async def test_create_and_get_user(async_client: AsyncClient):
    # 创建用户
    create_resp = await async_client.post("/users/", json={
        "username": "alice",
        "email": "alice@example.com",
        "password": "password123"
    })
    assert create_resp.status_code == 201
    user_id = create_resp.json()["id"]

    # 获取用户
    get_resp = await async_client.get(f"/users/{user_id}")
    assert get_resp.status_code == 200
    user = get_resp.json()
    assert user["username"] == "alice"
    assert "hashed_password" not in user

@pytest.mark.asyncio
async def test_login_and_access_protected(async_client: AsyncClient):
    # 先注册
    await async_client.post("/users/", json={
        "username": "bob",
        "email": "bob@example.com",
        "password": "mypassword"
    })

    # 登录获取 Token
    login_resp = await async_client.post(
        "/auth/login",
        data={"username": "bob", "password": "mypassword"}
    )
    assert login_resp.status_code == 200
    token = login_resp.json()["access_token"]

    # 用 Token 访问受保护端点
    me_resp = await async_client.get(
        "/auth/me",
        headers={"Authorization": f"Bearer {token}"}
    )
    assert me_resp.status_code == 200

dependency_overrides 模拟依赖

# 模拟认证依赖（测试受保护端点时无需真实 Token）
def mock_current_user():
    return {"id": 1, "username": "testuser", "role": "admin"}

@pytest.fixture
def authenticated_client():
    from app.routers.auth import get_current_user
    app.dependency_overrides[get_current_user] = mock_current_user
    with TestClient(app) as client:
        yield client
    app.dependency_overrides.clear()

def test_admin_endpoint(authenticated_client):
    # 不需要真实 JWT，依赖被替换为返回 mock 用户的函数
    response = authenticated_client.get("/admin/stats")
    assert response.status_code == 200

运行测试与覆盖率

# 运行所有测试
pytest

# 显示详细输出
pytest -v

# 生成覆盖率报告
pytest --cov=app --cov-report=html --cov-report=term-missing

# 仅运行某个测试文件
pytest tests/test_users.py -v

# 运行带特定标记的测试
pytest -m asyncio

# 覆盖率报告查看（HTML 版本）
# 打开 htmlcov/index.html