问题现象

在用 pytest-asyncio 编写异步集成测试时,测试套件里混合了两类用例:

  • 通过 httpx.AsyncClient 调用 FastAPI 接口的用例
  • 在测试函数体内直接操作 Motor(MongoDB 异步驱动)的用例,例如 await test_db[...].insert_many() 或 await test_db[...].find_one()

运行后,第一类用例全部通过,第二类用例稳定报错:

tests\utils\helpers.py:23: in seed_data
    await test_db[coll_name].insert_many(docs)
E   RuntimeError: Task <Task pending name='Task-14' coro=<test_job_api() running at
    tests\api\test_jobs.py:18> cb=[_run_until_complete_cb() at
    C:\...\asyncio\base_events.py:182]> got Future <Future pending
    cb=[_chain_future.<locals>._call_check_cancel() at
    C:\...\asyncio\futures.py:391]> attached to a different loop

报错信息的核心是 attached to a different loop,说明当前 Task 拿到了一个绑定在另一个 event loop 上的 Future,两者不属于同一个 loop,asyncio 拒绝执行。


原因分析

问题的直接触发点

翻查 pytest.ini,发现有这样一行配置:

asyncio_default_fixture_loop_scope = session

这个配置来自 pytest-asyncio,它控制异步 fixture 默认使用哪个级别的 event loop。设置为 session 意味着:所有异步 fixture 在整个测试会话期间共享同一个 event loop。

fixture loop 与测试函数 loop 的错位

问题在于,asyncio_default_fixture_loop_scope 只影响 fixture,不影响测试函数本身。每个 async def test_xxx 函数默认仍然运行在自己独立的 function 级别 event loop 上。

这就造成了一个结构性错位:

Session Event Loop(Loop A)
  └── test_db fixture 在此创建 → Motor 客户端内部绑定 Loop A

test_job_api[用例1] → 运行在 Loop B(function loop)
  └── await test_db["jobs"].insert_many()
        Motor 内部持有的 Future 属于 Loop A
        但当前执行环境是 Loop B  →  冲突

test_job_api[用例2] → 运行在 Loop C(function loop)
  └── 同上  →  冲突

为什么只有直接操作数据库的用例报错

通过 httpx.AsyncClient 调用 FastAPI 接口的用例,数据库操作发生在应用服务内部,Motor 客户端在那个上下文里的 loop 是一致的,所以不会触发冲突。

而在测试函数体内直接 await test_db[...].insert_many() 时,Motor 客户端是在 session loop(Loop A)里创建的,但 await 的执行环境是 function loop(Loop B),两者不一致,错误就在这里暴露出来。

解决方案

修改配置

找到 pytest.ini(或 pyproject.toml 中的 [tool.pytest.ini_options]),将 asyncio_default_fixture_loop_scopesession 改为 function

[pytest]
asyncio_mode = auto

# 修改此处:session → function
asyncio_default_fixture_loop_scope = function

改动后的运行逻辑

改为 function 后,每个测试函数和它依赖的所有 fixture 都在同一个 event loop 中创建和运行,Motor 客户端绑定的 loop 和 await 执行的 loop 完全一致,问题就消失了:

test_job_api[用例1] → Loop X
  ├── test_db fixture    → Motor 客户端绑定 Loop X  
  ├── client fixture     → httpx.AsyncClient 在 Loop X  
  └── await insert_many  → 执行在 Loop X  

test_job_api[用例2] → Loop Y
  ├── test_db fixture    → 新的 Motor 客户端绑定 Loop Y  
  └── await insert_many  → 执行在 Loop Y  

修复后的测试结果

tests\api\test_jobs.py::test_job_api[职位_列表_有数据]       PASSED
tests\api\test_jobs.py::test_job_api[职位_列表_关键词搜索]   PASSED
tests\api\test_jobs.py::test_job_api[职位_列表_分页]         PASSED
tests\api\test_jobs.py::test_job_api[职位_热门_默认]         PASSED
tests\api\test_jobs.py::test_job_api[职位_详情_存在]         PASSED
tests\flows\test_job_flows.py::TestJobCoreFlows::test_hotjobs_flow PASSED
======================== all passed ========================

延伸说明

session 级别的 loop scope 本身并不是错误的配置,它在某些场景下有合理的用途,比如需要在多个测试用例之间共享数据库连接以减少初始化开销。但一旦使用这个配置,就必须同时确保测试函数本身也运行在同一个 session loop 上,否则就会出现本文描述的问题。

如果确实需要 session 级别的共享连接,可以通过显式声明 @pytest.fixture(scope="session") 并配合 loop_scope 参数来精确控制,而不是依赖全局默认值。这样做的好处是意图更明确,出问题时也更容易定位。

Motor 客户端必须在它将要被使用的同一个 event loop 中创建,这是这类问题的根本约束。只要 fixture 的 loop scope 与测试函数的 loop scope 保持一致,attached to a different loop 就不会出现。

Logo

腾讯云面向开发者汇聚海量精品云计算使用和开发经验,营造开放的云计算技术生态圈。

更多推荐