Python 如何测试异步函数?

推荐使用 pytest-asyncio 插件,通过 @pytest.mark.asyncio 装饰器或 pytest.ini 配置 asyncio_mode = auto,使测试函数支持 async/await;简单验证可用 asyncio.run(),但每次新建事件循环且不可复用。

直接用 asyncio.run()pytest-asyncio 运行异步函数即可,关键是要让测试框架理解 awaitable 对象,不能像同步函数那样直接调用。

用 asyncio.run() 手动运行

适合简单验证或临时调试。把异步函数当作协程对象传给 asyncio.run(),它会自动创建事件循环并执行:

  • 定义一个异步函数,比如 async def fetch_data(): return "done"
  • 在测试中写 result = asyncio.run(fetch_data())
  • 然后用普通断言检查 result,例如 assert result == "done"

注意:每次调用 asyncio.run() 都会新建事件循环,不能复用;也不适用于需要共享状态或模拟事件循环行为的场景。

用 pytest-asyncio 插件(推荐)

这是最主流的方式,让 pytest 原生支持 async 测试函数。安装后只需加装饰器或配置,就能写 async def test_xxx():

  • 安装:pip install pytest-asyncio
  • 在测试函数上加 @pytest.mark.asyncio,或全局启用(在 pytest.ini 中设 asyncio_mode = auto
  • 函数内可直接 await 被测函数,比如 data = await fetch_data()
  • 支持 fixture 注入,包括 async fixture(如数据库连接),只要标记为 @pytest.fixture 并是 async 函数即可

测试带 await 的内部逻辑(Mock 协程)

当被测函数依赖其他异步调用(如 HTTP 请求、数据库查询),需要用 mock 替换它们,并让 mock 返回可 await 的对象:

  • AsyncMock(Python 3.8+)替代 Mock,它自带 __await__ 方法
  • 例如:mock_api = AsyncMock(return_value={"id": 1}),然后 result = await mock_api() 就能正常返回
  • 若用旧版本 Python,可用 return_value=coroutine + asyncio.Future 手动构造,但更麻烦

避免常见错误

容易踩的坑包括:

  • 忘记 await —— 写成 fetch_data() 而不是 await fetch_data(),结果得到一个 coroutine 对象而非实际返回值
  • 混用线程与事件循环 —— 比如在异步测试里调用 threading.Thread 后又试图 await 它,这不会按预期工作
  • 多次调用 asyncio.run() 在同一个测试里 —— 可能触发 “Event loop is closed” 错误,因为 Python 不允许嵌套 run
  • unittest 时没继承 IsolatedAsyncioTestCase(Python 3.8+)—— 导致 async de

    f test_...
    不被识别