FastAPI: 测试 — pytest 全链路质量保障

Q: TestClient 和 AsyncClient 哪个好用？

简单测试用 TestClient（同步），需要测试 async 逻辑（如 WebSocket）用 AsyncClient。大多数场景 TestClient 就够了。

Q: 测试数据库用 SQLite 还是 PostgreSQL？

开发用 SQLite（快），CI 用 PostgreSQL（与生产一致）。注意 SQLite 不支持某些 PG 特性（如 RETURNING、JSONB）。

Q: fixture 的 scope 怎么选？

function （每个测试重建）最安全； session （整个测试会话共享）最快但隔离差。数据库用 function ，TestClient 用 function 。

Q: 如何测试 WebSocket？

用 AsyncClient.websocket_connect() 建立连接，发送和接收消息验证行为。

Q: 测试跑得太慢怎么办？

用 pytest-xdist 并行执行（ pytest -n auto ），事务回滚替代建表，单元测试替代集成测试。

Q: dependency_overrides 会不会影响其他测试？

会，因为是修改 app 全局对象。务必在 fixture 的 yield 后 app.dependency_overrides.clear() 。

最后更新：2026-08-26

1. 你将学到

TestClient (httpx) 基础：同步/异步测试客户端
测试数据库隔离：每个测试用独立数据库，用 fixture 管理
依赖覆盖：app.dependency_overrides 替换 DB/认证依赖
测试金字塔：单元 → 集成 → E2E 的分层策略
Alice 场景：PriceTracker 的完整测试套件

2. Alice 的真实故事

(1) 痛点：上线后才发现 Bug

Alice 上线 PriceTracker 后才发现 JWT Token 过期时间设成了 1 秒、批量导入跳过所有校验、Pro 用户的限流和 Free 一样。每次修 Bug 又引入新 Bug，Bob 抱怨"上周好的功能这周又坏了"。Alice 没有自动化测试，全靠手动在 Swagger UI 点来点去，每次回归测试要 2 小时。

(2) pytest 自动化测试的解法

pytest + httpx TestClient 让每个 API 端点都有自动化测试，dependency_overrides 替换真实数据库为测试数据库，每次 git push 自动运行全部测试，2 分钟内发现所有回归问题。

PYTHON

def test_create_product(client):
    response = client.post("/api/v1/products", json={"name": "Widget", "price": 9.99})
    assert response.status_code == 201
    assert response.json()["name"] == "Widget"

(3) 收益

回归测试从手动 2 小时变成自动 2 分钟，JWT 过期时间 Bug 在开发阶段就被测试捕获，上线后 Bug 数从每周 5 个降到每月 1 个。

3. TestClient 基础

(1) 测试金字塔

graph TD
    E2E[E2E Tests - Few] --> INT[Integration Tests - Medium]
    INT --> UNIT[Unit Tests - Many]
    
    UNIT --- U1[Pydantic Model Validation]
    UNIT --- U2[Repository Functions]
    UNIT --- U3[Service Logic]
    
    INT --- I1[API Endpoint + DB]
    INT --- I2[Auth Flow]
    INT --- I3[CRUD Operations]
    
    E2E --- E1[Full User Journey]
    E2E --- E2[WebSocket + API]

层级	数量	速度	依赖
单元测试	多（100+）	快（< 1ms）	无外部依赖
集成测试	中（30-50）	中（10-100ms）	测试数据库
E2E 测试	少（5-10）	慢（1-5s）	完整环境

▶ 示例：TestClient 基础 fixture

PYTHON

# tests/conftest.py
import pytest
from fastapi.testclient import TestClient
from app.main import app
from app.core.deps import get_db

# Override database dependency
def get_test_db():
    # Use in-memory SQLite for testing
    engine = create_async_engine("sqlite+aiosqlite:///test.db")
    # ... session setup
    yield session
    # ... cleanup

@pytest.fixture
def client():
    app.dependency_overrides[get_db] = get_test_db
    with TestClient(app) as c:
        yield c
    app.dependency_overrides.clear()

TEXT 📖 仅展示

# 函数定义成功

▶ 示例：基础端点测试

PYTHON

# tests/test_health.py
def test_health_check(client):
    response = client.get("/health")
    assert response.status_code == 200
    data = response.json()
    assert data["status"] == "healthy"
    assert "service" in data

def test_openapi_docs_available(client):
    response = client.get("/docs")
    assert response.status_code == 200

TEXT 📖 仅展示

# 函数定义成功

4. 测试数据库隔离

(1) 每个测试独立数据库

▶ 示例：测试数据库 fixture

PYTHON

# tests/conftest.py
import pytest
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from app.models import Base

