← 返回题目列表

FastAPI 怎么写测试?TestClient 和 AsyncClient 该选哪个?

中等 第 18 / 27 题 更新于 2026/08/03
FastAPI测试TestClienthttpxdependency_overrides

简化版

FastAPI 的测试有两条路TestClient(同步)——基于 httpxportal它在内部起一个事件循环来跑你的异步应用,但测试函数本身是同步的,写起来最简单(client.get("/x") 直接拿结果),适合绝大多数接口测试;httpx.AsyncClient + ASGITransport(异步)——测试函数本身是 async def需要 pytest-asyncioanyio 插件,适合「测试里要直接 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

详细版

两种客户端对比

TestClienthttpx.AsyncClient
测试函数同步 defasync def
需要插件pytest-asyncio / anyio
内部机制起 portal 跑事件循环直接在当前循环跑
能否 await 应用代码
lifespanwith 语句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-lifespanLifespanManager)。

完整版教学

一、两种客户端的机制

★ ★都不发真实网络请求★:
  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) 触发 lifespanfixture 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 包的 LifespanManagerasync 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=StaticPoolconnect_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
  • 追问:TestClientAsyncClient 到底该怎么选? 判断标准是**「测试代码本身需不需要 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-lifespanLifespanManager),否则连接池、缓存预热这些初始化一行都不会跑;③ 数据库要隔离——推荐事务回滚(在连接上开外层事务、测完 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 几乎全在前四项。