首页
/ FastAPI 异步测试指南:用 anyio 与 HTTPX AsyncClient 编写 async 测试用例

FastAPI 异步测试指南:用 anyio 与 HTTPX AsyncClient 编写 async 测试用例

2026-09-06 19:10:33作者:伍霜盼Ellen

编写 FastAPI 应用的测试通常使用 TestClient,但它是面向同步 def 测试函数的封装;当测试中需要等待数据库写入、调用其他 async 库验证后端结果时,就必须让测试函数本身异步运行。本文以官方文档 Tests Asíncronos 为骨架,结合本仓库源码,系统讲解如何借助 AnyIO 的 pytest 插件与 HTTPX 的 AsyncClient/ASGITransport 写出真正可 await 的 FastAPI 异步测试,并覆盖 lifespan 事件、事件循环陷阱等实战细节。读完本文你将掌握一套可复制的“同步接口测试 + 异步后端校验”测试模式。

为什么需要异步测试

此前在 Testing 教程 中看到的测试全部是同步写法:测试函数是普通的 def,调用 TestClient 时直接使用 client.get("/"),不出现 async/await。这套写法在绝大多数场景下简单够用。

但当你需要在发起请求后继续调用其他异步函数时——例如向 FastAPI 发送请求、验证后端是否成功地把数据写进了异步数据库——同步测试就无能为力了。官方文档点出的典型场景是:

向 FastAPI 应用发送请求,然后用某个异步数据库驱动核对后端是否正确写入了数据。

异步测试的核心诉求因此非常明确:让测试函数变成 async def,从而能在其中 await 任何异步操作,包括向应用发起请求本身。

前置认知:TestClient 为什么“不够用”

FastAPI 应用本质上是 async 的

即使你的路径操作函数用的是普通 def(FastAPI 会把它丢进线程池执行),整个 FastAPI 应用底层依然是 ASGI/async 应用。这一点从本仓库的实现可以直接佐证:fastapi/testclient.py 内容极短,它只是把 Starlette 的 TestClient 原样再导出:

from starlette.testclient import TestClient as TestClient  # noqa

fastapi/testclient.py 只是官方文档里所说的“为开发者提供的便利再导出”,真正的实现来自 Starlette。

TestClient 的“魔法”在 async 测试中会失效

TestClient 之所以能在普通 def 测试函数里调用 async 应用,是因为它在内部悄悄维护了一个事件循环(portal)来驱动 ASGI 应用。这套“魔法”在同步测试环境中工作良好,但一旦把 TestClient 放进 async 测试函数里,由于外层已经存在由 pytest/anyio 启动的事件循环,内层魔法与循环调度相互冲突,官方文档明确给出结论:

当我们以异步方式运行测试时,就不能再在测试函数内部使用 TestClient 了。

因此,异步测试需要换一条技术路线。

两大技术基座:anyio 与 HTTPX

pytest.mark.anyio:让 pytest 异步调用测试函数

如果测试函数是 async def,标准 pytest 默认不会正确执行它。AnyIO 提供的 pytest 插件解决了这一问题:用 @pytest.mark.anyio 标记某个测试函数,pytest 就会在 anyio 管理的事件循环中异步调用它。

本仓库的开发依赖中固定了 anyio 的版本区间(含 trio 后端支持):pyproject.toml 中声明 anyio[trio] >=3.2.1,<5.0.0,同时项目还直接依赖 pytest(pytest >=9.0.0 等,见 pyproject.toml)。anyio 被 Starlette 间接携带,通常无需额外手动安装。

HTTPX AsyncClient:直接面向 ASGI 的异步客户端

好消息是 TestClient 本身就基于 HTTPX 构建(这一点在 Testing 教程 开头也有说明),所以我们可以绕开 TestClient,直接用 HTTPX 原生的异步客户端测试 FastAPI

在 httpx 中,要用任意 ASGI 应用充当请求后端,需要配合 ASGITransport 使用,把应用“塞进”HTTP 客户端:

from httpx import ASGITransport, AsyncClient

httpx 从 0.23 起以 ASGITransport 承担这一职责,本仓库的依赖约束为 httpx >=0.23.0,<1.0.0(见 pyproject.toml),与示例代码完全一致。

完整示例:结构与代码

下面是一个最简可运行的工程结构(与 Aplicaciones Más GrandesTesting 中介绍的多文件布局一致):

.
├── app
│   ├── __init__.py
│   ├── main.py
│   └── test_main.py

其中 main.py 定义一个简单的 FastAPI 应用与路径操作。仓库中的真实示例文件位于 docs_src/async_tests/app_a_py310/main.py

from fastapi import FastAPI

app = FastAPI()


@app.get("/")
async def root():
    return {"message": "Tomato"}

test_main.py 则换成异步测试写法,对应仓库文件 docs_src/async_tests/app_a_py310/test_main.py

import pytest
from httpx import ASGITransport, AsyncClient

from .main import app


@pytest.mark.anyio
async def test_root():
    async with AsyncClient(
        transport=ASGITransport(app=app), base_url="http://test"
    ) as ac:
        response = await ac.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Tomato"}

测试文件与被测文件放在同一包内(同目录且有 __init__.py),因此使用相对导入 from .main import app

