pytest运行多个asyncpg相关测试时触发cannot perform operation: another operation is in progress错误的求助

你碰到的这个InterfaceError，核心矛盾出在全局连接池和pytest-asyncio的事件循环机制不兼容，具体拆解下：

1. 你的get_pool函数创建了一个全局连接池实例，而asyncpg的连接池是和它初始化时所在的事件循环绑定的。
2. pytest-asyncio（尤其是0.17.0及以上版本）默认会给每个测试函数分配独立的事件循环。当第二个测试启动时，它会在新的事件循环里尝试复用这个全局池——但池里的连接还绑定在旧的事件循环上，新循环里发起的操作就会和旧循环的连接IO处理冲突，最终抛出"另一个操作正在进行"的错误。
3. 还有个潜在风险：如果多个测试同时首次调用get_pool，可能会触发多次池初始化，导致多个池实例共存，进一步加剧连接使用的混乱。

下面给你三种可行的解决思路，从简单到符合测试隔离原则排序：

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

修改pytest配置，强制所有测试使用同一个事件循环，这样全局连接池就能正常复用。

在项目根目录创建pytest.ini文件，添加：

[pytest]
asyncio_mode = auto

或者直接在运行测试时加参数：

pytest --asyncio-mode=auto

这个配置会让pytest-asyncio在所有测试中共享一个事件循环，全局连接池的绑定问题就直接解决了。

方案2：每个测试前重置连接池 #

如果不想共享事件循环，可以用pytest的自动fixture，在每个测试前后重置全局连接池，确保每个测试都拿到干净的池实例。

在测试文件或conftest.py中添加：

import pytest
from your_module import db_pool, get_pool

@pytest.fixture(autouse=True)
async def reset_db_pool():
    global db_pool
    # 先关闭已存在的池
    if db_pool:
        await db_pool.close()
        db_pool = None
    # 重新初始化池（也可以让测试自己触发初始化）
    await get_pool()

这个fixture会自动在每个测试执行前重置连接池，彻底隔离测试间的资源。

方案3：用fixture管理连接池（推荐） #

放弃全局连接池，改用pytest fixture来管理池的生命周期，让每个测试（或测试模块）拥有独立的池实例，这是最符合pytest测试隔离原则的做法。

示例代码：

import pytest
import asyncpg
import os
import socket
from dotenv import load_dotenv
import ujson
from your_module import src_dir

@pytest.fixture(scope="session")  # 可根据需求改成scope="module"或"function"
async def db_pool():
    dockerfiles_dir = os.path.join(src_dir, 'dockerfiles')
    env_path = os.path.join(dockerfiles_dir, 'dev.env')
    try:
        host = 'postgres-docker'
        socket.gethostbyname(host)
    except socket.error:
        host = 'localhost'
    load_dotenv(dotenv_path=env_path)
    
    async def init(con):
        await con.set_type_codec('jsonb', encoder=ujson.dumps, decoder=ujson.loads, schema='pg_catalog')
        await con.set_type_codec('json', encoder=ujson.dumps, decoder=ujson.loads, schema='pg_catalog')
    
    pool = await asyncpg.create_pool(
        database=os.getenv("POSTGRES_DBNAME"),
        user=os.getenv("POSTGRES_USER"),
        password=os.getenv("POSTGRES_PASSWORD"),
        host=host,
        init=init
    )
    yield pool
    # 测试完成后关闭池，释放资源
    await pool.close()

然后修改你的数据库操作函数，支持传入pool参数：

async def db_projects_element_ids_get(project_ids, pool):
    async with pool.acquire() as conn:
        # 执行数据库操作
        result = await conn.fetch("SELECT ... WHERE project_id = ANY($1)", project_ids)
        # 处理结果并返回

最后在测试中注入这个fixture：

@pytest.mark.asyncio
async def test_project_connection(superuser_id, project_id, db_pool):
    data = element_data_random(project_id)
    element_id = (await resolve_element_create(data=data, user_id=superuser_id, pool=db_pool))["id"]
    project_elements = (await db_projects_element_ids_get([project_id], pool=db_pool))[project_id]
    assert element_id in project_elements

• 检查你的数据库操作函数，确保没有手动持有连接引用（比如conn = await pool.acquire()后忘记释放，或者在同一个连接上并发执行多个操作）。使用pool.execute()、pool.fetch()等方法是安全的，它们会自动管理连接的获取和释放。
• 如果使用方案2或3，务必确保测试完成后关闭连接池，避免数据库连接泄漏。

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