使用pytest与SQLAlchemy实现测试会话回滚的问题排查

核心问题是全局scoped_session实例与测试fixture的db_session上下文不共享。你在manage.py创建的全局scoped_session和conftest.py里db_session fixture生成的会话相互独立，调用session_.rollback()只会回滚fixture会话的操作，而factory通过全局session提交的操作不在这个上下文里，自然无法回滚。

1. 移除全局scoped_session，改用fixture提供 #

把manage.py里的全局scoped_session定义移到conftest.py，作为fixture提供，确保测试和factory都使用同一个会话上下文：

# conftest.py
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session, sessionmaker
from your_app.models import Base  # 替换为你的模型基类

@pytest.fixture(scope="session")
def db_engine():
    engine = create_engine("sqlite:///:memory:")  # 测试用内存数据库，可替换为你的数据库URL
    Base.metadata.create_all(engine)
    yield engine
    Base.metadata.drop_all(engine)

@pytest.fixture(scope="function")
def db_session(db_engine):
    # 创建scoped_session，绑定到测试引擎
    session_factory = sessionmaker(bind=db_engine)
    scoped_sess = scoped_session(session_factory)
    
    # 开始事务
    transaction = scoped_sess.begin()
    try:
        yield scoped_sess
    finally:
        # 回滚事务并销毁会话
        transaction.rollback()
        scoped_sess.remove()

2. 修改factory配置，动态获取测试会话 #

修改factories.py，不再硬编码全局scoped_session，而是通过factory_boy的钩子动态获取当前测试的db_session：

# factories.py
from factory import Sequence
from factory.alchemy import SQLAlchemyModelFactory
from your_app.models import Order  # 替换为你的Order模型

class OrderFactory(SQLAlchemyModelFactory):
    class Meta:
        model = Order
        sqlalchemy_session_persistence = "flush"  # 保持flush，避免自动提交

    # 动态获取当前测试会话的钩子
    @classmethod
    def _get_sqlalchemy_session(cls):
        from pytest import fixture
        try:
            # 获取当前测试的db_session fixture
            return fixture("db_session")()
        except Exception:
            # 非测试环境回退到原全局session（如果需要）
            from your_app.manage import scoped_sess
            return scoped_sess

    # 定义工厂字段示例
    order_number = Sequence(lambda n: f"ORD-{n:04d}")

3. 调整测试用例，确保使用fixture会话 #

测试用例中直接注入db_session，所有数据库操作（包括factory创建的对象）都会在这个会话上下文里：

# test_orders.py
from your_app.models import Order
from your_app.factories import OrderFactory

def test_create_order(db_session):
    # 使用factory创建对象，自动关联db_session
    order = OrderFactory()
    
    # 验证对象存在
    assert db_session.query(Order).filter_by(id=order.id).first() is not None
    
    # 模拟业务提交操作
    db_session.commit()
    
    # 执行回滚（此时会生效，所有操作都在db_session的事务中）
    db_session.rollback()
    
    # 验证回滚结果
    assert db_session.query(Order).filter_by(id=order.id).first() is None

关键说明 #

• scoped_session的作用是在当前线程/上下文提供唯一会话，但全局实例会脱离测试的fixture生命周期，导致事务隔离失效。
• 把scoped_session作为函数级fixture的一部分，确保每个测试用例都有独立的会话，测试后自动回滚并销毁，避免测试间数据污染。
• factory通过_get_sqlalchemy_session钩子获取测试会话，保证factory操作和测试用的是同一个会话，回滚时能覆盖所有操作。

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