FastAPI pytest-unit-testing完全指南

📂 所属阶段：第五阶段 — 工程化与部署（实战篇）
🔗 相关章节：FastAPIdependency-injection · FastAPI多环境配置

单元测试是现代软件开发中确保代码质量和可维护性的核心实践。FastAPI 作为高性能 Python Web 框架，与 Pytest 深度集成，让编写测试变得简单而强大。本教程将从零开始，带你掌握 FastAPI 项目的测试全流程，包括同步/异步测试、数据库测试、覆盖率分析和 TDD 开发方法。

为什么需要单元测试？

在开始之前，我们先通过两个对比场景直观感受一下测试的价值。

没有测试的日常
你修改了用户登录逻辑，手动点了几下页面，感觉没问题。上线后第二天，客户反馈注册功能崩溃了。你不得不紧急回滚，忙到深夜，用户投诉不断 😱。久而久之，团队害怕修改代码，项目逐渐变成“不敢碰的烂摊子”。

拥有测试的日常
同样修改登录逻辑，你只需运行一条命令 pytest，所有测试在三秒内完成。发现一个注册相关测试失败，立即定位并修复。推送代码、自动构建、测试通过，自信上线 🚀。开发体验顺畅，代码始终保持高质量。

测试不是负担，而是开发过程中的安全网和设计工具。

理解测试金字塔

在 FastAPI 应用中，我们遵循经典的测试金字塔原则来分配各类测试的比重。

        ▲
       ╱ ╲
      ╱   ╲     ← 10% 端到端测试 (E2E)
     ╱─────╲      模拟用户完整操作流程
    ╱       ╲
   ╱─────────╲  ← 20% 集成测试 (Integration)
  ╱           ╲   测试 API 端点、数据库交互等
 ╱─────────────╲
╱               ╲ ← 70% 单元测试 (Unit Tests)
─────────────────── 验证单个函数/类行为
  • 单元测试：速度最快、最稳定，直接测试纯逻辑函数或类方法。
  • 集成测试：检查多个组件协作是否正常，比如路由 + 依赖注入 + 数据库。
  • 端到端测试：模拟真实用户行为，覆盖完整流程，成本最高，只覆盖核心链路。

本教程重点关注前两类：单元测试和集成测试。

environment-setup

安装核心依赖

在已有 FastAPI 项目的基础上，只需增加几个测试相关库：

pip install pytest pytest-asyncio httpx pytest-cov pytest-mock factory-boy faker
  • pytest：Python 测试框架。
  • pytest-asyncio：让 pytest 支持异步测试函数。
  • httpx：用于异步 HTTP 请求（代替 requests）。
  • pytest-cov：生成覆盖率报告。
  • pytest-mock：简化 mock 对象的使用。
  • factory-boy / faker：生成测试数据的神器，可选。

项目配置

推荐在 pyproject.toml 中统一管理 pytest 配置：

[tool.pytest.ini_options]
testpaths = ["tests"]
python_files = ["test_*.py", "*_test.py"]
python_classes = ["Test*"]
python_functions = ["test_*"]
asyncio_mode = "auto"
addopts = [
    "-ra",                # 显示所有简略结果
    "--showlocals",       # 失败时打印局部变量
    "--tb=short",         # 短回溯格式
    "--strict-markers",  # 未注册的标记会报错
    "--strict-config",   # 检测错误配置
]
markers = [
    "slow: 标记长时间运行的测试",
    "integration: 标记集成测试",
    "unit: 标记单元测试",
    "api: 标记 API 测试",
]

[tool.coverage.run]
source = ["src/", "app/"]   # 根据项目结构调整
omit = [
    "*/venv/*",
    "*/tests/*",
    "*/migrations/*",
    "*/__init__.py"
]

asyncio_mode = "auto" 会自动检测测试函数是否为异步，无需在每个测试上都加 @pytest.mark.asyncio（本文仍然显式标注以便阅读）。

创建 conftest.py 共享夹具

tests/conftest.py 是 pytest 的全局配置和夹具存放处。下面是一个完整示例：

import pytest
from httpx import AsyncClient, ASGITransport
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool
from fastapi.testclient import TestClient

from app.main import app
from app.database import get_db, Base

TEST_DATABASE_URL = "sqlite:///./test.db"

@pytest.fixture(scope="function")
def test_engine():
    """为每个测试函数创建独立的数据库引擎"""
    engine = create_engine(
        TEST_DATABASE_URL,
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )
    Base.metadata.create_all(bind=engine)
    yield engine
    Base.metadata.drop_all(bind=engine)
    engine.dispose()

