014、测试之道：使用Pytest进行单元测试、集成测试与异步测试

---

一、从线上一个诡异bug说起

上周排查一个生产环境问题，用户上传文件偶尔会超时，日志里没有任何异常堆栈。最初怀疑是Nginx配置问题，折腾半天无果。后来在本地用curl循环测试了上百次，终于复现了一次——服务端日志显示请求进来了，但业务逻辑根本没执行。

问题出在依赖注入的一个异步函数里，某个条件分支下忘了写await。这种问题在开发时很难发现，因为大多数测试用例都走了正常流程。这件事让我重新审视了测试策略：光有单元测试不够，集成测试覆盖不全更危险，而异步代码的测试必须用对工具和方法。

今天我们就用Pytest这把瑞士军刀，把FastAPI的测试体系彻底讲透。

---

二、Pytest基础：别再用unittest那套了

先看一个典型错误示范：

# test_wrong.py
import unittest
from fastapi.testclient import TestClient
from main import app

class TestApp(unittest.TestCase):
    def setUp(self):
        self.client = TestClient(app)
    
    def test_read_item(self):
        response = self.client.get("/items/42")
        self.assertEqual(response.status_code, 200)  # 太啰嗦了

Pytest的写法清爽得多：

# test_right.py
from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_read_item():
    response = client.get("/items/42")
    assert response.status_code == 200  # 直接用assert，多自然
    assert response.json()["item_id"] == 42  # 链式断言

关键优势：

1. 不用写类，函数就是测试用例
2. assert后面可以接任何表达式，失败时会自动输出详细信息
3. 夹具（fixture）系统比setUp/tearDown灵活十倍

---

三、夹具（Fixture）：测试的共享工具箱

夹具是Pytest的灵魂，很多人只用了皮毛。看个实际场景：

# conftest.py
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from main import app, get_db
from database import Base

# 数据库夹具 - 每个测试用独立的数据库
@pytest.fixture(scope="function")
def test_db():
    engine = create_engine("sqlite:///./test.db")
    TestingSessionLocal = sessionmaker(bind=engine)
    
    # 创建表
    Base.metadata.create_all(bind=engine)
    
    db = TestingSessionLocal()
    try:
        yield db
    finally:
        db.close()
        # 清理表，避免测试间相互影响
        Base.metadata.drop_all(bind=engine)

# 客户端夹具 - 依赖数据库夹具
@pytest.fixture
def client(test_db):
    def override_get_db():
        try:
            yield test_db
        finally:
            pass
    
    app.dependency_overrides[get_db] = override_get_db
    with TestClient(app) as c:
        yield c
    app.dependency_overrides.clear()  # 重要！一定要清理覆盖

使用起来极其顺手：

def test_create_item(client):
    response = client.post("/items/", json={"name": "测试物品"})
    assert response.status_code == 201
    data = response.json()
    assert data["name"] == "测试物品"
    assert "id" in data  # 验证生成了ID

经验之谈：

• conftest.py里的夹具对整个目录生效
• 作用域（scope）选function最安全，module级可能引入测试污染
• 夹具可以嵌套，形成依赖链
• 记得清理dependency_overrides，我在这栽过跟头

---

四、单元测试：专注单个组件

单元测试要快、要隔离。FastAPI的依赖注入系统让这变得简单：

# 测试纯函数
def test_password_hash():
    from auth import hash_password
    hashed = hash_password("mypassword")
    assert hashed != "mypassword"  # 确实加密了
    assert len(hashed) == 64  # SHA256输出长度

# 测试依赖项
def test_get_current_user():
    from auth import get_current_user
    from unittest.mock import Mock
    
    mock_request = Mock()
    mock_request.headers = {"Authorization": "Bearer valid_token"}
    
    # 模拟验证逻辑
    user = get_current_user(mock_request, token="valid_token")
    assert user.username == "testuser"

关键点：

• 用unittest.mock模拟外部依赖
• 每个测试只关注一个函数或类
• 测试边界条件和异常分支

---

五、集成测试：验证组件协作

集成测试要模拟真实调用链，但不用启动完整服务：

