FastAPI+Gino项目使用pytest测试报asyncpg操作冲突错误

这个报错的根本原因是两个独立的事件循环同时操作同一个asyncpg数据库连接，触发了asyncpg的并发操作限制：

• 同步的TestClient内部会启动独立的事件循环，用来运行FastAPI的生命周期事件（即你代码中startup钩子内的init_db逻辑）和处理接口请求
• 被@pytest.mark.anyio/@pytest.mark.asyncio标记的异步fixture运行在pytest插件创建的另一个独立事件循环中
• 两个事件循环共用全局db实例绑定的连接池，会取出同一个连接执行操作，就触发了another operation is in progress的错误

推荐改用全异步测试栈，保证所有逻辑运行在同一个事件循环中，具体修改步骤如下：

1. 新增依赖 #

安装异步测试客户端httpx：

pip install httpx==0.23.0

如果你使用@pytest.mark.anyio标记，还需要安装pytest-anyio，否则直接换成@pytest.mark.asyncio即可。

2. 调整测试代码 #

修改fixture定义 #

将同步的TestClient替换为异步的AsyncClient，调整后的fixture代码如下：

from httpx import AsyncClient

# 删掉原来的同步client fixture，替换为以下代码
@pytest.fixture
async def client():
    app = get_app()
    async with AsyncClient(app=app, base_url="http://test") as ac:
        yield ac

修改测试函数为异步 #

测试函数需要改为异步，添加异步标记，同时修正接口请求路径（原代码缺少末尾的斜杠会返回404）：

# 如果你用pytest-asyncio就换成@pytest.mark.asyncio
@pytest.mark.anyio
async def test_users(user, client):
    response = await client.get(
        "/API/v1/users/",
        headers={},
    )
    print('=====', user, response.text)
    assert response.status_code == 200

可选优化：测试数据隔离 #

可以在fixture执行完后删除测试数据，避免脏数据影响其他测试：

@pytest.fixture
@pytest.mark.asyncio
async def user(client):
    print('[-------->')
    user = await UserModel.create(email='test@gmail.com')
    users = await UserModel.query.limit(200).gino.all()
    print('.....=', user)
    print('....._', users)
    yield user
    # 测试结束后删除测试数据
    await user.delete()

修改完成后再运行pytest即可正常执行，不会再出现连接冲突报错。

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