Pytest+SQLAlchemy+FastAPI测试：创建对象无法在测试方法间回滚

问题描述 #

以下是测试代码，测试类TestMe包含test_created和test_not_created两个方法：

import pytest
from sqlalchemy import String, URL, select
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import declarative_base, Mapped, mapped_column

Base = declarative_base()


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    first_name: Mapped[str] = mapped_column(String(255))
    last_name: Mapped[str] = mapped_column(String(255))


class UsersRepository:

    def __init__(self, session: AsyncSession):
        self._session = session

    async def create(self, **create_data):
        instance = User(**create_data)
        self._session.add(instance)

        return instance
    
    async def get(self):        
        return (await self._session.scalars(select(User))).one()

    async def delete(self, instance):
        self._session.delete(instance)

    async def commit(self):
        await self._session.commit()
        
    async def refresh(self, instance):
        await self._session.refresh(instance)


@pytest.fixture(scope="session")
async def engine():
    url = URL.create(
        drivername="drivername",
        host="host",
        port="port",
        username="username",
        password="password",
        database="database",
    )
    engine_ = create_async_engine(url)
    yield engine_
    await engine_.dispose()


@pytest.fixture
async def session(engine):
    session_ = AsyncSession(engine)
    session_.begin_nested()
    yield session_
    await session_.rollback()  # 未生效
    await session_.close()


@pytest.fixture
async def users_repository(session):
    return UsersRepository(session)


class TestMe:

    async def test_created(self, users_repository):
        user = await users_repository.create(first_name="John", last_name="Snow")
        await users_repository.commit()
        await users_repository.refresh(user)
        
        # 期望测试结束后创建的用户被回滚，目前只能手动删除
        # await users_repository.delete(user)
        # await users_repository.commit()
        
    async def test_not_created(self, users_repository):
        await users_repository.get()  
        # 这里获取到了test_created中创建的对象，但期望该对象已被回滚

执行test_created时创建了User对象并提交，但session fixture中的rollback并未生效，导致test_not_created仍能获取到该对象。需解决以下问题：

• 如何实现测试对象的自动删除？
• 将engine fixture的scope设置为function是否可行？
• 问题产生的原因是什么？

问题产生的原因 #

• 测试方法中调用了await users_repository.commit()，该操作会提交顶层事务，而你在session fixture中开启的是嵌套事务(begin_nested())。一旦顶层事务被提交，嵌套事务的回滚无法撤销已持久化到数据库的变更，后续的rollback自然无效。
• 嵌套事务仅能在顶层事务范围内做局部回滚，若顶层事务被提交，所有嵌套事务的变更都会被永久写入数据库。

实现测试对象自动删除的方法 #

方案1：使用顶层事务，测试后统一回滚 #

修改session fixture，开启顶层事务，测试完成后直接回滚所有变更，同时移除测试方法内的commit()调用：

@pytest.fixture
async def session(engine):
    async with AsyncSession(engine) as session_:
        await session_.begin()  # 开启顶层事务
        yield session_
        await session_.rollback()  # 回滚所有未提交的变更

这种方式下，所有测试操作都在未提交的顶层事务中执行，测试结束后回滚即可清空所有测试数据，保证测试独立性。

方案2：嵌套事务+保存点，支持测试内commit #

如果测试中需要验证提交逻辑（必须调用commit），可以用嵌套事务的保存点机制，避免提交顶层事务：

@pytest.fixture
async def session(engine):
    async with AsyncSession(engine) as session_:
        await session_.begin()
        savepoint = await session_.begin_nested()  # 创建保存点
        yield session_
        await savepoint.rollback()  # 回滚到保存点，撤销测试内的变更
        await session_.rollback()  # 回滚顶层事务

此时测试中的commit()仅会提交嵌套事务（保存点内的变更），不会影响顶层事务，最终通过回滚保存点和顶层事务实现数据清理。

将engine fixture的scope设置为function是否可行？ #

可行，但不推荐：

• 可行原因：每个测试方法都会创建独立的engine和连接池，测试之间的数据库连接完全隔离，不会互相干扰。
• 不推荐原因：创建engine的开销较高，每个测试都新建engine会大幅增加测试总耗时，尤其是测试用例数量较多时。通常将engine的scope设为session（整个测试会话仅创建一次）是更高效的方案，通过事务隔离保证测试独立性即可。

内容的提问来源于stack exchange，提问作者Альберт Александров