使用pytest-asyncio运行多个异步测试时出现事件循环错误

错误原因 #

pytest-asyncio默认每个测试函数会创建独立的事件循环，而你的数据库连接（由get_db_connection返回）绑定了第一个测试的事件循环，当第二个测试使用新循环时，调用该连接的异步操作就会触发RuntimeError: Task got Future attached to a different loop错误。

解决方案 #

方案1：让所有测试共享同一个事件循环 #

你可以通过两种方式配置共用循环：

• 给测试用例指定循环作用域
  给每个@pytest.mark.asyncio装饰器加上loop_scope参数，指定会话级或模块级共用循环：

  @pytest.mark.asyncio(loop_scope="session")
  async def test_foo():
      client = await get_db_connection()
      # ... 测试代码
  
  @pytest.mark.asyncio(loop_scope="session")
  async def test_error_foo():
      client = await get_db_connection()
      # ... 测试代码
  

• 通过fixture定义会话级循环
  在项目的conftest.py文件中添加以下fixture，强制所有测试使用同一个会话级事件循环：

  import pytest
  import asyncio
  
  @pytest.fixture(scope="session")
  def event_loop():
      loop = asyncio.new_event_loop()
      yield loop
      loop.close()
  

方案2：确保每个测试使用独立的数据库连接 #

检查get_db_connection是否有缓存逻辑（比如单例、lru_cache装饰器），如果存在缓存，移除后让每个测试都创建全新的连接：

# 移除缓存，每次调用创建新连接
import asyncpg

async def get_db_connection():
    return await asyncpg.connect(dsn="你的数据库DSN")

更规范的做法是用pytest fixture管理数据库连接，确保每个测试的连接独立且自动清理：

# 在conftest.py中定义fixture
import pytest
import asyncpg

@pytest.fixture(scope="function")
async def db_client():
    conn = await asyncpg.connect(dsn="你的数据库DSN")
    yield conn
    await conn.close()

然后在测试用例中直接注入这个fixture：

@pytest.mark.asyncio
async def test_foo(db_client):
    user = await db_client.fetch("SELECT ...")
    response = await FunctionUnderTest(db=db_client, ...args)
    assert response.id is not None

@pytest.mark.asyncio
async def test_error_foo(db_client):
    with pytest.raises(MissingRequiredError):
        await FunctionUnderTest(db=db_client, ...args)

内容的提问来源于stack exchange，提问作者Alex