运行pytest测试单个API接口时出现跨事件循环RuntimeError如何解决
pytest完全支持单个API接口的请求测试,你遇到的got Future attached to a different loop是asyncpg驱动与pytest异步测试事件循环不匹配的典型问题,按以下方法排查修复即可:
报错核心原因
asyncpg连接池初始化时会绑定到创建它的那个asyncio事件循环,而pytest-asyncio默认会为每个异步测试用例创建全新的事件循环。如果你的FastAPI应用在启动时(模块导入阶段、startup生命周期事件中)就初始化了全局asyncpg连接池,测试时连接池绑定的旧循环和测试用例使用的新循环不一致,就会触发该错误。
可落地修复方案
方案1:统一事件循环作用域(适配全局连接池场景)
如果你的应用确实需要复用全局连接池,直接修改pytest事件循环的作用域,让整个测试会话共用同一个事件循环即可。
在测试根目录下新建conftest.py,添加如下fixture:
import pytest import asyncio @pytest.fixture(scope="session") def event_loop(): # 会话级事件循环,整个测试过程复用同一个循环 loop = asyncio.new_event_loop() yield loop loop.close()
如果使用pytest-asyncio 0.21及以上版本,需要在项目根目录的pytest.ini中添加如下配置,允许自定义事件循环作用域:
[pytest] asyncio_mode = auto
方案2:测试环境重写数据库依赖(更推荐的测试实践)
不要复用应用启动时创建的生产环境连接池,测试时通过FastAPI的依赖覆盖机制,用当前测试循环创建测试专用连接池,避免循环绑定冲突。
在conftest.py中添加测试客户端fixture:
import pytest import asyncpg from httpx import AsyncClient from app.server import app # 导入你项目中获取数据库连接的依赖函数 from app.database import get_db # 测试数据库连接串,和生产库隔离 TEST_DB_DSN = "postgresql://用户名:密码@localhost:5432/测试库名" @pytest.fixture async def async_client(): # 用当前测试的事件循环创建测试专用连接池 test_pool = await asyncpg.create_pool(TEST_DB_DSN) # 重写项目的数据库依赖 async def override_get_db(): async with test_pool.acquire() as conn: yield conn app.dependency_overrides[get_db] = override_get_db # 初始化测试客户端 async with AsyncClient(app=app, base_url="http://test") as client: yield client # 测试结束清理资源 await test_pool.close() app.dependency_overrides.clear()
修改你的测试用例,直接复用封装好的客户端fixture即可:
import pytest @pytest.mark.asyncio async def test_index_one(async_client): response = await async_client.get("/v1/user/test/user/2") assert response.status_code == 200 @pytest.mark.asyncio async def test_index(async_client): response = await async_client.get("/v1/user/test/user/1") assert response.status_code == 200
方案3:排查版本兼容问题
部分低版本的asyncpg、pytest-asyncio存在事件循环绑定的已知bug,先升级相关依赖到稳定版本:
pip install -U pytest pytest-asyncio httpx asyncpg fastapi
避坑提醒:不要在模块顶层(导入时)直接调用
asyncpg.create_pool()初始化连接池,所有异步资源的初始化必须放在异步函数、异步fixture内部,保证创建时使用的是当前激活的事件循环。
内容的提问来源于stack exchange,提问作者maulik prajapati

