使用pytest-asyncio时遭遇MongoDB异常交互问题求助

从你提供的代码片段来看，有几个核心问题可能导致异步测试和MongoDB的交互异常，我逐一拆解并给出针对性的修复建议：

1. 未等待异步数据库操作执行完成 #

如果你的db.drop_collection是异步方法（比如基于Motor这类异步MongoDB驱动实现），当前代码没有加await关键字，会导致这些删除操作仅创建了协程对象但并未实际执行。后续插入数据时，旧的集合数据可能残留，甚至引发上下文异常。

修复方式：给异步的drop操作加上await：

await db.drop_collection('db1')
await db.drop_collection('db2')

2. DBClient实例化的复用问题 #

你先调用DBClient(loop=event_loop, test=True)但未保存这个实例，随后又通过db = DBClient()获取实例。如果DBClient是单例模式，这会导致第一次传入的event_loop可能没有被正确复用——后续获取的客户端可能绑定了错误的事件循环，直接引发异步操作无法在pytest的event_loop中运行的问题。

建议改为直接保存并复用同一个实例：

@pytest.yield_fixture(scope='session', autouse=True)
async def init_db(event_loop):
    # 实例化时绑定指定的event_loop，并复用该实例
    db_client = DBClient(loop=event_loop, test=True)
    db = db_client  # 如果DBClient直接封装了数据库操作
    # ... 后续数据库操作

3. 异步Fixture的语法与作用域匹配问题 #

@pytest.yield_fixture是较旧的pytest语法，对于异步fixture更推荐使用@pytest.fixture配合async def和yield。同时要确保你的event_loop fixture和init_db的作用域一致（都设为session），避免出现事件循环不匹配的情况。

如果是自定义event_loop fixture，示例如下：

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

4. 插入操作的验证（可选但推荐） #

为了确保测试数据正确插入，可以在操作后添加简单的断言验证，快速定位数据插入环节的问题：

check1 = await db["db1"].insert_many([item1, item2])
assert len(check1.inserted_ids) == 2  # 验证插入了2条数据

check2 = await db["db2"].insert_one(params)
assert check2.inserted_id is not None  # 验证单条数据插入成功

修复后的完整代码示例 #

import pytest
from your_module import DBClient  # 替换为你的DBClient实际导入路径

@pytest.fixture(scope='session', autouse=True)
async def init_db(event_loop):
    # 正确绑定event_loop并实例化客户端
    db_client = DBClient(loop=event_loop, test=True)
    db = db_client
    # 等待集合删除完成
    await db.drop_collection('db1')
    await db.drop_collection('db2')
    # 准备测试数据
    item1 = { "name": "something" }
    item2 = { "name": "something" }
    params = {"name": ["hue", "hon"]}
    # 执行插入并验证
    check1 = await db["db1"].insert_many([item1, item2])
    assert len(check1.inserted_ids) == 2
    check2 = await db["db2"].insert_one(params)
    assert check2.inserted_id is not None
    yield True

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

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