@pytest.fixture(scope="function")
def test_session(test_engine):
    """提供 SQLAlchemy 会话"""
    Session = sessionmaker(bind=test_engine)
    session = Session()
    yield session
    session.rollback()
    session.close()

@pytest.fixture(scope="function")
def client(test_session):
    """同步 TestClient，并注入测试数据库会话"""
    def override_get_db():
        try:
            yield test_session
        finally:
            pass
    app.dependency_overrides[get_db] = override_get_db
    with TestClient(app) as test_client:
        yield test_client
    app.dependency_overrides.clear()

@pytest.fixture(scope="function")
async def async_client(test_session):
    """异步 AsyncClient，同样注入测试数据库"""
    def override_get_db():
        try:
            yield test_session
        finally:
            pass
    app.dependency_overrides[get_db] = override_get_db
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        yield ac
    app.dependency_overrides.clear()

这段配置为每个测试函数提供了全新的 SQLite 数据库，并通过依赖覆盖（dependency_overrides）将生产环境的数据库替换为测试库，实现完全隔离。

TestClient 同步测试入门

FastAPI 内置了 TestClient，基于 requests 库，可以像调用函数一样发送 HTTP 请求，非常适合编写同步测试。

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

@pytest.mark.unit
def test_root_endpoint(client):
    """测试根路径返回正常"""
    response = client.get("/")
    assert response.status_code == 200
    assert "message" in response.json()

@pytest.mark.api
def test_create_user_valid_data(client):
    """测试使用有效数据创建用户"""
    test_data = {"name": "Test User", "email": "test@example.com"}
    response = client.post("/users/", json=test_data)
    assert response.status_code == 201
    data = response.json()
    assert data["name"] == "Test User"
    assert data["email"] == "test@example.com"

@pytest.mark.unit
def test_404_for_unknown_endpoint(client):
    """测试不存在的路由返回 404"""
    response = client.get("/nonexistent-endpoint")
    assert response.status_code == 404
  • client 夹具由 conftest.py 提供，它已经完成了数据库依赖的替换。
  • 每个测试函数都是独立的，互不干扰。

异步测试详解

现代 FastAPI 项目大量使用 async/await，Pytest 通过 pytest-asyncio 和 httpx.AsyncClient 提供了原生的异步支持。

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

@pytest.mark.asyncio
@pytest.mark.api
async def test_async_root(async_client):
    """异步测试根端点"""
    response = await async_client.get("/")
    assert response.status_code == 200
    data = response.json()
    assert data["message"] == "Hello, World!"

@pytest.mark.asyncio
@pytest.mark.integration
async def test_async_create_user(async_client):
    """异步 POST 请求创建用户"""
    payload = {"name": "Async User", "email": "async@example.com", "password": "Str0ngPass!"}
    response = await async_client.post("/users/", json=payload)
    assert response.status_code == 201
    user = response.json()
    assert user["email"] == payload["email"]
    assert "id" in user

async_client 夹具提供了基于 ASGI transport 的异步客户端，完全模拟真实网络请求，但运行在内存中，速度极快。

Fixtures 与依赖注入测试

Pytest 的 fixture 系统是测试代码复用和隔离的核心。你可以自由组合 fixture，通过依赖注入的方式为测试函数准备环境。

# tests/test_fixtures.py
import pytest
from unittest.mock import AsyncMock
from app.services.user_service import UserService

@pytest.fixture
def mock_user_service():
    """模拟 UserService，避免访问真实数据库或外部 API"""
    mock = AsyncMock(spec=UserService)
    mock.get_user_by_id.return_value = {
        "id": 1,
        "email": "mock@example.com",
        "name": "Mock User"
    }
    return mock

@pytest.fixture
def sample_user_data():
    """标准测试数据"""
    return {
        "email": "test@example.com",
        "password": "TestPassword123!",
        "name": "Test User"
    }

class TestUserWithMockedService:
    @pytest.mark.unit
    def test_get_user_returns_mocked_data(self, mock_user_service):
        """直接测试模拟服务"""
        user = mock_user_service.get_user_by_id(1)
        assert user["email"] == "mock@example.com"

    @pytest.mark.api
    def test_create_user_with_sample_data(self, client, sample_user_data):
        """使用样本数据测试 API"""
        response = client.post("/users/", json=sample_user_data)
        assert response.status_code == 201
        assert response.json()["email"] == sample_user_data["email"]