TEST_DATABASE_URL = "sqlite+aiosqlite:///test_pricetracker.db"

@pytest.fixture(scope="function")
async def test_db():
    # Create fresh test database for each test
    engine = create_async_engine(TEST_DATABASE_URL, echo=False)
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    
    TestSession = async_sessionmaker(engine, expire_on_commit=False)
    async with TestSession() as session:
        yield session
    
    # Drop all tables after test
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
    await engine.dispose()

TEXT 📖 仅展示

# 函数定义成功

▶ 示例：异步测试客户端

PYTHON

# tests/conftest.py
import pytest
from httpx import AsyncClient, ASGITransport
from app.main import app

@pytest.fixture
async def async_client():
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        yield ac

TEXT 📖 仅展示

# 函数定义成功

(2) 隔离策略对比

策略	速度	隔离性	适用
每个测试新建表	慢	完全隔离	数据完整性测试
事务回滚	快	好	大部分集成测试
内存 SQLite	快	好	不依赖 PG 特性的测试

5. 依赖覆盖技巧

(1) 替换认证和数据库

▶ 示例：覆盖认证依赖

PYTHON

# tests/conftest.py
from app.core.deps import get_current_user, get_db

def get_test_user():
    """Mock authenticated user for testing"""
    return {"id": 1, "email": "alice@test.com", "role": "admin", "subscription": "pro"}

@pytest.fixture
def auth_client(client):
    # Override authentication - no real JWT needed
    app.dependency_overrides[get_current_user] = get_test_user
    yield client
    app.dependency_overrides.clear()

TEXT 📖 仅展示

# 函数定义成功

▶ 示例：测试受保护端点

PYTHON

# tests/test_products.py
def test_list_products_unauthorized(client):
    """Without auth token, should return 401"""
    response = client.get("/api/v1/products")
    assert response.status_code == 401

def test_list_products_authorized(auth_client):
    """With mocked auth, should return 200"""
    response = auth_client.get("/api/v1/products")
    assert response.status_code == 200

def test_create_product(auth_client):
    response = auth_client.post(
        "/api/v1/products",
        json={"name": "Widget", "category": "electronics", "base_price": 29.99},
    )
    assert response.status_code == 201
    assert response.json()["name"] == "Widget"

TEXT 📖 仅展示

# 函数定义成功

▶ 示例：覆盖订阅级别测试权限

PYTHON

def get_free_user():
    return {"id": 2, "email": "free@test.com", "role": "user", "subscription": "free"}

def get_pro_user():
    return {"id": 3, "email": "pro@test.com", "role": "user", "subscription": "pro"}

def test_bulk_import_free_user_limited(client):
    """Free users can import max 1000 prices"""
    app.dependency_overrides[get_current_user] = get_free_user
    prices = [{"product_id": i, "price": 9.99} for i in range(1500)]
    response = client.post("/api/v1/prices/bulk", json=prices)
    assert response.status_code == 403
    assert "limit" in response.json()["detail"].lower()
    app.dependency_overrides.clear()

def test_bulk_import_pro_user(client):
    """Pro users can import up to 100000 prices"""
    app.dependency_overrides[get_current_user] = get_pro_user
    prices = [{"product_id": i, "price": 9.99} for i in range(5000)]
    response = client.post("/api/v1/prices/bulk", json=prices)
    assert response.status_code == 201
    app.dependency_overrides.clear()

TEXT 📖 仅展示

# 函数定义成功

6. 完整测试套件示例

▶ 示例：Pydantic 模型单元测试

PYTHON

# tests/test_models.py
import pytest
from pydantic import ValidationError
from app.schemas import ProductCreate, PriceCreate

def test_product_create_valid():
    p = ProductCreate(name="Widget", category="electronics", base_price=29.99)
    assert p.name == "Widget"
    assert p.base_price == 29.99

def test_product_create_negative_price():
    with pytest.raises(ValidationError) as exc:
        ProductCreate(name="Widget", category="electronics", base_price=-1)
    assert "greater than 0" in str(exc.value)

def test_product_create_empty_name():
    with pytest.raises(ValidationError):
        ProductCreate(name="", category="electronics", base_price=9.99)

def test_price_create_rounds_precision():
    p = PriceCreate(product_id=1, price=9.999, currency="USD", source="test")
    assert p.price == 10.0  # Rounded to 2 decimal places

def test_price_create_invalid_currency():
    with pytest.raises(ValidationError):
        PriceCreate(product_id=1, price=9.99, currency="XYZ", source="test")

TEXT 📖 仅展示

# 函数定义成功

▶ 示例：CRUD 操作集成测试

PYTHON

