TaskGroup在pytest中抛出RuntimeError: no running event loop,根本原因是其被用于无运行事件循环的上下文(如同步fixture、conftest.py初始化阶段),须严格限定在async test函数或启用strict模式的async fixture内,并配合event_loop fixture和pytest-asyncio插件使用。

TaskGroup在pytest中抛出RuntimeError: no running event loop
Python 3.11 引入的 asyncio.TaskGroup 要求必须在运行中的 event loop 内使用,而 pytest 默认不启动 loop(尤其在非 async test 函数里调用 async fixture 时)。常见报错是:RuntimeError: no running event loop 或 RuntimeError: Task group is closed。
根本原因不是 TaskGroup 本身有问题,而是它被创建在了没有 loop 的上下文中——比如同步 fixture、setup_method、或 pytest 的 setup 阶段。
- 只在真正的 async test 函数(即用
async def test_...定义的)内部使用TaskGroup,才安全 - 避免在
@pytest.fixture(未标记scope="function"且无autouse=True)中直接实例化TaskGroup - 若 fixture 必须异步初始化资源,改用
asyncio.create_task+await asyncio.gather(...)替代TaskGroup,更可控
如何正确编写支持TaskGroup的async fixture
pytest 原生不支持 async fixture,需依赖 pytest-asyncio 插件,并显式声明 asyncio_mode = "auto" 或用 @pytest.mark.asyncio 标记测试函数。但 fixture 本身不能是 async def ——除非你用 pytest-asyncio>=0.23.0 并启用 asyncio_mode = "strict"。
推荐做法:用 event_loop fixture 注入已启动的 loop,再在 async fixture 中用 asyncio.TaskGroup 管理子任务:
立即学习“Python免费学习笔记(深入)”;
import pytest
import asyncio
<p>@pytest.fixture
def event_loop():
loop = asyncio.new_event_loop()
yield loop
loop.close()</p><p>@pytest.fixture
async def db_pool(event_loop):
async with asyncio.TaskGroup() as tg:</p><h1>注意:此处 tg 是在 event_loop 已运行的前提下创建的</h1><pre class="brush:php;toolbar:false;"> tg.create_task(init_db())
tg.create_task(warm_up_cache())
return await get_pool()关键点:fixture 必须声明为 async def,且 pytest 配置中启用 strict 模式;否则会静默降级为同步执行,导致 TaskGroup 失效。
为什么在conftest.py里用TaskGroup常出错
很多用户把 TaskGroup 放进 conftest.py 的全局 fixture(如 pytest_configure 或模块级 autouse=True fixture),结果一跑就崩。这是因为 conftest 加载和 fixture 解析发生在 pytest 初始化阶段,此时 event loop 还没启动,甚至可能压根没被创建。
-
pytest_configure和pytest_sessionstart是纯同步 hook,无法 await,也不能启动 loop - 模块级或 session 级的 async fixture 在 pytest-asyncio 中默认不被支持(除非显式配置
loop_scope = "session") - 若真需 session 级异步初始化,改用同步方式预热(如提前启动线程/进程),或拆成 lazy-init:首次调用时再用
TaskGroup初始化
替代TaskGroup的轻量方案:asyncio.create_task + gather
当兼容性或控制粒度更重要时,asyncio.create_task + await asyncio.gather(...) 更稳妥,尤其适合 fixture 场景:
@pytest.fixture
async def service_client(event_loop):
task1 = asyncio.create_task(start_server())
task2 = asyncio.create_task(connect_to_redis())
await asyncio.gather(task1, task2) # 等全部完成
return Client()
这个写法绕过了 TaskGroup 的生命周期约束,也不依赖 __aenter__ 的 loop 检查。缺点是错误传播不如 TaskGroup 精确(单个 task 失败不会自动 cancel 其他),但对 fixture 初始化这类“全成功才可用”的场景影响不大。
真正容易被忽略的是:即使用了 TaskGroup,也要确保所有 create_task 调用都在同一个 async with 块内 ——跨 scope 的 task 可能被提前 cancel,且不会报错,只表现为资源未就绪。


















