📚 编程进阶教程站

第 17 章 · 测试异步代码 ​

本章目标：学会用 pytest-asyncio 与 anyio 运行 async 测试，理解事件循环作用域对 fixture 的影响，并写出基于 httpx.AsyncClient 的异步集成测试。

17.1 为什么原生 pytest 跑不了 async def ​

pytest 的收集器只会调用测试函数。async def test_xxx() 被调用时返回的是一个 coroutine 对象——它从未被 await，pytest 只能警告"coroutine was never awaited"并标记通过（假绿！）。所以 async 测试必须由插件接管执行，把协程丢进事件循环里跑完。

主流选择有两个：

插件事件循环特点
pytest-asyncio仅 asyncio功能全、fixture 集成深，FastAPI/SQLAlchemy 项目首选
anyio（自带 pytest 插件）asyncio + trio装 anyio 即得，适合需要兼容 trio 后端的库

17.2 pytest-asyncio：两种模式与基本用法 ​

pytest-asyncio 有 strict（默认）与 auto 两种发现模式：

  • strict：必须显式给每个 async 测试打 @pytest.mark.asyncio；
  • auto：所有 async def test_* 自动识别为 asyncio 测试。
toml
# pyproject.toml
[tool.pytest.ini_options]
asyncio_mode = "auto"          # 或 "strict"，团队二选一
python
# test_async_basic.py
import asyncio
import pytest


@pytest.mark.asyncio                 # strict 模式下必需；auto 模式可省略
async def test_runs_in_a_loop():
    assert asyncio.get_running_loop() is not None


async def fetch_user(client, uid):   # 被测的异步函数示例
    return {"id": uid}


@pytest.mark.asyncio
async def test_fetch_user():
    user = await fetch_user(None, 42)
    assert user["id"] == 42

假绿陷阱

如果忘了装插件也没打标记，pytest 会输出 PytestUnhandledCoroutineWarning 但结果可能仍是 PASSED。CI 中建议加 -W error::pytest.PytestUnhandledCoroutineWarning 让它显式失败。

17.3 事件循环作用域：loop_scope ​

pytest-asyncio 默认每个测试一个新的事件循环（function 级），隔离性最好。但有些资源绑定在特定循环上（如连接池），跨循环复用会报错：

text
RuntimeError: Task ... got Future ... attached to a different loop

解法是把相邻测试固定到同一循环：

python
# test_loop_scope.py
import asyncio
import pytest


@pytest.mark.asyncio(loop_scope="module")   # 本模块内所有测试共享 module 级循环
async def test_remember_loop():
    global _loop
    _loop = asyncio.get_running_loop()


@pytest.mark.asyncio(loop_scope="module")
async def test_same_loop():
    global _loop
    assert asyncio.get_running_loop() is _loop

官方建议：相邻测试尽量使用相同的 loop scope（同模块统一 module 是常见约定）。异步 fixture 同理，用 loop_scope 对齐：

python
# conftest.py
import pytest


@pytest.fixture(loop_scope="module", scope="module")
async def engine():
    """module 级异步引擎：建一次，整个模块共用同一个循环。"""
    eng = await create_test_engine()
    yield eng
    await eng.dispose()

原则是：fixture 的 scope ≤ 它绑定的 loop_scope，否则 function 级循环销毁后 module 级 fixture 还活着，就会踩到"different loop"错误。

17.4 anyio 插件：一份代码跑两个后端 ​

anyio 自带 pytest 插件，通过 anyio_backend fixture 决定后端，还能参数化让同一测试分别跑 asyncio 和 trio：

python
# test_anyio_demo.py
import anyio
import pytest


@pytest.mark.anyio                       # 需要 anyio_backend fixture 存在
async def test_sleep():
    await anyio.sleep(0.01)


@pytest.fixture
def anyio_backend():
    return "asyncio"                     # 换成 "trio" 即跑 trio 后端


# 参数化多后端：
@pytest.fixture(params=["asyncio", "trio"])
def anyio_backend(request):
    return request.param                 # 上面的测试会各跑两次

选型建议：应用项目（FastAPI 等）几乎都在 asyncio 上，用 pytest-asyncio；编写通用库且想宣称 trio 兼容时用 anyio。两者不要同时启用发现模式，避免重复执行。

17.5 实战：httpx.AsyncClient 异步集成测试 ​

现代 ASGI 应用（FastAPI/Starlette/Litestar）不需要真的起服务器——httpx 的 ASGITransport 直接把请求送进 app 对象：

python
# test_api_async.py
import httpx
import pytest
from fastapi import FastAPI

app = FastAPI()


@app.get("/hello/{name}")
async def hello(name: str):
    return {"msg": f"hello {name}"}


@pytest.fixture
async def client():
    # ASGITransport：进程内直连 ASGI app，无网络开销
    transport = httpx.ASGITransport(app=app)
    async with httpx.AsyncClient(
        transport=transport, base_url="http://test"
    ) as c:
        yield c


@pytest.mark.anyio
@pytest.fixture(anyio_backend="asyncio")
def backend():
    ...


@pytest.mark.asyncio
async def test_hello(client):
    resp = await client.get("/hello/agno")
    assert resp.status_code == 200
    assert resp.json() == {"msg": "hello agno"}


# 若改用 anyio 插件，则客户端 fixture 需要声明后端：
# import anyio
#
# @pytest.fixture
# def anyio_backend():
#     return "asyncio"
#
# @pytest.fixture
# async def client(anyio_backend):
#     transport = httpx.ASGITransport(app=app)
#     async with httpx.AsyncClient(transport=transport,
#                                  base_url="http://test") as c:
#         yield c

三个要点：

  1. 用 transport= 注入 ASGITransport(app)，而不是已弃用的 AsyncClient(app=app) 快捷方式；
  2. base_url 只是占位，请求不会出网；
  3. 客户端放进 async with + fixture，保证连接池正确关闭；配合 17.3 的 loop_scope 规则避免跨循环问题。

本章小结 ​

  • 原生 pytest 不执行协程，async 测试必须交给 pytest-asyncio 或 anyio；
  • strict 模式需显式 @pytest.mark.asyncio，auto 模式自动识别；
  • 事件循环默认 per-test；共享资源用 loop_scope="module" 把测试和异步 fixture 固定到同一循环；
  • anyio 插件通过参数化 anyio_backend 可同时验证 asyncio/trio；
  • 进程内集成测试用 httpx.ASGITransport(app) + AsyncClient(transport=...)。

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. 未安装任何异步插件时，直接写 async def test_xxx() 会发生什么？

2. pytest-asyncio 默认（不配置 loop_scope 时）的事件循环策略是？

3. 出现 "Future attached to a different loop" 错误，最常见的根因是？

4. 用 httpx 对 FastAPI 应用做进程内异步集成测试，推荐的写法是？

🛠️ 动手实践 ​

  1. 写一个异步的 rate_limiter 类（内部用 asyncio.sleep），并用 pytest-asyncio 编写至少 3 个异步测试覆盖"允许/限流/窗口重置"。
  2. 构造一个故意混用不同 loop_scope 的失败案例，观察报错信息，然后修复它并写下你的结论。
  3. 用 anyio 参数化后端的方式重写第 1 题，确认测试在 asyncio 与 trio 下都通过。

学会了确定性输入下的异步测试，下一章看看如何测试“未知输入空间”——Hypothesis 登场。