使用Pytest+Anyio创建Motor MongoDB会话Fixture,测试后数据未清除
问题原因与修复方案
核心问题:数据库操作没绑定到测试会话的事务里
你启动了会话和事务,但测试中通过mongo_client执行的操作根本没关联这个会话,所以这些操作不在事务范围内,执行abort_transaction()自然不会回滚它们,数据就留在库里了。
具体修复步骤
1. 调整Fixture,让会话能被测试用例使用
把Fixture改成返回会话实例,同时建议把作用域从session改成function,保证每个测试用例用独立事务,避免用例间干扰:
# conftest.py from .db import mongo_client import pytest from motor.motor_asyncio import AsyncIOMotorClientSession @pytest.fixture(scope='session') def anyio_backend() -> str: return 'asyncio' @pytest.fixture(scope='function', autouse=True) async def mongo_session(anyio_backend) -> AsyncGenerator[AsyncIOMotorClientSession, None]: async with await mongo_client.start_session() as session: async with session.start_transaction(): yield session # 无需手动abort,with块结束后会自动回滚未提交的事务
2. 测试用例中绑定会话到数据库操作
所有增删改查都要指定session参数,这样操作才会纳入事务:
# test_mongo.py from .db import mongo_client async def test_add_item(mongo_session) -> None: # 插入时指定会话 await mongo_client.test_db.test_collection.insert_one( {"name": "test_item"}, session=mongo_session ) # 查询也要指定会话,否则读不到事务内未提交的数据 inserted_item = await mongo_client.test_db.test_collection.find_one( {"name": "test_item"}, session=mongo_session ) assert inserted_item is not None
3. 必要的环境检查
- 确保MongoDB版本≥4.0,低于这个版本不支持事务。
- 数据库必须是副本集模式(单节点副本集也可以),standalone模式不支持事务。
原代码无效的关键原因
- 测试里的数据库操作完全独立于Fixture启动的会话,根本没进入事务流程,所以事务回滚对它们毫无作用。
内容的提问来源于stack exchange,提问作者ffempegging
相关产品推荐
相关产品推荐

