在pytest中为Flask+SQLAlchemy实现嵌套事务测试的问题

看起来你遇到的核心问题是模块级会话的事务被意外关闭，导致函数级fixture尝试回滚时触发错误。我们先拆解问题原因，再给出可行的修复方案。

为什么模块级fixture能正常工作？ #

你的模块级fixture逻辑是正确的：

1. 它在连接层面开启了一个顶级事务（conn.begin()），所有模块内的会话都绑定到这个连接。
2. 通过begin_nested()创建savepoint，配合after_transaction_end监听，在测试中调用session.commit()时，会自动重启savepoint——这样测试里的commit不会真正提交到数据库，只是释放当前savepoint并新建一个，顶级事务始终保持活跃状态。
3. 最后模块结束时，回滚连接级的顶级事务，所有模块内的静态数据都会被清理，不会污染数据库。

函数级fixture出错的原因 #

你当前的函数级fixture存在几个关键问题：

1. 重复创建scoped session：绑定到模块级会话的同一个连接，但两个会话共享连接的事务状态，操作时会互相干扰，导致父事务被意外关闭。
2. 事务层级混淆：module_db_session.begin(subtransactions=True)创建的子事务，和模块级的savepoint监听逻辑冲突——当测试中触发after_transaction_end时，模块级的监听会重置事务状态，导致你要回滚的parent_transaction已经被关闭。
3. commit的影响：如果测试中调用了session.commit()，模块级的监听会重启savepoint，但函数级的事务处理没有适配这个逻辑，导致事务状态不一致。

修复后的完整代码 #

我们可以复用模块级的会话，用嵌套savepoint实现函数级的事务隔离，避免创建新会话带来的冲突：

模块级fixture（保持核心逻辑，优化注释） #

@pytest.fixture(scope='module')
def module_db_session(app, db):
    """Module-scoped session with connection-level transaction for static data."""
    with app.app_context():
        conn = db.engine.connect()
        # 开启连接级的顶级事务，所有模块内操作都在这个事务里
        top_transaction = conn.begin()
        
        options = {'bind': conn, 'binds': {}}
        sess = db.create_scoped_session(options=options)
        
        # 创建初始savepoint，处理测试中的commit操作
        sess.begin_nested()
        
        @event.listens_for(sess(), 'after_transaction_end')
        def restart_module_savepoint(sess2, trans):
            # 仅处理模块级的顶级savepoint（非嵌套的父事务）
            if trans.nested and not trans._parent.nested:
                sess2.expire_all()
                sess.begin_nested()
        
        # 替换全局db.session，确保测试中使用的是这个模块级会话
        db.session = sess
        yield sess
        
        # 清理：回滚整个连接级事务，移除会话，关闭连接
        top_transaction.rollback()
        sess.remove()
        conn.close()

函数级fixture（复用模块会话，用savepoint实现隔离） #

@pytest.fixture(scope='function')
def db_session(module_db_session):
    """Function-scoped session with nested savepoint that rolls back after test."""
    # 在模块级会话的基础上创建嵌套savepoint
    func_savepoint = module_db_session.begin_nested()
    
    @event.listens_for(module_db_session, 'after_transaction_end')
    def restart_function_savepoint(sess, trans):
        # 仅处理当前函数级的savepoint（父事务是模块级的顶级事务）
        if trans.nested and trans._parent == module_db_session.transaction:
            sess.expire_all()
            # 重启函数级savepoint，避免测试中commit后无法继续操作
            func_savepoint = sess.begin_nested()
    
    yield module_db_session
    
    # 回滚函数级的savepoint，清除测试产生的临时数据
    func_savepoint.rollback()
    # 移除当前函数的监听，避免影响后续测试
    event.remove(module_db_session, 'after_transaction_end', restart_function_savepoint)

测试用例使用示例 #

@pytest.fixture(scope='module')
def static_test_data(module_db_session):
    """生成模块级静态数据，只执行一次"""
    for _ in range(1000):  # 替换成你的耗时数据生成逻辑
        module_db_session.add(StaticModel(name="static_data"))
    module_db_session.commit()  # 实际会被模块级监听转成savepoint提交

@pytest.fixture(scope='function')
def temp_test_data(static_test_data, db_session):
    """生成函数级临时数据，每次测试后回滚"""
    temp_item = TempModel(name="temp_data")
    db_session.add(temp_item)
    db_session.commit()  # 实际会被函数级监听转成savepoint提交
    return temp_item

def test_temp_data(temp_test_data, db_session):
    # 测试临时数据逻辑
    assert db_session.query(TempModel).count() == 1
    temp_test_data.name = "updated_temp"
    db_session.commit()

# 测试结束后，temp_test_data会被函数级savepoint回滚，static_test_data会被模块级事务回滚

关键修复点说明 #

1. 复用模块级会话：不需要创建新的scoped session，避免连接和事务状态冲突。
2. 用savepoint替代子事务：函数级操作都在模块级事务的嵌套savepoint中，顶级事务始终保持活跃，不会被关闭。
3. 独立的监听逻辑：函数级的监听只处理自身的savepoint，不会干扰模块级的监听逻辑，避免事务状态被意外重置。
4. 显式移除监听：测试结束后移除函数级的监听，防止影响后续测试用例的事务处理。

关于commit的疑问 #

commit操作不会关闭所有子事务：

• 在模块级，session.commit()会提交当前的savepoint，但模块级的监听会立即新建一个savepoint，所以顶级事务始终处于活跃状态。
• 在函数级，session.commit()会提交当前的函数级savepoint，函数级的监听会新建一个savepoint，不影响模块级的事务状态。
• 只有当连接级的顶级事务被rollback或commit时，才会真正结束所有嵌套的savepoint。

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