如何在FastAPI+SQLAlchemy的pytest测试中实现测试间回滚？

我用FastAPI开发应用，搭配pytest写测试用例时遇到了数据污染的问题：test_a调用POST接口往数据库插入新条目后，test_b查询时会包含这些不属于它的测试数据，导致结果不符合预期。单独运行每个测试都正常，但所有测试一起跑且test_a在test_b之前执行时，test_b就会失败。我不想用drop_all的方式清理数据，只想通过事务回滚实现测试间的完全隔离。

使用版本：SQLAlchemy 1.4.41、FastAPI 0.78.0、pytest 7.1.3

现有代码 #

conftest.py #

import pytest
from fastapi.testclient import TestClient
from sqlmodel import Session, SQLModel, create_engine

from application.core.config import get_database_uri
from application.core.db import get_db
from application.main import app


@pytest.fixture(scope="module", name="engine")
def fixture_engine():
    engine = create_engine(
        get_database_uri(uri="postgresql://user:secret@localhost:5432/mydb")
    )
    SQLModel.metadata.create_all(bind=engine)
    yield engine
    SQLModel.metadata.drop_all(bind=engine)


@pytest.fixture(scope="function", name="db")
def fixture_db(engine):
    connection = engine.connect()
    transaction = connection.begin()
    session = Session(bind=connection)
    yield session
    session.close()
    transaction.rollback()
    connection.close()


@pytest.fixture(scope="function", name="client")
def fixture_client(db):
    app.dependency_overrides[get_db] = lambda: db
    with TestClient(app) as client:
        yield client

测试文件中的模块级seed fixture #

@pytest.fixture(scope="module", autouse=True)
def seed(engine):
    connection = test_db_engine.connect()  # 这里存在笔误，应该用传入的engine参数
    seed_data_session = Session(bind=connection)
    seed_data(seed_data_session)
    yield
    seed_data_session.rollback()

测试用例示例 #

def test_a(client):
    # 调用POST接口创建条目逻辑
    ...

def test_b(client):
    # 查询条目并验证结果逻辑
    ...

问题原因 #

1. seed fixture的错误实现：模块级的seed fixture直接用engine创建连接插入初始化数据，但没有开启顶层事务，导致数据直接提交到数据库，成为所有测试的共享数据。而且代码里的test_db_engine是笔误，实际应该用传入的engine参数，最后回滚操作也因为没有事务上下文而无效。
2. 测试事务隔离失效：虽然每个测试的db fixture会开启事务并回滚，但如果接口中的数据库操作没有正确使用测试fixture提供的session（比如自己创建了新session），或者seed的数据不在测试事务的嵌套范围内，就会导致测试间数据污染。

修复方案 #

1. 修复seed fixture，用顶层事务保存初始化数据 #

修改seed fixture，开启模块级的顶层事务，插入初始化数据后提交到事务（不会写到数据库），所有测试共享这个事务上下文，测试结束后整体回滚清理：

@pytest.fixture(scope="module", autouse=True)
def seed(engine):
    connection = engine.connect()
    # 开启模块级顶层事务
    transaction = connection.begin()
    session = Session(bind=connection)
    # 插入初始化数据
    seed_data(session)
    session.commit()  # 仅提交到顶层事务，不写入数据库
    yield session
    # 所有测试结束后回滚顶层事务，清理初始化数据
    transaction.rollback()
    connection.close()

2. 修改db fixture，使用嵌套事务实现测试隔离 #

让每个测试的session复用seed的连接和顶层事务，通过嵌套事务（保存点）隔离测试操作，回滚时仅撤销当前测试的修改，保留初始化数据：

@pytest.fixture(scope="function", name="db")
def fixture_db(seed):
    session = seed
    # 开启嵌套事务作为当前测试的边界
    savepoint = session.begin_nested()
    yield session
    # 回滚当前测试的嵌套事务，清理测试产生的数据
    savepoint.rollback()

3. 修正client fixture的依赖覆盖 #

确保client fixture正确绑定测试用的session，并在测试后清理依赖覆盖，避免影响其他测试：

@pytest.fixture(scope="function", name="client")
def fixture_client(db):
    def override_get_db():
        yield db
    app.dependency_overrides[get_db] = override_get_db
    with TestClient(app) as client:
        yield client
    # 清理依赖覆盖
    del app.dependency_overrides[get_db]

4. 检查接口的数据库依赖 #

确保接口中通过依赖注入获取db session，而不是自行创建：

from fastapi import Depends
from sqlmodel import Session
from application.core.db import get_db

@app.post("/items/")
def create_item(item: Item, db: Session = Depends(get_db)):
    db.add(item)
    db.commit()
    db.refresh(item)
    return item

关键原理 #

• 模块级seed fixture开启顶层事务，初始化数据提交到事务但不写入数据库
• 每个测试的db fixture在顶层事务内创建嵌套事务（保存点）
• 测试结束后回滚嵌套事务，仅清理当前测试产生的数据，保留初始化数据
• 所有测试完成后，回滚顶层事务，彻底清理所有测试数据

这种方式既实现了测试间的完全隔离，又避免了重复创建/删除表的开销，提升测试效率。

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