You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

使用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

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.06.21 03:10:13