def test_full_flow(client, test_db):
    # 1. 创建用户
    user_resp = client.post("/users/", json={
        "username": "test",
        "password": "secret"
    })
    user_id = user_resp.json()["id"]
    
    # 2. 登录获取token
    login_resp = client.post("/login", data={
        "username": "test",
        "password": "secret"
    })
    token = login_resp.json()["access_token"]
    
    # 3. 用token创建物品
    item_resp = client.post(
        "/items/",
        json={"title": "我的物品"},
        headers={"Authorization": f"Bearer {token}"}
    )
    assert item_resp.status_code == 201
    
    # 验证数据库确实写入了
    from models import Item
    db_item = test_db.query(Item).first()
    assert db_item.title == "我的物品"
    assert db_item.owner_id == user_id  # 关键：验证关联关系

踩坑提醒：

• 测试顺序很重要，上面的测试依赖下面的夹具
• 用test_db直接查库验证，比只检查API响应更可靠
• 集成测试可以暴露接口设计缺陷，比如缺少必要的关联查询

---

六、异步测试：FastAPI的核心难点

异步测试最容易出错，重点看三种写法：

import pytest
from httpx import AsyncClient

# 方法1：使用AsyncClient - 推荐
@pytest.mark.asyncio
async def test_async_endpoint():
    async with AsyncClient(app=app, base_url="http://test") as ac:
        response = await ac.get("/async-endpoint")
    assert response.status_code == 200

# 方法2：测试异步依赖
@pytest.mark.asyncio
async def test_async_dependency():
    from dependencies import async_redis_client
    
    # 直接调用异步依赖函数
    value = await async_redis_client.get("key")
    assert value is None  # 或具体的预期值

# 方法3：测试后台任务
import asyncio
from tasks import process_upload

@pytest.mark.asyncio
async def test_background_task():
    # 模拟文件上传
    mock_file = Mock()
    mock_file.read.return_value = b"test content"
    
    # 直接调用任务函数
    result = await process_upload(mock_file)
    assert result["status"] == "processed"
    
    # 验证副作用，比如数据库写入
    from models import TaskLog
    log = test_db.query(TaskLog).first()
    assert log.filename == "upload.txt"

特别注意事项：

1. 一定要加@pytest.mark.asyncio装饰器
2. TestClient不支持异步，必须用AsyncClient
3. 异步夹具要用@pytest_asyncio.fixture
4. 测试BackgroundTasks时，最好直接测试任务函数本身

---

七、测试覆盖率与实战技巧

光写测试不够，要知道覆盖了哪些代码：

# 安装覆盖率工具
pip install pytest-cov

# 运行测试并生成报告
pytest --cov=app --cov-report=html tests/

# 查看哪些行没覆盖到
pytest --cov=app --cov-report=term-missing tests/

我的经验规则：

1. 控制器层（路由）重点测状态码和响应格式
2. 业务逻辑层测所有分支，包括异常路径
3. 数据访问层测CRUD和查询边界
4. 工具函数要求100%覆盖率

一个实用技巧——用monkeypatch模拟环境变量：

def test_config_with_env(monkeypatch):
    monkeypatch.setenv("DATABASE_URL", "sqlite:///test.db")
    from config import settings
    assert settings.database_url == "sqlite:///test.db"

---

八、个人建议

测试不是写完就完事的，要让它成为开发流程的一部分。我在团队里推行这些实践：

1. 本地提交前：跑一遍相关模块的测试，用pytest -xvs tests/test_specific.py快速验证
2. CI流水线：配置pytest --cov --cov-fail-under=80，覆盖率不达标就失败
3. 遇到bug时：先写一个复现bug的测试用例，修复后再确保它通过
4. 异步代码：用asyncio.run()在同步环境里调试，但测试一定要用异步方式

最后说个真事：我们有个服务曾经因为测试没覆盖timeout参数，上线后在高并发下频繁超时。后来补了个压力测试才发现问题。所以记住——测试不仅要覆盖“应该发生什么”，还要覆盖“可能发生什么”。

好的测试像安全网，让你敢在代码高空作业。开始可能觉得麻烦，等它救过你几次命，你就离不开了。