第 16 章：测试体系（pytest + httpx）
前 15 章写了依赖注入、分层架构、认证、WebSocket，却没有一行代码在 CI 里跑过。本章解决「改完敢不敢发」：用 pytest 驱动 ASGI 应用，用依赖覆盖把测试数据库塞进应用，并让每个测试互不污染。
学习目标
- 理解单元测试 / 集成测试 / 端到端测试的分工，知道本教程为什么聚焦 API 集成测试
- 掌握
TestClient与httpx.AsyncClient + ASGITransport两种驱动方式及其取舍 - 能配置
pytest-asyncio，并用conftest.py组织client/db_session/auth_headers夹具 - 能设计测试数据库隔离方案（含 SQLite 内存库的
StaticPool要求），并对 201 / 422 / 404 / 401 / 403 写出稳定断言
16.1 测试分层：先想清楚测什么
- 单元测试：纯函数、Service 业务规则、Pydantic 校验器。毫秒级、维护成本低，用
pytest直接调。 - 集成测试：路由 + 依赖注入 + 数据库 + 序列化，走进程内 ASGI，不起真实网络，单例十毫秒级。
- 端到端：已部署进程、网关、外部服务。秒级、维护成本高，只在关键链路保留少量。
三层是性价比排序而非替代关系：单元测试最多、集成测试次之、端到端最少。本教程聚焦 API 集成测试——它覆盖 FastAPI 项目最容易出错的一段：参数声明（第 3、7 章）、依赖装配（第 6 章）、响应字段过滤（第 5 章）、状态码与权限边界（第 8、12 章）、事务提交与回滚（第 10、11 章）。这些纯单元测试测不到，端到端测试又太慢。
16.2 安装依赖与 pytest 配置
uv add --dev pytest pytest-asyncio httpx pytest-cov
# 测试库用 SQLite 时还需要异步驱动：uv add --dev aiosqlite# pyproject.toml
[tool.pytest.ini_options]
asyncio_mode = "auto" # async 测试与异步夹具无需逐个加标记
asyncio_default_fixture_loop_scope = "function" # 显式声明，消除 pytest-asyncio 告警
addopts = "-q --strict-markers" # 标记拼错直接报错，避免静默失效strict（默认）要求每个测试与夹具都写 @pytest.mark.asyncio，好处是显式；auto 少一层噪音，代价是看不出这是异步测试。技术栈统一的项目选 auto。
16.3 两种驱动方式：TestClient 与 AsyncClient
同步：TestClient
from fastapi.testclient import TestClient
from app.main import app
client = TestClient(app)
def test_health_sync() -> None:
response = client.get("/healthz")
assert response.status_code == 200 and response.json()["status"] == "ok"TestClient 是 Starlette 对 httpx 的封装：不监听任何端口，而是通过 ASGI 传输层把请求直接交给 app，所以不用起服务、不占端口，断点调试和调普通函数一样。注意不带 with 使用时 lifespan 不会执行，要触发启动/关闭事件得写 with TestClient(app) as client:。
异步：httpx.AsyncClient + ASGITransport
import httpx
async def test_health_async() -> None:
transport = httpx.ASGITransport(app=app) # 直接把请求送进 ASGI app
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
assert (await client.get("/healthz")).status_code == 200异步方案是异步应用的主力：测试函数是 async def，与应用同一个事件循环，夹具可直接操作 AsyncSession（造数据、断言落库），不会撞上跨循环冲突——这正是 TestClient 最容易出问题的地方。两者底层都是 httpx、都直接调用 ASGI app，AsyncClient 只是把同步外观换成原生协程。
16.4 conftest.py：组织夹具
conftest.py 是 pytest 的共享夹具文件，tests/ 下的测试直接使用其中定义的夹具，无需 import。
# tests/conftest.py
from collections.abc import AsyncGenerator
import httpx
import pytest_asyncio
from sqlalchemy.ext.asyncio import AsyncSession
from app.db.session import get_db
from app.main import app
API = "/api/v1"
@pytest_asyncio.fixture
async def db_session() -> AsyncGenerator[AsyncSession, None]:
"""绑定测试事务的会话，结束整体回滚（完整实现见 16.6）。"""
...
@pytest_asyncio.fixture
async def client(db_session: AsyncSession) -> AsyncGenerator[httpx.AsyncClient, None]:
async def override_get_db() -> AsyncGenerator[AsyncSession, None]:
yield db_session
app.dependency_overrides[get_db] = override_get_db
try:
async with httpx.AsyncClient(
transport=httpx.ASGITransport(app=app), base_url="http://test"
) as http_client:
yield http_client
finally:
app.dependency_overrides.clear() # 测试失败也要清，否则污染后续测试
@pytest_asyncio.fixture
async def auth_headers(client: httpx.AsyncClient) -> dict[str, str]:
"""先调 /auth/token 拿 token，再放进 Authorization 头（OAuth2 密码流吃表单）。"""
response = await client.post(
f"{API}/auth/token", data={"username": "alice@example.com", "password": "s3cret-password"}
)
assert response.status_code == 200, response.text
return {"Authorization": f"Bearer {response.json()['access_token']}"}异步夹具用 @pytest_asyncio.fixture 显式声明，不依赖隐式转换；夹具依赖用参数表达，pytest 自动构建、逆序销毁。
16.5 依赖覆盖：把测试数据库塞进应用
app.dependency_overrides 是个字典：键为原依赖函数对象，值为替代实现。FastAPI 遇到该依赖时直接调用替代实现，原来的 get_db 一行都不执行。
❌ 不推荐（测试里就地覆盖，忘记清理）：
async def test_create_article_bad(client: httpx.AsyncClient) -> None:
app.dependency_overrides[get_db] = lambda: test_session
assert (await client.post("/api/v1/articles", json={"title": "x"})).status_code == 201
# 覆盖没清理：后面的测试全部拿到这个已关闭的 test_session✅ 推荐（覆盖与清理放进夹具的 try / finally）：
app.dependency_overrides[get_db] = override_get_db
try:
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app)) as http_client:
yield http_client
finally:
app.dependency_overrides.clear()dependency_overrides 是模块级全局状态：app 在同一个 pytest 进程内共享，一次泄漏就会让后续测试连到已关闭的会话（ResourceClosedError）或误判覆盖仍生效，典型表现是「单独跑通过、全量跑失败」。测试执行链路如下：
16.6 测试数据库隔离与建表时机
方案 A（推荐）：每个测试一个事务，结束回滚
engine = create_async_engine(
"sqlite+aiosqlite:///:memory:",
connect_args={"check_same_thread": False}, # 内存库要被同一连接复用
poolclass=StaticPool, # 这两项缺一不可
)
@pytest_asyncio.fixture
async def db_session() -> AsyncGenerator[AsyncSession, None]:
async with engine.connect() as connection:
transaction = await connection.begin()
session = AsyncSession(
bind=connection,
expire_on_commit=False,
# 被测代码内部的 commit() 降级为 SAVEPOINT，不提交外层事务
join_transaction_mode="create_savepoint",
)
try:
yield session
finally:
await session.close()
await transaction.rollback()方案 B：每轮重建数据库（适合要真实验证迁移 / DDL 的场景）
@pytest_asyncio.fixture
async def db_session() -> AsyncGenerator[AsyncSession, None]:
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
async with async_sessionmaker(engine, expire_on_commit=False)() as session:
yield session
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.drop_all)为什么内存库必须配 StaticPool：:memory: 的生命周期等于那条连接的生命周期，默认池一旦换连接就是另一个空库，现象是 no such table: articles，而建表语句明明成功过。StaticPool 让整个池只维持一条连接并始终复用；check_same_thread=False 解除 aiosqlite 在独立线程执行 SQL 时的跨线程检查。方案 A 回滚后零残留、无 DDL、最快，是 CI 主力；方案 B 更简单，但每个测试都要建表删表。
建表时机：第 10 章把建表交给 Alembic，lifespan 只做连接池预热与资源释放。测试要在夹具里显式建表——httpx.ASGITransport 与不带 with 的 TestClient 都不会触发 lifespan，而在生产启动时 create_all 又会绕过迁移。
@pytest_asyncio.fixture(scope="session", loop_scope="session", autouse=True)
async def _create_schema() -> None:
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)要专门测试 lifespan 本身（例如「启动时注册了什么」），用 with TestClient(app) as client:，别和业务测试混在一起。会话级异步夹具必须声明 loop_scope="session"，否则 pytest-asyncio 会在每个测试函数的循环里复用它，报 attached to a different loop。
16.7 六类典型断言的写法
断言原则：状态码 + 一个关键业务字段。不要断言整个响应体（新增字段就会红），也不要断言完整错误文案（改个标点就失败）。健康检查的断言见 16.3。
# tests/test_articles.py
from datetime import UTC, datetime
import httpx
API = "/api/v1/articles"
async def test_create_201(client: httpx.AsyncClient, auth_headers: dict[str, str]) -> None:
response = await client.post(
API, json={"title": "用 pytest 测试 FastAPI", "body": "集成测试优先"}, headers=auth_headers
)
data = response.json()
assert response.status_code == 201
assert data["title"] == "用 pytest 测试 FastAPI" # 关键字段回显
assert isinstance(data["id"], int)
# 时间只断言「可解析且不晚于现在」，不断言具体值
assert datetime.fromisoformat(data["created_at"]) <= datetime.now(UTC)
async def test_empty_title_422(client: httpx.AsyncClient, auth_headers: dict[str, str]) -> None:
response = await client.post(API, json={"title": "", "body": "x"}, headers=auth_headers)
error = response.json()["detail"][0]
assert response.status_code == 422
assert error["loc"] == ["body", "title"] # 断言结构，不断言文案
assert error["type"] == "string_too_short"
async def test_missing_article_404(client: httpx.AsyncClient) -> None:
assert (await client.get(f"{API}/999999")).status_code == 404
async def test_without_token_401(client: httpx.AsyncClient) -> None:
response = await client.post(API, json={"title": "x", "body": "y"})
assert response.status_code == 401
assert response.headers["WWW-Authenticate"] == "Bearer"
async def test_delete_other_users_article_403(client: httpx.AsyncClient, auth_headers: dict[str, str]) -> None:
created = await client.post(API, json={"title": "alice 的文章", "body": "..."}, headers=auth_headers)
assert created.status_code == 201
# 换成第二个用户（bob）的 token，去访问 alice 的资源
login = await client.post("/api/v1/auth/token", data={"username": "bob@example.com", "password": "s3cret"})
response = await client.delete(
f"{API}/{created.json()['id']}", headers={"Authorization": f"Bearer {login.json()['access_token']}"}
)
assert response.status_code == 40316.8 覆盖率与测试边界
uv run pytest --cov=app --cov-report=term-missing--cov=app 只统计应用包，term-missing 会列出没有被任何测试执行到的行号——这才是最有用的部分。覆盖率衡量「执行到没有」，不衡量「验证过没有」：一行被走到但没有断言，覆盖率照样 +1。所以把它当下限保险（例如 CI 要求 ≥70%），别当目标；重点是边界与错误路径。
| 该测（有业务价值） | 不该测（成本高于收益） |
|---|---|
| Service 的业务分支：能借 / 不能借、库存足够 / 不足 | 框架行为：FastAPI 是否返回 422、Pydantic 是否报类型错 |
| 错误码与错误体的触发条件：404 / 401 / 403 / 409 | 第三方库实现：argon2 结果、SQLAlchemy 生成的 SQL |
| 权限边界：本人、他人、匿名、管理员四种身份 | 私有实现细节：断言内部函数被调用几次 |
状态流转与数据边界：软删除后再删、page=0、空串、超长输入 | 脆弱文案与外部服务真实调用（应打桩或指向测试替身） |
常见坑与排查
| 现象 | 原因 | 解决 |
|---|---|---|
| 单个测试通过，全量运行随机失败 | dependency_overrides 未清理，app 是进程级共享对象 | 覆盖放夹具，清理写在 finally |
async def functions are not natively supported | 没装 pytest-asyncio，或未启用 asyncio_mode | 装插件并配 asyncio_mode = "auto" |
no such table: articles，但建表确实跑过 | SQLite 内存库随连接销毁，新连接是新库 | poolclass=StaticPool + check_same_thread=False |
测试里 lifespan 没执行 | 不带 with 的 TestClient(app) 不触发启动/关闭事件 | 用 with TestClient(app) as client:，或把建表放夹具 |
| 上一个测试的数据出现在下一个测试 | 测试内 commit() 提交了外层事务，没有回滚 | 事务回滚方案 + join_transaction_mode="create_savepoint" |
httpx.AsyncClient(app=app) 报 TypeError | httpx 0.28 起移除了 app= 简写 | 改用 httpx.ASGITransport(app=app)，并保持 Starlette 较新 |
断言 created_at 精确值，CI 上偶发失败 | 时区、精度、数据库默认值差异 | 只断言可解析、在合理区间、不晚于当前时间 |
本章小结
| 要点 | 说明 |
|---|---|
| 分层 | 单元测试最多、集成测试为主力、端到端最少 |
| 依赖安装 | uv add --dev pytest pytest-asyncio httpx pytest-cov（SQLite 再加 aiosqlite） |
| 驱动方式 | 同步 TestClient；异步推荐 httpx.AsyncClient + ASGITransport，都不占端口、不起真实服务 |
| pytest 配置 | asyncio_mode = "auto" 省去逐个标记，是「简洁」与「显式」的取舍 |
| 夹具组织 | conftest.py 定义 db_session / client / auth_headers，依赖用参数声明 |
| 依赖覆盖 | app.dependency_overrides[get_db] = override，必须清理 |
| 数据库隔离 | 方案 A 事务回滚（推荐、最快）；方案 B 每轮重建（适合测 DDL） |
| 建表与断言 | 夹具里 Base.metadata.create_all，别依赖 lifespan；断言状态码 + 关键字段 |
练习题
- 为第 12 章的
/auth/token写三个测试：密码正确返回200且带access_token；密码错误与用户不存在都返回401，并断言两种错误响应无法区分账号是否存在。 - 给 16.6 节方案 A 补两个测试：一个在
POST创建资源后用同一个db_session查库确认存在；另一个查库确认读不到上一条记录。 - 把 16.3 的健康检查改成
with TestClient(app) as client，并写一个测试验证lifespan确实执行了（例如启动时写入app.state）。
下一章预告
测试让代码敢改，但「敢改」不等于「敢上」。下一章把应用搬进生产：多进程启动的约束、配置管理、多阶段 Docker 镜像、Nginx 反向代理与上线检查清单。