FastAPI 怎么写测试?TestClient 和 AsyncClient 该选哪个?
简化版
FastAPI 的测试有两条路:TestClient(同步)——基于 httpx 加 portal,它在内部起一个事件循环来跑你的异步应用,但测试函数本身是同步的,写起来最简单(client.get("/x") 直接拿结果),适合绝大多数接口测试;httpx.AsyncClient + ASGITransport(异步)——测试函数本身是 async def,需要 pytest-asyncio 或 anyio 插件,适合「测试里要直接 await 应用内部的异步代码」的场景(比如先用 AsyncSession 准备数据、再调接口验证)。两者都不发真实的网络请求——它们通过 ASGI 接口直接调用应用,所以又快又不需要起服务器。测试 FastAPI 最核心的工具是 app.dependency_overrides:它是一个字典,把「原依赖函数」映射到「测试替身」,用来替换数据库 session、当前用户、外部服务客户端——这是 FastAPI 依赖注入设计带来的最大测试便利,比 mock 模块路径可靠得多(不依赖 import 位置)。几个必须做对的细节:① 用完要 app.dependency_overrides.clear()(否则污染后续测试);② 数据库测试要隔离——每个测试用独立事务并回滚,或用独立的测试库;③ lifespan 默认不会执行——TestClient 要用 with TestClient(app) as client: 才会触发 startup/shutdown;④ 异步测试要注意事件循环作用域,数据库连接和测试要在同一个循环里。核心记忆:TestClient 同步简单、AsyncClient 用于异步场景;dependency_overrides 是核心武器;记得 clear 和用 with 触发 lifespan。
详细版
两种客户端对比:
TestClient | httpx.AsyncClient | |
|---|---|---|
| 测试函数 | 同步 def | async def |
| 需要插件 | 否 | pytest-asyncio / anyio |
| 内部机制 | 起 portal 跑事件循环 | 直接在当前循环跑 |
能否 await 应用代码 | ❌ | ✅ |
| lifespan | 需 with 语句 | 需 LifespanManager |
| 推荐场景 | 大多数接口测试 | 需要异步 fixture 时 |
# ① ★TestClient(同步,最常用)★
from fastapi.testclient import TestClient
from myapp.main import app
client = TestClient(app)
def test_read_item():
r = client.get("/items/1")
assert r.status_code == 200
assert r.json()["id"] == 1
def test_create():
r = client.post("/items", json={"name": "x", "price": 10})
assert r.status_code == 201
assert "Location" in r.headers
def test_validation():
r = client.post("/items", json={})
assert r.status_code == 422
assert r.json()["detail"][0]["loc"] == ["body", "name"]
# ★触发 lifespan(startup/shutdown)★
def test_with_lifespan():
with TestClient(app) as client: # ★★with 才会跑 lifespan★★
r = client.get("/health")
# ② ★httpx.AsyncClient(异步)★
import pytest, httpx
from httpx import ASGITransport
@pytest.mark.anyio # ★或 @pytest.mark.asyncio★
async def test_async():
transport = ASGITransport(app=app) # ★★不发真实网络请求★★
async with httpx.AsyncClient(transport=transport,
base_url="http://test") as ac:
r = await ac.get("/items/1")
assert r.status_code == 200
# ★配 lifespan(需要 asgi-lifespan 包)★
from asgi_lifespan import LifespanManager
@pytest.mark.anyio
async def test_with_lifespan():
async with LifespanManager(app):
async with httpx.AsyncClient(transport=ASGITransport(app=app),
base_url="http://test") as ac:
r = await ac.get("/health")
# ③ ★★dependency_overrides:测试的核心武器★★
from myapp.deps import get_db, get_current_user
def override_get_db():
db = TestingSessionLocal()
try: yield db
finally: db.close()
def override_get_current_user():
return User(id=1, name="tester", is_admin=False)
app.dependency_overrides[get_db] = override_get_db
app.dependency_overrides[get_current_user] = override_get_current_user
# ★用完清理★
app.dependency_overrides.clear() # ★★必须★★
# ④ ★完整的 conftest.py★
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from myapp.database import Base, get_db
from myapp.main import app
@pytest.fixture(scope="session")
def engine():
eng = create_engine("sqlite:///:memory:",
connect_args={"check_same_thread": False},
poolclass=StaticPool) # ★★内存库必须用 StaticPool★★
Base.metadata.create_all(eng)
yield eng
Base.metadata.drop_all(eng)
@pytest.fixture
def db_session(engine):
conn = engine.connect()
trans = conn.begin() # ★★外层事务★★
session = sessionmaker(bind=conn)()
yield session
session.close()
trans.rollback() # ★★回滚,测试间隔离★★
conn.close()
@pytest.fixture
def client(db_session):
def _get_db():
yield db_session
app.dependency_overrides[get_db] = _get_db
with TestClient(app) as c:
yield c
app.dependency_overrides.clear() # ★★清理★★
@pytest.fixture
def auth_client(client):
app.dependency_overrides[get_current_user] = lambda: User(id=1)
yield client
app.dependency_overrides.clear()
⚠️ 三个必须记住的点:①
app.dependency_overrides是 FastAPI 测试的核心,比unittest.mock.patch更可靠。用patch("myapp.routers.items.get_db")这种方式打桩,依赖于「在哪个模块 import 的」这个脆弱的细节——重构时把 import 挪个位置就失效了,而且报错很难懂。dependency_overrides是 FastAPI 官方提供的机制:它按「依赖函数对象本身」做映射,无论这个依赖被哪个路由、以什么路径引用,都会被替换掉。注意 key 必须是原始的依赖函数对象(不是字符串、也不是Depends(...)包装后的东西)。② 用完必须clear()。dependency_overrides挂在app对象上,而app通常是模块级的全局单例——一个测试设置的覆盖会一直生效到进程结束,导致「单独跑这个测试通过、整个套件跑就失败」或者反过来的诡异现象,而且测试顺序一变结果就变。最稳妥的做法是在 fixture 的 teardown 里clear(),或者用try/finally。③lifespan(startup/shutdown)默认不会执行。直接client = TestClient(app)然后client.get(...),应用的lifespan里的初始化代码(连接池、缓存预热、模型加载)一行都不会跑——如果你的路由依赖这些,测试会以「NoneType 没有属性」之类的莫名错误失败。必须用with TestClient(app) as client:(异步侧用asgi-lifespan的LifespanManager)。
完整版教学
一、两种客户端的机制
★ ★都不发真实网络请求★:
TestClient / AsyncClient(ASGITransport)
↓
★直接构造 ASGI 的 scope / receive / send★
↓
★调用 app(scope, receive, send)★
↓
收集 send 出来的响应消息 → 组装成 Response
★ → ★不占端口、不走 TCP、比起真服务器快几十倍★
★ ★TestClient 的内部机制(值得知道)★:
FastAPI 的 TestClient 继承自 httpx.Client
但应用是 ★异步的★,而测试函数是 ★同步的★
→ 用 ★anyio 的 blocking portal★ 桥接:
① 在★后台线程★里起一个事件循环
② 同步调用被转发到那个循环里执行
③ 阻塞等待结果返回
★ 影响:
✓ 测试代码可以是普通的 def
✗ ★测试函数里不能直接 await 应用的异步代码★
✗ ★事件循环在另一个线程 → 某些"同一循环"的假设会失效★
★ ★AsyncClient 的机制★:
ASGITransport 直接在 ★当前事件循环★ 里调用 app
→ ★测试代码和应用代码在同一个循环里★
→ ★可以在测试里 await 数据库、await 应用的异步函数★
★ ★怎么选(★实际判断★)★:
┌────────────────────────────────┬──────────────────┐
│ 只测 HTTP 接口的输入输出 │ ★TestClient★ │
│ 需要在测试里 await 异步 fixture │ ★AsyncClient★ │
│ 用异步 SQLAlchemy 准备测试数据 │ ★AsyncClient★ │
│ 测试 WebSocket │ ★TestClient★(有 │
│ │ websocket_connect)│
│ 团队不熟悉 pytest-asyncio │ ★TestClient★ │
└────────────────────────────────┴──────────────────┘
★ ★经验:能用 TestClient 就用它(少一层复杂度)★
★只有当测试本身必须异步时才上 AsyncClient★
★ ★pytest 的异步支持(两个选择)★:
# ① pytest-asyncio
# pytest.ini: asyncio_mode = auto ← ★★推荐,不用每个都加 mark★★
@pytest.mark.asyncio
async def test_x(): ...
# ② anyio(★FastAPI 官方文档用的★)
@pytest.fixture
def anyio_backend(): return "asyncio"
@pytest.mark.anyio
async def test_x(): ...
★ 两者别混用(会冲突)
★ ★常见报错★:
"async def functions are not natively supported"
→ ★没装/没配 pytest-asyncio 或 anyio★
"attached to a different loop"
→ ★fixture 和测试不在同一个事件循环★(见后文)
两种客户端都不发真实网络请求——它们直接构造 ASGI 的 scope/receive/send 调用应用,不占端口、不走 TCP,比起真服务器快几十倍。TestClient 的机制值得知道:应用是异步的而测试函数是同步的,它用 anyio 的 blocking portal 在后台线程里起一个事件循环来桥接——好处是测试代码可以写成普通 def,代价是测试函数里不能直接 await 应用的异步代码,而且事件循环在另一个线程,某些「同一循环」的假设会失效。AsyncClient 则在当前循环里直接调用应用,测试和应用在同一个循环,可以自由 await。经验判断是「能用 TestClient 就用它」(少一层复杂度),只有当测试本身必须异步时才上 AsyncClient——典型场景是「用异步 SQLAlchemy 准备测试数据」。
二、dependency_overrides 深入
★ ★为什么它比 mock 好★:
# ✗ mock 方式(★脆弱★)
@patch("myapp.routers.items.get_db") # ★依赖 import 路径★
def test_x(mock_db): ...
★ 问题:
① ★重构移动 import 就失效★
② ★同一个依赖在多个模块被 import 时要 patch 多处★
③ ★patch 错路径时不报错,只是没生效★(最坑)
# ✓ dependency_overrides(★官方机制★)
app.dependency_overrides[get_db] = fake_get_db
★ 按★函数对象★匹配 → ★与 import 位置无关★
★ ★覆盖的粒度★:
# ① 覆盖数据库
app.dependency_overrides[get_db] = override_get_db
# ② 覆盖当前用户(★跳过认证★)
app.dependency_overrides[get_current_user] = lambda: test_user
# ③ 覆盖外部服务客户端
app.dependency_overrides[get_payment_client] = lambda: FakePaymentClient()
# ④ 覆盖配置
app.dependency_overrides[get_settings] = lambda: Settings(debug=True)
★ ★注意 key 必须是"原始函数对象"★:
# deps.py
def get_db(): ...
# router.py
from .deps import get_db
@app.get("/x")
def x(db = Depends(get_db)): ...
# test.py
from myapp.deps import get_db # ★★同一个对象★★
app.dependency_overrides[get_db] = fake
★ ✗ 如果依赖是 ★类实例★(Depends(SomeClass())),key 要是那个实例
★ ✗ 如果用了 ★Annotated[X, Depends(get_db)]★,key 仍是 get_db
★ ★子依赖也能覆盖★:
def get_db(): ...
def get_repo(db = Depends(get_db)): return Repo(db)
def get_service(repo = Depends(get_repo)): return Service(repo)
# ★覆盖最外层★ → 里面的都不会执行
app.dependency_overrides[get_service] = lambda: FakeService()
# ★或只覆盖最底层★ → 中间层照常构造
app.dependency_overrides[get_db] = fake_db
★ ★选择:想测中间层逻辑就覆盖底层,想跳过整条链就覆盖顶层★
★ ★用 fixture 管理(★推荐★)★:
@pytest.fixture
def override_user():
def _override(user: User):
app.dependency_overrides[get_current_user] = lambda: user
yield _override
app.dependency_overrides.clear()
def test_admin_only(client, override_user):
override_user(User(id=1, is_admin=False))
assert client.get("/admin").status_code == 403
override_user(User(id=2, is_admin=True))
assert client.get("/admin").status_code == 200
★ ★★忘记 clear 的后果(真实案例)★★:
test_a 里覆盖了 get_current_user 返回管理员
→ ★test_b 里所有请求都是管理员身份★
→ ★权限测试全部"通过"(假绿)★
→ ★上线后发现权限根本没生效★
✓ ★永远在 fixture 的 teardown 里 clear★
✓ 或者用 autouse fixture 做全局清理:
@pytest.fixture(autouse=True)
def _clear_overrides():
yield
app.dependency_overrides.clear()
★ ★不能覆盖的东西★:
✗ ★中间件★(不是依赖)→ 要么改配置,要么测试时不加
✗ ★lifespan 里的初始化★ → 用不同的 settings 或 monkeypatch
✗ ★全局单例★(模块级的 client 对象)→ ★这就是为什么要用依赖注入★
dependency_overrides 比 mock 好的根本原因是「按函数对象匹配,与 import 位置无关」——而 patch("myapp.routers.items.get_db") 依赖 import 路径,重构一移动就失效,而且 patch 错路径时不报错、只是没生效(最坑的一种)。覆盖的粒度很灵活:数据库、当前用户(跳过认证)、外部服务客户端、配置都可以;子依赖链上任意一层都能覆盖——想测中间层逻辑就覆盖底层,想跳过整条链就覆盖顶层。忘记 clear() 的后果是真实的事故:test_a 里把当前用户覆盖成管理员,test_b 的权限测试就会全部假绿,上线后才发现权限根本没生效——所以要在 fixture 的 teardown 里 clear,或者写一个 autouse 的全局清理 fixture。最后要知道中间件、lifespan 初始化、模块级全局单例都不能用它覆盖——这正是「应该用依赖注入而不是全局单例」的理由。
三、数据库测试的隔离
★ ★三种隔离策略★:
┌────────────────────┬──────────────────────────────────┐
│ ★① 内存 SQLite★ │ ★最快★;★但和生产数据库行为不同★ │
│ ★② 事务回滚★ │ ★快 + 真实数据库★(★推荐★) │
│ ★③ 每次重建库★ │ 最干净;★最慢★ │
└────────────────────┴──────────────────────────────────┘
★ ★① 内存 SQLite 的坑★:
create_engine("sqlite:///:memory:")
★ ✗ ★每个连接是独立的内存库★ → 建的表在另一个连接里不存在
✓ 必须用 StaticPool:
create_engine("sqlite:///:memory:",
connect_args={"check_same_thread": False},
★poolclass=StaticPool★) # ★★所有连接共用一个★★
★ ✗ 其他差异:
- ★没有真正的 ALTER TABLE★
- ★类型宽松★(VARCHAR(10) 不限制长度)
- ★没有 JSONB / 数组 / 全文检索★
- ★外键约束默认关闭★(要 PRAGMA foreign_keys=ON)
- ★并发写会锁库★
→ ★★用了 PostgreSQL 特性就必须用真 PG 测试★★(testcontainers)
★ ★★② 事务回滚(推荐方案)★★:
@pytest.fixture
def db_session(engine):
connection = engine.connect()
transaction = connection.begin() # ★★外层事务★★
Session = sessionmaker(bind=connection)
session = Session()
yield session
session.close()
★transaction.rollback()★ # ★★丢弃所有改动★★
connection.close()
★ 原理:★所有操作在一个未提交的事务里,回滚即恢复★
★ ✗ 问题:★如果被测代码里有 commit(),事务就提交了★
✓ 解法:★嵌套事务(SAVEPOINT)★
session.begin_nested()
@event.listens_for(session, "after_transaction_end")
def restart_savepoint(sess, trans):
if trans.nested and not trans._parent.nested:
sess.begin_nested()
★ ★③ testcontainers(★最真实★)★:
from testcontainers.postgres import PostgresContainer
@pytest.fixture(scope="session")
def pg():
with PostgresContainer("postgres:16") as p:
yield p.get_connection_url()
★ ✓ ★和生产完全一致的数据库★
★ ✗ 需要 Docker、启动慢(几秒)
★ 折中:★session 级启动容器 + 函数级事务回滚★
★ ★异步数据库的测试★:
@pytest_asyncio.fixture
async def async_session(async_engine):
async with async_engine.connect() as conn:
trans = await conn.begin()
async_session = AsyncSession(bind=conn)
yield async_session
await async_session.close()
await trans.rollback()
★ ★关键:fixture 和测试必须在同一个事件循环★
→ pytest-asyncio 的 ★loop_scope★ 要匹配
→ 常见报错:★"attached to a different loop"★
✓ pytest-asyncio ≥0.23:
@pytest_asyncio.fixture(loop_scope="session")
@pytest.mark.asyncio(loop_scope="session")
★ ★测试数据的准备★:
✓ ★factory-boy / polyfactory★(★根据 Pydantic 模型自动生成★)
from polyfactory.factories.pydantic_factory import ModelFactory
class UserFactory(ModelFactory[UserIn]): ...
user = UserFactory.build() # ★随机但合法的数据★
✓ 或手写 fixture
✗ ★不要用固定的 JSON fixture 文件★(模型一改就全废)
数据库测试有三种隔离策略。内存 SQLite 最快但有一堆坑:每个连接是独立的内存库(必须用 StaticPool)、类型宽松、没有 JSONB 和数组、外键约束默认关闭——用了 PostgreSQL 特性就必须用真 PG 测试。事务回滚是推荐方案(快且用真实数据库)——原理是「所有操作在一个未提交的事务里,回滚即恢复」,但如果被测代码里有 commit() 事务就提交了,解法是用 SAVEPOINT 嵌套事务。testcontainers 最真实(和生产完全一致),折中做法是「session 级启动容器 + 函数级事务回滚」。异步数据库测试的关键是 fixture 和测试必须在同一个事件循环——不匹配会报 attached to a different loop,pytest-asyncio 0.23+ 要用 loop_scope 对齐。测试数据推荐用 polyfactory 根据 Pydantic 模型自动生成。
四、认证、外部服务与异步代码
★ ★① 认证的三种测试方式★:
# ★方式一:覆盖依赖(★最简单★)★
app.dependency_overrides[get_current_user] = lambda: test_user
★ ✓ 快、不用真的登录
★ ✗ ★绕过了认证逻辑本身★(认证逻辑要单独测)
# ★方式二:真实登录拿 token★
@pytest.fixture
def token(client, test_user):
r = client.post("/login", data={"username": "x", "password": "y"})
return r.json()["access_token"]
def test_me(client, token):
r = client.get("/me", headers={"Authorization": f"Bearer {token}"})
★ ✓ ★端到端验证了整个认证链路★
★ ✗ 慢(每次都要哈希密码)→ ★测试环境把 bcrypt rounds 调低★
# ★方式三:直接构造 token★
token = create_access_token({"sub": str(user.id)})
★ ✓ 快 + 走了真实的 token 校验逻辑
★ ★必须单独测的认证场景★:
□ ★无 token → 401★
□ ★token 过期 → 401★
□ ★token 签名错误 → 401★
□ ★权限不足 → 403★
□ ★只能操作自己的资源★
★ ★权限测试是最该写的测试★(漏洞后果最严重)
★ ★② 外部服务的处理★:
# ✓ ★方案一:依赖注入 + 覆盖(★最干净★)★
def get_http_client() -> httpx.AsyncClient: ...
@app.get("/x")
async def x(client = Depends(get_http_client)): ...
# 测试
app.dependency_overrides[get_http_client] = lambda: FakeClient()
# ✓ ★方案二:respx(拦截 httpx 请求)★
import respx
@respx.mock
def test_call_api(client):
respx.get("https://api.example.com/user").mock(
return_value=httpx.Response(200, json={"id": 1}))
r = client.get("/proxy-user")
assert r.json()["id"] == 1
★ ✓ ★不用改生产代码★,能验证请求参数
★ 对 requests 用 responses 库
# ✓ ★方案三:pytest-httpx★
def test_x(httpx_mock):
httpx_mock.add_response(url="...", json={...})
★ ★③ 测试异步的应用代码(不经过 HTTP)★:
@pytest.mark.asyncio
async def test_service_layer(async_session):
service = UserService(async_session)
user = await service.create(UserIn(name="x")) # ★直接测 service★
assert user.id is not None
★ ★分层测试的价值★:
- ★service 层测业务逻辑★(快、不用 HTTP)
- ★路由层测参数校验、状态码、权限★
- ★别把所有测试都写成端到端★
★ ★④ BackgroundTasks 的测试★:
@app.post("/send")
async def send(bt: BackgroundTasks):
bt.add_task(send_email, addr)
return {"ok": True}
★ TestClient 下 ★背景任务会在响应返回前执行完★
→ ★可以直接断言副作用★
def test_bg(client, monkeypatch):
called = []
monkeypatch.setattr("myapp.routers.send_email",
lambda a: called.append(a))
client.post("/send")
assert called == ["x@y.com"]
★ ★⑤ WebSocket 测试★:
def test_ws():
with client.websocket_connect("/ws") as ws:
ws.send_text("hello")
assert ws.receive_text() == "echo: hello"
★ TestClient 内置支持,★AsyncClient 不支持★(要用别的库)
认证的三种测试方式各有用途:覆盖 get_current_user 依赖最简单(但绕过了认证逻辑本身,认证要单独测)、真实登录拿 token 能端到端验证(慢,测试环境要把 bcrypt rounds 调低)、直接构造 token 是折中。权限测试是最该写的测试(无 token → 401、过期 → 401、权限不足 → 403、只能操作自己的资源)。外部服务有三种处理方式:依赖注入加覆盖(最干净)、respx 拦截 httpx 请求(不用改生产代码,还能验证请求参数)、pytest-httpx。分层测试很有价值——service 层测业务逻辑(快)、路由层测参数校验和权限,别把所有测试都写成端到端。两个细节:TestClient 下 BackgroundTasks 会在响应返回前执行完(可以直接断言副作用)、WebSocket 只有 TestClient 支持。
五、常见坑与调试
★ ★坑一:lifespan 没执行★
client = TestClient(app) # ★★不会跑 startup★★
client.get("/x") # → 依赖 lifespan 初始化的东西全是 None
✓ with TestClient(app) as client: ...
✓ 异步侧:async with LifespanManager(app):
★ ★坑二:全局 app 的状态污染★
app.dependency_overrides 没清 → ★测试互相影响★
app.state.xxx 被某个测试改了 → 后续测试受影响
✓ ★autouse fixture 做清理★
✓ 或者每个测试 ★create_app() 新建应用★(★最干净但慢★)
★ ★坑三:事件循环不匹配(异步测试)★
报错:"Task got Future attached to a different loop"
"Event loop is closed"
原因:★fixture 创建的连接和测试运行在不同的循环★
✓ pytest-asyncio ≥0.23 用 loop_scope 对齐:
@pytest_asyncio.fixture(loop_scope="session")
async def engine(): ...
@pytest.mark.asyncio(loop_scope="session")
async def test_x(engine): ...
✓ 或者★所有异步 fixture 都用函数级作用域★(慢但不出错)
★ ★坑四:TestClient 里的 async 数据库★
TestClient 在★另一个线程★跑事件循环
→ 在测试函数(主线程)里创建的 AsyncSession 和应用不在同一循环
→ ★报 "attached to a different loop"★
✓ ★这种场景就该用 AsyncClient★
★ ★坑五:响应流式内容★
r = client.get("/stream")
r.text # ★★会一次性读完★★
✓ 测试流式:
with client.stream("GET", "/stream") as r:
for line in r.iter_lines(): ...
★ ★坑六:测试顺序依赖★
test_create 建了数据,test_list 依赖它存在
→ ★-p no:randomly 时能过,随机顺序就挂★
✓ ★每个测试自己准备数据★
✓ 用 pytest-randomly 主动暴露这类问题
★ ★调试技巧★:
① ★raise_server_exceptions=False★:
client = TestClient(app, raise_server_exceptions=False)
→ ★让 500 错误返回响应而不是抛异常★(★测试异常处理器时必需★)
② ★看完整的响应★:
print(r.status_code, r.json())
③ ★打开 SQL 日志★:
create_engine(..., echo=True)
④ ★用 -s 看 print★,用 --pdb 出错时进调试器
⑤ ★检查 app.dependency_overrides 当前内容★
★ ★测试覆盖的优先级★:
① ★权限★(401/403/越权) ← ★最该写★
② ★参数校验★(422 的边界)
③ ★业务规则★(状态流转、金额计算)
④ ★错误路径★(不存在、冲突、外部服务失败)
⑤ 正常路径(happy path)
★ ★很多团队只写了 ⑤,而 bug 都在 ①~④★
六个常见坑里,「lifespan 没执行」和「全局 app 状态污染」最高频。异步测试的两个坑都和事件循环有关:fixture 和测试的循环不匹配会报 attached to a different loop(pytest-asyncio 0.23+ 用 loop_scope 对齐)、TestClient 在另一个线程跑循环所以不能配异步 session(这种场景该用 AsyncClient)。调试技巧里最有价值的是 TestClient(app, raise_server_exceptions=False)——让 500 返回响应而不是抛异常,测试自定义异常处理器时必需。最后是测试覆盖的优先级:权限 > 参数校验 > 业务规则 > 错误路径 > 正常路径——很多团队只写了正常路径,而 bug 全在前四类。
六、实践清单
★ 推荐的 conftest.py 骨架:
# conftest.py
import pytest
from fastapi.testclient import TestClient
@pytest.fixture(scope="session")
def engine(): ... # ★建表一次★
@pytest.fixture
def db_session(engine): ... # ★★事务回滚★★
@pytest.fixture
def client(db_session):
app.dependency_overrides[get_db] = lambda: db_session
with TestClient(app) as c: # ★★with 触发 lifespan★★
yield c
app.dependency_overrides.clear() # ★★清理★★
@pytest.fixture(autouse=True)
def _cleanup(): # ★★兜底清理★★
yield
app.dependency_overrides.clear()
@pytest.fixture
def auth_client(client, test_user):
app.dependency_overrides[get_current_user] = lambda: test_user
yield client
★ 检查清单:
□ ★用 with TestClient(app) 触发 lifespan★
□ ★dependency_overrides 用完 clear(autouse 兜底)★
□ ★数据库用事务回滚隔离★
□ ★内存 SQLite 配 StaticPool★
□ ★用了 PG 特性就用 testcontainers★
□ ★测试环境把 bcrypt rounds 调低★
□ ★权限测试覆盖 401/403/越权★
□ ★422 的字段级错误有断言★
□ ★外部 HTTP 用 respx 拦截★
□ ★异步测试的 loop_scope 对齐★
□ ★分层测试:service 层单独测★
□ ★用 pytest-randomly 暴露顺序依赖★
★ ★测试环境的 settings★:
class TestSettings(Settings):
database_url: str = "sqlite:///:memory:"
★bcrypt_rounds: int = 4★ # ★★默认 12,测试调到 4 快 200 倍★★
rate_limit_enabled: bool = False # ★关限流★
cache_backend: str = "null" # ★关缓存(避免测试间串)★
app.dependency_overrides[get_settings] = lambda: TestSettings()
★ 一句话总结:
★"TestClient 同步简单(内部用 portal 跑事件循环),AsyncClient 用于
测试本身要 await 的场景;dependency_overrides 是核心武器——
按函数对象匹配所以比 patch 路径可靠,但用完必须 clear;
lifespan 要用 with 才会执行;数据库用事务回滚做隔离。"★
推荐的 conftest.py 骨架里三个关键点:with TestClient(app) 触发 lifespan、fixture teardown 里 clear、再加一个 autouse 的兜底清理。检查清单里容易漏的:内存 SQLite 要配 StaticPool、测试环境把 bcrypt rounds 调低(默认 12 调到 4 能快 200 倍)、用 pytest-randomly 暴露顺序依赖。测试用的 settings 还应该关掉限流和缓存(避免测试之间互相串)。
记忆钩子:「FastAPI 测试有两条路:★TestClient(同步)★——内部用 ★anyio 的 blocking portal 在后台线程起一个事件循环★来桥接『异步应用 + 同步测试函数』,写起来最简单,★但测试函数里不能直接 await 应用的异步代码★;★httpx.AsyncClient + ASGITransport(异步)★——★测试和应用在同一个事件循环★所以能自由 await,需要 pytest-asyncio 或 anyio 插件。★两者都不发真实网络请求★(直接构造 ASGI 的 scope/receive/send 调用 app,不占端口、比真服务器快几十倍)。★经验:能用 TestClient 就用它,只有测试本身必须异步(如用异步 SQLAlchemy 准备数据)才上 AsyncClient★;★WebSocket 只有 TestClient 支持★。★核心武器是 app.dependency_overrides★——它★按依赖函数对象匹配,与 import 位置无关★,所以★比 patch(‘模块路径.get_db’) 可靠得多★(后者重构一移动就失效,★而且 patch 错路径不报错只是没生效★);可以覆盖数据库/当前用户/外部客户端/配置,★依赖链上任意一层都能覆盖(想测中间层就覆盖底层,想跳过整条链就覆盖顶层)★。★三个必做的细节★:★① 用完必须 clear()★——app 是模块级单例,★一个测试把用户覆盖成管理员会让后续权限测试全部假绿★,要在 fixture teardown 里清理并加一个 autouse 兜底;★② lifespan 默认不执行★——必须 ★with TestClient(app) as client★(异步侧用 asgi-lifespan 的 LifespanManager),否则连接池、缓存预热这些初始化一行都不跑;★③ 数据库要隔离★——推荐★事务回滚★(连接上开外层事务、测完 rollback;★但被测代码里有 commit 就会破功,要用 SAVEPOINT 嵌套事务★),★内存 SQLite 必须配 poolclass=StaticPool★(否则每个连接是独立的库),★用了 PG 特性(JSONB/数组)就必须用 testcontainers★。异步测试的头号报错是 ★‘attached to a different loop’★——fixture 和测试不在同一个循环,★pytest-asyncio ≥0.23 要用 loop_scope 对齐★;★TestClient 在另一个线程跑循环,所以配异步 session 必然报这个错,该换 AsyncClient★。其他实用点:★外部 HTTP 用 respx 拦截★(不改生产代码还能验证请求参数)、★TestClient 下 BackgroundTasks 会在响应返回前执行完所以能直接断言副作用★、★raise_server_exceptions=False 让 500 返回响应(测异常处理器时必需)★、★测试环境把 bcrypt rounds 从 12 调到 4 能快 200 倍★。★测试覆盖优先级:权限(401/403/越权)> 参数校验 > 业务规则 > 错误路径 > 正常路径★——很多团队只写了最后一项而 bug 全在前四项。」
七、常见误区与追问
- 误区:
client = TestClient(app)之后直接发请求就行。 这样写lifespan(startup/shutdown)完全不会执行。FastAPI 的lifespan里通常放着数据库连接池初始化、Redis 客户端创建、机器学习模型加载、缓存预热这些关键的启动逻辑——不执行的话,路由里访问app.state.redis会得到AttributeError,或者拿到None然后报「NoneType 没有 xxx 方法」这种令人困惑的错误。正确写法是用上下文管理器:with TestClient(app) as client:——进入时触发 startup、退出时触发 shutdown。异步侧的httpx.AsyncClient本身不管 lifespan,需要额外用asgi-lifespan包的LifespanManager:async with LifespanManager(app):。反过来说,如果你的测试不需要那些初始化(纯粹测参数校验),不用with反而更快——但要清楚自己在做什么。 - 误区:用
unittest.mock.patch替换数据库依赖,和dependency_overrides效果一样。patch在这个场景下脆弱得多。它按字符串路径工作:patch("myapp.routers.items.get_db")替换的是「myapp.routers.items这个模块命名空间里名为get_db的那个引用」。三个问题:① 重构时把 import 挪到别的模块,patch 路径就失效了;② 同一个依赖被多个路由模块 import 时,得 patch 多处;③ 最坑的是路径写错了不会报错——patch会成功创建一个 Mock,但你的代码根本不走那条路径,测试照常「通过」,实际上打的是真实数据库。而app.dependency_overrides[get_db] = fake是 FastAPI 内建的机制:它在解析依赖树时按函数对象本身查表,无论这个依赖被谁、以什么方式引用都会被替换。注意 key 必须是原始的依赖函数对象(如果依赖是类实例Depends(SomeClass()),key 就是那个实例;用了Annotated[X, Depends(get_db)]时 key 仍然是get_db)。 - 误区:测试跑完不用清理
dependency_overrides,反正每个测试都会重新设置。 会造成「假绿」这种最危险的测试失效。app通常是模块级的全局对象,dependency_overrides挂在它上面——一个测试设置的覆盖会一直生效到进程结束。真实的事故场景:test_admin_dashboard里把get_current_user覆盖成一个管理员用户来测试后台功能,忘了清理;后面的test_normal_user_cannot_delete期望得到 403,结果因为身份还是管理员而返回了 200——但如果这个测试写的是assert r.status_code != 500之类的宽松断言,它就静默通过了,权限漏洞被完全掩盖。更隐蔽的是测试顺序一变结果就变:单独跑通过、整套跑失败,或者反过来。解法有三层:① 在设置覆盖的 fixture 的 teardown 里clear();② 加一个autouse=True的兜底 fixture 在每个测试后清理;③ 用pytest-randomly随机化测试顺序,主动把这类隐藏的依赖暴露出来。 - 误区:数据库测试用内存 SQLite 最方便,反正 SQLAlchemy 会屏蔽差异。 屏蔽不了,而且有个必踩的坑。首先是配置问题:
create_engine("sqlite:///:memory:")时每个连接对应一个独立的内存数据库——你在 fixture 里建的表,应用通过另一个连接访问时根本不存在,报「no such table」;必须加poolclass=StaticPool和connect_args={"check_same_thread": False}让所有连接共用同一个。其次是行为差异:SQLite 的类型是宽松的(VARCHAR(10)不会限制长度,所以「字段超长」的测试在 SQLite 上永远通过、上线到 PG 就报错)、外键约束默认关闭、没有真正的ALTER TABLE(迁移测试做不了)、没有 JSONB、数组、ON CONFLICT的完整语义、没有全文检索、并发写会直接锁库。所以判断标准很清楚:如果你的代码用到了任何 PostgreSQL/MySQL 特有的功能,就必须用真实的数据库测试——用testcontainers在 session 级起一个容器、函数级用事务回滚做隔离,是兼顾真实性和速度的最佳组合。 - 误区:异步测试报「attached to a different loop」是 pytest-asyncio 的 bug。 是事件循环作用域配置的问题。异步资源(数据库引擎、连接池、Redis 客户端)在创建时会绑定到当时的事件循环上;如果你的 fixture 是
scope="session"(在 session 级循环里创建),而测试函数默认在每个测试各自的新循环里运行,那么测试里使用这个连接时就会发现「它属于另一个循环」——asyncio 直接抛错。pytest-asyncio 0.23+ 引入了loop_scope参数来解决:把 fixture 和测试的循环作用域对齐,@pytest_asyncio.fixture(loop_scope="session")配@pytest.mark.asyncio(loop_scope="session")。另一个更简单粗暴的方案是所有异步 fixture 都用函数级作用域(每个测试重建引擎,慢但绝不出错)。还有一类相关问题:在TestClient里使用异步数据库 session——TestClient通过 portal 在后台线程里跑事件循环,而你在测试函数(主线程)创建的 session 属于另一个循环,必然冲突;这种场景就该换成AsyncClient。 - 追问:
TestClient和AsyncClient到底该怎么选? 判断标准是**「测试代码本身需不需要await」。如果测试只是「发请求 → 断言响应」,TestClient是更好的选择**:测试函数是普通的def,不需要 pytest-asyncio,不用操心事件循环作用域,代码也更易读;而且它内置支持 WebSocket 测试(client.websocket_connect),AsyncClient不支持。需要换成AsyncClient的场景主要有三类:① 用异步 ORM 准备测试数据——await session.execute(insert(...))必须在异步函数里;② 需要直接await应用的内部函数(测 service 层);③ 需要精确控制并发(比如测试同时发 10 个请求的竞态)。有个折中方案值得知道:同一个项目里两者可以共存——大部分接口测试用TestClient,少数需要异步 fixture 的测试用AsyncClient。注意TestClient的一个限制:因为它在后台线程跑事件循环,任何「必须和应用在同一个循环」的东西都用不了,这也是遇到 loop 报错时最常见的原因。 - 追问:数据库测试的事务回滚方案有什么坑? 核心机制是:在 fixture 里先建立连接、开启一个外层事务、把 session 绑定到这个连接上,测试结束后
rollback()丢弃所有改动——比「每次 drop/create 表」快几十倍,又用的是真实数据库。最大的坑是被测代码里的commit():如果你的 service 层调用了session.commit(),外层事务就被提交了,回滚也回滚不了,数据留在了库里污染后续测试。解法是嵌套事务(SAVEPOINT)——在 session 上begin_nested(),并监听after_transaction_end事件,每当嵌套事务结束就重新开一个 SAVEPOINT,这样应用的commit()实际提交的是 SAVEPOINT、外层事务依然可以整体回滚。第二个坑是测试代码和应用代码必须共用同一个 session——所以dependency_overrides[get_db]要返回 fixture 里那个 session 实例,而不是新建一个。第三个坑是它测不了真正的事务行为(比如「并发下的行锁」「commit 后触发器是否执行」),那类测试需要真实提交。 - 追问:FastAPI 项目的测试该怎么分层? 按「快而多 → 慢而少」分三层。① service/领域层单元测试(最多)——直接调用业务函数,用真实的数据库 session(事务回滚隔离)或纯内存的假实现,不经过 HTTP:测业务规则(状态流转合不合法、金额算得对不对、库存扣减的边界)。这层最快、最容易定位问题,应该占测试数量的大头。② 路由层集成测试(次多)——用
TestClient发请求,重点测HTTP 层面的东西:参数校验(422 的字段级错误)、状态码(201 有没有带Location、204 有没有 body)、权限(未登录 401、无权 403、只能操作自己的资源)、响应结构(有没有泄露password字段)。③ 端到端测试(最少)——真的起服务、真的连数据库和依赖服务,只覆盖最关键的几条主流程。优先级上,权限测试是最该写的——因为它的失败后果是数据泄露,而测试成本极低(几行client.get加断言状态码)。反过来,很多团队只写了「正常路径」的测试,但线上 bug 几乎都出在权限、边界值、并发和错误路径上。
八、加强记忆
FastAPI 测试有两条路:TestClient(同步)——内部用 anyio 的 blocking portal 在后台线程起一个事件循环来桥接「异步应用 + 同步测试函数」,写起来最简单,但测试函数里不能直接 await 应用的异步代码;httpx.AsyncClient + ASGITransport(异步)——测试和应用在同一个事件循环所以能自由 await,需要 pytest-asyncio 或 anyio 插件。两者都不发真实网络请求(直接构造 ASGI 的 scope/receive/send 调用 app,不占端口、比真服务器快几十倍)。经验是「能用 TestClient 就用它,只有测试本身必须异步(比如用异步 SQLAlchemy 准备数据)才上 AsyncClient」;另外 WebSocket 只有 TestClient 支持。核心武器是 app.dependency_overrides——它按依赖函数对象匹配、与 import 位置无关,所以比 patch("模块路径.get_db") 可靠得多(后者重构一移动就失效,而且 patch 错路径时不报错、只是没生效);它能覆盖数据库、当前用户、外部客户端、配置,依赖链上任意一层都能覆盖(想测中间层就覆盖底层,想跳过整条链就覆盖顶层)。三个必做的细节:① 用完必须 clear()——app 是模块级单例,一个测试把当前用户覆盖成管理员会让后续的权限测试全部假绿,要在 fixture teardown 里清理并加一个 autouse 兜底;② lifespan 默认不执行——必须用 with TestClient(app) as client:(异步侧用 asgi-lifespan 的 LifespanManager),否则连接池、缓存预热这些初始化一行都不会跑;③ 数据库要隔离——推荐事务回滚(在连接上开外层事务、测完 rollback;但被测代码里有 commit() 就会破功,要用 SAVEPOINT 嵌套事务),内存 SQLite 必须配 poolclass=StaticPool(否则每个连接是独立的库),而用了 PG 特性(JSONB、数组)就必须用 testcontainers。异步测试的头号报错是 attached to a different loop——fixture 和测试不在同一个循环,pytest-asyncio ≥0.23 要用 loop_scope 对齐;TestClient 在另一个线程跑循环,所以配异步 session 必然报这个错,该换 AsyncClient。其他实用点:外部 HTTP 用 respx 拦截(不改生产代码还能验证请求参数)、TestClient 下 BackgroundTasks 会在响应返回前执行完,所以能直接断言副作用、raise_server_exceptions=False 让 500 返回响应(测自定义异常处理器时必需)、测试环境把 bcrypt rounds 从 12 调到 4 能快 200 倍。测试覆盖的优先级是:权限(401/403/越权)> 参数校验 > 业务规则 > 错误路径 > 正常路径——很多团队只写了最后一项,而 bug 几乎全在前四项。