最佳实践：

  • 将可复用的测试数据抽成 fixture，比如 sample_user_data。
  • 对于外部依赖（如支付服务、邮件服务），使用 unittest.mock 创建替身，保证测试快速且稳定。

数据库测试策略

数据库测试的核心是 事务回滚 和 独立数据库。前面已经通过 SQLite 内存库实现了隔离。接下来我们看如何直接测试数据库操作。

# tests/test_database.py
import pytest
from sqlalchemy import text
from app.models import User
from app.schemas import UserCreate

class TestUserCRUD:
    @pytest.mark.unit
    def test_create_user_persists_in_db(self, test_session):
        """测试用户创建后确实写入了数据库"""
        from app.crud import create_user

        user_data = UserCreate(
            email="dbuser@example.com",
            password="Secret123!",
            name="DB User"
        )
        user = create_user(test_session, user_data)
        assert user.email == user_data.email

        # 通过原生 SQL 进一步验证
        result = test_session.execute(
            text("SELECT COUNT(*) FROM users WHERE email = :email"),
            {"email": user_data.email}
        )
        count = result.scalar()
        assert count == 1

    @pytest.mark.unit
    def test_duplicate_email_rejected(self, test_session):
        """测试重复邮箱会被拒绝或抛出异常"""
        from app.crud import create_user
        from app.exceptions import DuplicateEmailError

        user_data = UserCreate(
            email="dupe@example.com",
            password="Password1!",
            name="First User"
        )
        create_user(test_session, user_data)

        with pytest.raises(DuplicateEmailError):
            create_user(test_session, UserCreate(
                email="dupe@example.com",
                password="Another1!",
                name="Second User"
            ))

要点：

  • 每个测试函数结束后，test_session 夹具自动回滚事务，数据库恢复干净状态。
  • 测试不仅验证成功路径，也要覆盖异常情况。

API 端点集成测试

集成测试验证整条“请求 → 路由 → 业务逻辑 → 数据库”链路。

# tests/test_user_api.py
import pytest

class TestUserAPI:
    @pytest.mark.integration
    @pytest.mark.asyncio
    async def test_full_user_lifecycle(self, async_client):
        """完整用户生命周期测试：创建 → 获取 → 更新 → 删除"""
        # 1. 创建
        new_user_data = {
            "email": "lifecycle@example.com",
            "password": "Lifecycle1!",
            "name": "Lifecycle User"
        }
        create_res = await async_client.post("/users/", json=new_user_data)
        assert create_res.status_code == 201
        user = create_res.json()
        user_id = user["id"]

        # 2. 获取
        get_res = await async_client.get(f"/users/{user_id}")
        assert get_res.status_code == 200
        assert get_res.json()["email"] == "lifecycle@example.com"

        # 3. 更新
        update_res = await async_client.put(f"/users/{user_id}", json={"name": "Updated User"})
        assert update_res.status_code == 200
        assert update_res.json()["name"] == "Updated User"

        # 4. 删除
        delete_res = await async_client.delete(f"/users/{user_id}")
        assert delete_res.status_code == 204

        # 5. 确认删除后获取返回 404
        get_again = await async_client.get(f"/users/{user_id}")
        assert get_again.status_code == 404

集成测试可以放心地走完整流程，因为它们使用了隔离的数据库。

参数化测试与边界测试

参数化让你用相同的测试逻辑覆盖多种输入，大幅减少重复代码。

# tests/test_parameterized.py
import pytest

class TestUserValidation:
    @pytest.mark.unit
    @pytest.mark.parametrize("email,expected_status", [
        ("valid@example.com", 201),
        ("user.name+tag@example.com", 201),
        ("plainaddress", 422),         # 无效邮箱
        ("@example.com", 422),         # 缺少用户名
        ("user@.com", 422),             # 无效域名
    ])
    @pytest.mark.asyncio
    async def test_email_validation(self, async_client, email, expected_status):
        """参数化测试邮箱格式校验"""
        payload = {
            "email": email,
            "password": "Password123!",
            "name": "Test User"
        }
        response = await async_client.post("/users/", json=payload)
        assert response.status_code == expected_status, f"邮箱 '{email}' 应返回 {expected_status}"

    @pytest.mark.unit
    @pytest.mark.parametrize("password,valid", [
        ("short", False),
        ("onlyLetters", False),
        ("NoDigits!", False),
        ("ValidPass1!", True),
        ("Str0ng!Pass", True),
    ])
    @pytest.mark.asyncio
    async def test_password_strength(self, async_client, password, valid):
        """参数化测试密码强度要求"""
        payload = {
            "email": "pwdtest@example.com",
            "password": password,
            "name": "Pwd User"
        }
        response = await async_client.post("/users/", json=payload)
        if valid:
            assert response.status_code == 201
        else:
            assert response.status_code == 422
  • 每个参数组合都会独立生成一个测试用例，失败时可以一目了然地知道哪组数据出了问题。
  • 非常适合测试验证逻辑、边界条件和业务规则。