运行测试

运行方式与普通测试完全一致,直接用 pytest 即可:

$ uv run pytest

---> 100%

仓库中对此示例有对应的自动化回归测试:tests/test_tutorial/test_async_tests/test_main_a.py。该测试在自身函数上再次标记 @pytest.mark.anyio,并直接 await test_root() 调用示例中的异步测试,从侧面印证了“标记 anyio 的 async 测试可以被另一层 async 测试复用调用”这一机制:

import pytest

from docs_src.async_tests.app_a_py310.test_main import test_root


@pytest.mark.anyio
async def test_async_testing():
    await test_root()

逐行解析异步测试代码

@pytest.mark.anyio:标记异步执行的测试函数

@pytest.mark.anyio
async def test_root():

这个标记告诉 pytest:本测试函数应当异步调用。官方文档特别提示:注意此处测试函数是 async def,而此前使用 TestClient 时是普通 def——这是两种测试范式最直观的区别。

AsyncClient + ASGITransport:异步发起请求

    async with AsyncClient(
        transport=ASGITransport(app=app), base_url="http://test"
    ) as ac:
        response = await ac.get("/")

这里有三个要点:

  1. ASGITransport(app=app):把 FastAPI 应用注册为 HTTPX 的传输后端,请求不会走真实网络,而是直接驱动 ASGI 应用。
  2. base_url="http://test":指定基础 URL,供相对路径拼接使用(可任意取一个合法的测试域名,如 http://test)。
  3. async with + await:客户端以异步上下文管理器方式进入,ac.get("/") 需要用 await 等待。

这段写法与旧式同步调用严格等价。过去我们用 TestClient 时写的是:

response = client.get('/')

现在则变成 await ac.get("/"),返回的 response 同样是 HTTPX 的 Response 对象,因此 response.status_coderesponse.json() 等断言方式与 TestClient 时代完全一致。

断言保持不变

    assert response.status_code == 200
    assert response.json() == {"message": "Tomato"}

响应对象 API 未变,原有断言经验可以无缝迁移。

注意:AsyncClient 不会触发 lifespan 事件

官方文档给出了重要警告:

如果你的应用依赖 lifespan 事件(启动/关闭逻辑),AsyncClient 不会触发这些事件。

TestClient 作为上下文管理器(with TestClient(app) as client:)进入时会驱动 lifespan;而直接用 AsyncClient + ASGITransport 的方式不会主动执行 startup/shutdown。若你的应用在启动阶段初始化了数据库连接池等资源,需要借助 asgi-lifespan 提供的 LifespanManager 包裹应用,确保事件被触发:

from asgi_lifespan import LifespanManager

# 例:在 async 测试中先启动 lifespan 再发请求
# async with LifespanManager(app):
#     async with AsyncClient(...) as ac:
#         ...

在测试中调用其他异步函数

把测试函数变成 async def 的最大收益在于:除了向 FastAPI 应用发送请求之外,你现在可以像写普通业务代码一样 await 任何其他 async 函数。例如:

  • 用异步数据库驱动查询断言数据确实写入了;
  • await 其他外部服务的异步客户端;
  • 在一个测试中串行/并发执行多个异步步骤。

这正是本文开篇“先发请求、再验数据库”场景的落点。

常见陷阱:Task attached to a different loop

当你把数据库连接对象等需要事件循环的实例在 async 测试中混用时,可能遇到:

RuntimeError: Task attached to a different loop

典型触发场景是使用 MongoDB 的 MotorClient。官方建议的规避原则是:

只在 async 函数内部(例如 @app.on_event("startup") 回调中)创建需要事件循环的对象实例。

即在应用启动回调里创建客户端,测试仅负责发请求与断言,让对象与事件循环的生命周期保持一致,避免跨循环复用引发上述错误。

从同步到异步:决策速查

场景 推荐写法
测试函数中无需 async 普通 def + TestClient(经典方式)
需要在测试中 await 数据库等 async 调用 async def + @pytest.mark.anyio + AsyncClient
应用依赖 lifespan/startup 逻辑 额外用 LifespanManager 包裹,或考虑其他能驱动 lifespan 的方式
依赖与语言无关(同一断言 API) HTTPX 的 Response.status_code.json()

小结

异步测试并不是要用新框架推翻旧习惯,而是对 TestClient 体系的自然延伸:底层同为 HTTPX,只需把测试函数改为 async def 并加上 @pytest.mark.anyio,再改用 AsyncClient(transport=ASGITransport(app=app)) 发请求,即可在测试中获得完整的异步能力。需要注意的是 lifespan 事件不会由 AsyncClient 自动触发,以及事件循环相关的对象生命周期管理问题。掌握了这两点,你就能把“接口请求”与“异步数据库校验”放进同一个测试函数中,覆盖此前同步 TestClient 难以企及的验证场景。

如果你需要回顾前置知识,可参考 Testing 教程(同步测试基础)与 Aplicaciones Más Grandes(多文件工程布局);完整的可运行示例与仓库级回归测试分别位于 docs_src/async_teststests/test_tutorial/test_async_tests 目录下。

登录后查看全文
热门项目推荐
相关项目推荐