# tests/test_crud.py
import pytest
from fastapi.testclient import TestClient

def test_product_crud_lifecycle(auth_client):
    # Create
    create_resp = auth_client.post(
        "/api/v1/products",
        json={"name": "Test Widget", "category": "electronics", "base_price": 19.99},
    )
    assert create_resp.status_code == 201
    product_id = create_resp.json()["id"]

    # Read
    get_resp = auth_client.get(f"/api/v1/products/{product_id}")
    assert get_resp.status_code == 200
    assert get_resp.json()["name"] == "Test Widget"

    # Update
    update_resp = auth_client.put(
        f"/api/v1/products/{product_id}",
        json={"name": "Updated Widget", "base_price": 24.99},
    )
    assert update_resp.status_code == 200
    assert update_resp.json()["base_price"] == 24.99

    # Delete
    delete_resp = auth_client.delete(f"/api/v1/products/{product_id}")
    assert delete_resp.status_code == 200

    # Verify deleted
    get_resp2 = auth_client.get(f"/api/v1/products/{product_id}")
    assert get_resp2.status_code == 404

TEXT 📖 仅展示

# 函数定义成功

❓ 常见问题

Q TestClient 和 AsyncClient 哪个好用？

A 简单测试用 TestClient（同步），需要测试 async 逻辑（如 WebSocket）用 AsyncClient。大多数场景 TestClient 就够了。

Q 测试数据库用 SQLite 还是 PostgreSQL？

A 开发用 SQLite（快），CI 用 PostgreSQL（与生产一致）。注意 SQLite 不支持某些 PG 特性（如 RETURNING、JSONB）。

Q fixture 的 scope 怎么选？

A function（每个测试重建）最安全；session（整个测试会话共享）最快但隔离差。数据库用 function，TestClient 用 function。

Q 如何测试 WebSocket？

A 用 AsyncClient.websocket_connect() 建立连接，发送和接收消息验证行为。

Q 测试跑得太慢怎么办？

A 用 pytest-xdist 并行执行（pytest -n auto），事务回滚替代建表，单元测试替代集成测试。

Q dependency_overrides 会不会影响其他测试？

A 会，因为是修改 app 全局对象。务必在 fixture 的 yield 后 app.dependency_overrides.clear()。

📖 小节

测试金字塔：单元测试（多快）→ 集成测试（中等）→ E2E（少慢）
TestClient 同步测试简单端点，AsyncClient 测试异步逻辑（WebSocket 等）
每个测试用独立数据库，fixture 管理创建和清理
app.dependency_overrides 替换认证和数据库依赖，无需真实 JWT 和 PG
完整测试覆盖：Pydantic 校验（单元）、CRUD 流程（集成）、权限控制（集成）

📝 作业

基础题（难度⭐）：用 TestClient 编写 /health 端点的测试，验证返回 200 和正确 JSON 结构。提示：TestClient(app) + client.get("/health")
进阶题（难度⭐⭐）：编写 CRUD 生命周期测试——创建商品、查询商品、更新商品、删除商品，验证每步状态码和返回数据。用 dependency_overrides 跳过认证。提示：app.dependency_overrides[get_current_user] = mock_fn
挑战题（难度⭐⭐⭐）：实现完整测试套件：Pydantic 校验测试（非法数据抛 ValidationError）、认证测试（无 Token 401、Free 用户 403、Pro 用户 200）、分页测试（skip/limit 正确返回）。提示：pytest.raises(ValidationError) + 多个 dependency_overrides fixture

---|

Web-Tutorial 技术团队

由多位开发者共同维护的编程教程平台。每篇教程由对应领域的开发者编写和审核，确保内容准确可靠。如发现任何问题，欢迎向我们反馈。

关于我们

FastAPI: 测试 — pytest 全链路质量保障

1. 你将学到

2. Alice 的真实故事

(1) 痛点：上线后才发现 Bug

(2) pytest 自动化测试的解法

(3) 收益

3. TestClient 基础

(1) 测试金字塔

▶ 示例：TestClient 基础 fixture

▶ 示例：基础端点测试

4. 测试数据库隔离

(1) 每个测试独立数据库

▶ 示例：测试数据库 fixture

▶ 示例：异步测试客户端

(2) 隔离策略对比

5. 依赖覆盖技巧

(1) 替换认证和数据库

▶ 示例：覆盖认证依赖

▶ 示例：测试受保护端点

▶ 示例：覆盖订阅级别测试权限

6. 完整测试套件示例

▶ 示例：Pydantic 模型单元测试

▶ 示例：CRUD 操作集成测试

❓ 常见问题

📖 小节

📝 作业

🙏 帮我们做得更好