根本原因是 pytest-asyncio 与 sqlalchemy 异步引擎缓存机制的事件循环归属冲突:@lru_cache 缓存的 asyncengine 绑定首个测试的事件循环,后续测试启用新循环时触发跨循环访问,导致 runtimeerror。

根本原因不是 pytest-asyncio 本身有 bug,而是它和 SQLAlchemy 异步引擎的缓存机制发生了事件循环归属冲突。
为什么第一个测试通过、第二个就报 RuntimeError: Event loop is closed
pytest-asyncio 默认为每个测试函数创建一个新事件循环(loop_scope="function"),而你的 get_async_engine() 很可能用了 @lru_cache 装饰器缓存了 AsyncEngine 实例。这个实例在第一个测试的循环(Loop A)里初始化,内部所有 Future 和连接都绑定到了 Loop A。当第二个测试启动 Loop B 时,fixture 仍返回那个绑定到 Loop A 的 engine —— 此时任何数据库操作(比如 await conn.run_sync(...))都会触发跨循环访问。
- 错误堆栈里出现
asyncpg/protocol/protocol.pyx或_on_waiter_completed(),基本可确认是 asyncpg 在用旧循环的 Future 响应新循环的请求 - 即使你没显式调
loop.close(),Loop A 在测试结束时已被 pytest-asyncio 自动关闭,但 engine 缓存还拿着它的残留引用 - 模块级定义的
AsyncSessionFactory = async_sessionmaker(bind=get_async_engine(), ...)会让问题更隐蔽:它在 import 时就执行了一次get_async_engine(),锁定的是最早那个循环
为什么 @lru_cache 在异步测试中是危险操作
@lru_cache 是线程安全的,但不感知事件循环生命周期。它把 AsyncEngine 当成普通对象缓存,完全忽略其内部持有的 loop 引用。结果就是:缓存越久,跨循环风险越高。
- 正确做法是去掉
@lru_cache,改用 pytest fixture 管理 engine 生命周期,例如:@pytest.fixture(scope="session")创建 engine,再用@pytest.fixture(scope="function")创建 session - 如果必须复用 engine(比如为了加速测试),确保该 fixture 的
scope和所有依赖它的 async fixture 的loop_scope严格一致(都设为"session",且 pytest-asyncio 配置asyncio_default_fixture_loop_scope = "session") - 检查
src/configs/db.py里是否还有其他地方(如Base.metadata.create_all初始化)隐式触发了 engine 创建
如何验证当前循环是否被意外复用
在关键 fixture 或测试开头加一行日志:print(f"Current loop: {id(asyncio.get_running_loop())}")。对比多个测试的输出,如果 ID 不同但 engine 复用,就是典型冲突。
- 不要依赖
asyncio.get_event_loop(),它在 pytest-asyncio 下行为不稳定;始终用asyncio.get_running_loop() - 避免在 fixture 中直接调
asyncio.run()—— 它会强制关闭当前 loop,导致后续测试拿不到可用循环 - 如果你看到错误信息里夹着
_ProactorBasePipeTransport.__del__,说明有资源在 GC 阶段试图操作已关闭的 loop,大概率是 asyncpg 连接池没 clean up 干净
最易被忽略的一点:asyncpg 的连接池默认复用连接,而连接对象内部持有对原始 loop 的强引用。哪怕 engine 重建了,只要底层连接没彻底释放,旧 loop 的影子还在。
Python免费学习笔记(深入):立即使用
在学习笔记中,你将探索 Python 的核心概念和高级技巧!











