pytest-asyncio + Motor 报错 RuntimeError: Task got Future attached to a different loop
在使用 pytest-asyncio 测试 FastAPI + Motor 时,部分测试用例抛出 RuntimeError: Task got Future attached to a different loop?本文记录这个问题的完整排查过程与修复方案。
问题现象
在用 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_scope 从 session 改为 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 就不会出现。
更多推荐
所有评论(0)