如何在SQLAlchemy异步pytest夹具中实现测试隔离？

问题描述 #

我在使用asyncio和SQLAlchemy运行pytest测试套件时，遇到测试隔离失效的问题——约200个测试用例执行失败，test_function2(db)会访问test_function1(db)之前创建的数据，引发断言错误。

当前conftest.py配置：

# conftest.py

DATABASE_URL = 'postgresql+asyncpg://user:pass@host:5432/db'

async_engine = create_async_engine(DATABASE_URL, echo=True)
AsyncSessionLocal = async_sessionmaker(
    bind=async_engine,
    class_=AsyncSession,
    autocommit=False,
    autoflush=False,
    expire_on_commit=False,
)

@pytest.fixture(scope='session')
def event_loop(request) -> asyncio.AbstractEventLoop:
    # making the event loop accessible in the session
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()


@pytest.fixture(scope='session', autouse=True)
async def create_test_database():
    # setup and teardown of database
    async with async_engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    async with async_engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)


@pytest.fixture(scope='function')
async def db() -> AsyncSession:
    # yield a database session used by test functions
    async with AsyncSessionLocal() as session:
        yield session

我尝试过两种方案，但均存在不足：

• 方案1：将create_test_database夹具作用域改为function，每次测试重建数据库，但测试耗时呈指数级增长。
• 方案2：新增truncate_tables夹具，每个测试后清空表数据，虽能正常运行，但不够优雅。

依赖版本：

aiosqlite==0.19.0
alembic==1.11.2
anyio==3.7.1
asyncpg==0.27.0
fastapi==0.97.0
pytest==7.4.0
pytest-asyncio==0.21.1
SQLAlchemy==2.0.19

最优解决方案：利用事务回滚实现测试隔离 #

核心思路是为每个测试用例开启独立数据库事务，测试结束后自动回滚事务，既保证隔离性，又避免重复建表/删表的性能损耗。

修改后的conftest.py配置 #

# conftest.py

DATABASE_URL = 'postgresql+asyncpg://user:pass@host:5432/db'

async_engine = create_async_engine(DATABASE_URL, echo=True)
AsyncSessionLocal = async_sessionmaker(
    bind=async_engine,
    class_=AsyncSession,
    autocommit=False,
    autoflush=False,
    expire_on_commit=False,
)

@pytest.fixture(scope='session')
def event_loop(request) -> asyncio.AbstractEventLoop:
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()

@pytest.fixture(scope='session', autouse=True)
async def create_test_database():
    # 仅在会话启动时创建一次表结构
    async with async_engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    async with async_engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)

@pytest.fixture(scope='function')
async def db() -> AsyncSession:
    async with AsyncSessionLocal() as session:
        try:
            # 开启测试专属事务
            await session.begin()
            yield session
        finally:
            # 测试结束强制回滚事务，清理所有测试数据
            await session.rollback()

方案优势 #

• 性能高效：仅在会话初始化时创建一次表结构，每个测试仅通过事务回滚清理数据，耗时远低于重建数据库方案，且比truncate方案更符合数据库操作规范。
• 隔离彻底：每个测试的操作都在独立事务中，回滚后不会残留任何数据，完全避免测试间的数据污染。
• 实现简洁：无需额外编写表清理逻辑，直接利用SQLAlchemy的事务机制完成隔离。

注意事项 #

• 测试用例中不要主动调用session.commit()，否则事务会被提交，无法回滚。如果测试需要模拟提交操作，可以使用嵌套事务：

  async def test_example(db: AsyncSession):
      async with db.begin_nested():
          # 执行需要提交的操作
          db.add(Model(name="test"))
          await db.commit()
      # 外层事务会回滚所有嵌套事务的操作
  

• 该方案适用于PostgreSQL等支持事务DDL的数据库；对于SQLite，SQLAlchemy会自动处理兼容问题，无需额外配置。

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