测试覆盖率分析

覆盖率是衡量测试充分性的一个重要指标，但不能盲目追求 100%。合理的覆盖率目标通常在 80%～90% 之间。

生成覆盖率报告

# 运行测试并输出覆盖率
pytest --cov=app --cov-report=html --cov-report=term-missing --cov-fail-under=80
  • --cov=app：指定要统计的源码目录。
  • --cov-report=html：生成 HTML 报告，打开 htmlcov/index.html 即可可视化查看。
  • --cov-report=term-missing：在终端显示未覆盖的行号。
  • --cov-fail-under=80：若覆盖率低于 80%，则测试流程失败，适用于 CI/CD 管道。

覆盖率配置文件

除了命令行参数，你也可以在 .coveragerc 或 pyproject.toml 中配置更细粒度的规则：

# .coveragerc
[run]
source = app/
omit = 
    */venv/*
    */tests/*
    */__init__.py
branch = True

[report]
exclude_lines =
    pragma: no cover           # 标记为不纳入统计的代码
    def __repr__
    raise AssertionError
    raise NotImplementedError
    if __name__ == .__main__.:
show_missing = True

搭配编辑器插件（如 VS Code 的 Coverage Gutters），可以实时查看每一行代码是否被测试覆盖。

TDD 开发实践

测试驱动开发（TDD）的核心循环是：红灯 → 绿灯 → 重构。我们通过一个购物车功能的开发示例，来体会 TDD 的节奏。

第一步：红灯（写一个失败的测试）

class TestShoppingCartTDD:
    def test_new_cart_is_empty(self):
        """初始购物车应该是空的"""
        cart = ShoppingCart()
        assert len(cart.items) == 0
        assert cart.total_price == 0

此时运行测试会报错：NameError: name 'ShoppingCart' is not defined。这正是红灯——测试阻止了未实现的代码上线。

第二步：绿灯（用最少的代码让测试通过）

class ShoppingCart:
    def __init__(self):
        self.items = []
    
    @property
    def total_price(self):
        return sum(item["price"] * item["quantity"] for item in self.items)

再次运行测试，全部通过 ✅。我们只实现了刚够通过测试的功能，没有过度设计。

第三步：重构（优化内部实现，保持测试绿灯）

class ShoppingCart:
    def __init__(self):
        self._items = {}          # 改用字典存储，避免重复项
    
    def add_item(self, item):
        item_id = item["id"]
        if item_id in self._items:
            self._items[item_id]["quantity"] += item["quantity"]
        else:
            self._items[item_id] = item.copy()
    
    @property
    def items(self):
        return list(self._items.values())
    
    @property
    def total_price(self):
        return sum(item["price"] * item["quantity"] 
                  for item in self._items.values())

重构后测试依然通过，而代码结构更清晰，易于扩展（未来还可以添加 remove_item 等方法）。这就是 TDD 的基本循环：先定目标，再实现，最后改进。

在真实的 FastAPI 项目中，TDD 同样适用——先写一个 API 测试，然后实现路由和业务逻辑，最后重构（提取服务层、添加缓存等）。

总结

通过本指南，我们完整覆盖了 FastAPI 项目中 Pytest 测试的核心要点：

  1. 简洁的断言：assert 语句让测试易读易写。
  2. 强大的 Fixture 系统：轻松管理测试依赖，实现复用和隔离。
  3. 同步与异步测试：TestClient 和 AsyncClient 满足各种需求。
  4. 数据库测试策略：独立 SQLite 库 + 事务回滚，保证测试纯净。
  5. 参数化与覆盖率：覆盖更多边界条件，量化测试质量。
  6. TDD 实践：将测试融入开发流程，驱动高质量代码。

良好的测试套件就像项目的安全网和设计文档，它让你敢于重构、快速迭代、持续交付价值。

💡 下一步：尝试为你当前的项目添加一个测试文件，从最简单的 test_root_endpoint 开始，感受测试带来的自信。然后逐步扩展覆盖你的 API、业务逻辑和数据库层。

如果你正在使用 CI/CD 流水线，将 pytest 和覆盖率检查集成进去，每次提交都会自动验证代码质量。