SQLAlchemy数据库会话测试后未重置，引发测试数据干扰问题

问题背景 #

基于SQLAlchemy官方示例编写TakedownsDBInternalTestRunner测试基类，采用内存SQLite数据库，但执行测试时发现测试用例间数据互相干扰：test_one提交的数据会影响test_two，反之亦然，触发断言错误。

测试基类代码 #

import unittest

from sqlalchemy import create_engine
from sqlalchemy.event import listens_for
from sqlalchemy.orm import scoped_session
from sqlalchemy.orm import sessionmaker

from compliance.takedowns.data.models import Base

engine = create_engine('sqlite:///:memory:')
SessionFactory = scoped_session(sessionmaker(bind=engine))
Base.metadata.create_all(engine)  # Initial schema setup


class TakedownsDBInternalTestRunner(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.engine = engine

    def setUp(self):
        self.connection = self.engine.connect()
        self.trans = self.connection.begin()
        self.session = SessionFactory

        self.nested = self.connection.begin_nested()

        @listens_for(self.session, "after_transaction_end")
        def end_savepoint(session, transaction):
            assert self.nested is not None
            if not self.nested.is_active:
                self.nested = self.connection.begin_nested()

    def tearDown(self):
        self.session.close()
        self.trans.rollback()
        self.connection.close()

测试用例代码 #

class RequestTests(TakedownsDBInternalTestRunner):
    def test_one(self):
        request = create_base_request()
        self.session.add(request)
        self.session.commit()

        result = self.session.query(Request).all()
        self.assertEqual(len(result), 1)

    def test_two(self):
        request = create_base_request()
        self.session.add(request)
        self.session.commit()

        result = self.session.query(Request).all()
        self.assertEqual(len(result), 2)

报错信息 #

========================================================================================================= short test summary info =========================================================================================================
FAILED test_request.py::RequestTests::test_two - AssertionError: 1 != 2
FAILED test_request.py::RequestTests::test_one - AssertionError: 2 != 1

问题原因 #

1. 全局scoped_session共享：SessionFactory是全局的scoped_session实例，所有测试用例会复用同一个session，导致数据跨测试残留。
2. 未创建独立session：setUp中直接赋值self.session = SessionFactory，没有为每个测试生成新的session实例。
3. 事件监听未清理：after_transaction_end的监听函数绑定到全局session，且未在tearDown中移除，会影响后续测试的事务处理。
4. session关闭不彻底：scoped_session的close()只是放回池中，并未真正销毁实例，导致之前的数据仍能被访问。

解决方案 #

修改测试基类，确保每个测试用例拥有独立的session和事务，测试结束后彻底清理：

修改后的测试基类代码 #

import unittest

from sqlalchemy import create_engine
from sqlalchemy.event import listens_for, remove
from sqlalchemy.orm import sessionmaker

from compliance.takedowns.data.models import Base

# 全局仅初始化一次引擎和表结构
engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)


class TakedownsDBInternalTestRunner(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.engine = engine
        # 每个测试用例使用独立的session工厂，而非全局scoped_session
        cls.Session = sessionmaker(bind=cls.engine)

    def setUp(self):
        # 建立新连接，开启顶层事务
        self.connection = self.engine.connect()
        self.trans = self.connection.begin()

        # 为当前测试创建独立session
        self.session = self.Session(bind=self.connection)

        # 开启嵌套事务，处理session.commit()的情况
        self.nested = self.connection.begin_nested()

        # 绑定事务结束后的监听函数，保存引用以便后续移除
        self.end_savepoint_listener = listens_for(self.session, "after_transaction_end")(self._end_savepoint)

    def _end_savepoint(self, session, transaction):
        if not self.nested.is_active:
            self.nested = self.connection.begin_nested()

    def tearDown(self):
        # 移除事件监听
        remove(self.end_savepoint_listener)
        # 关闭当前测试的session
        self.session.close()
        # 回滚顶层事务，清除所有测试数据
        self.trans.rollback()
        # 关闭连接
        self.connection.close()

关键修改点说明 #

• 移除全局scoped_session：改用每个测试类独立的sessionmaker，确保每个测试用例生成全新的session实例。
• 独立session创建：在setUp中调用self.Session()创建当前测试专属的session，绑定到当前连接。
• 监听函数清理：保存监听函数的引用，在tearDown中调用remove()移除，避免影响后续测试。
• 事务彻底回滚：顶层事务回滚会清除所有测试期间的修改，确保测试间数据完全隔离。
• 避免session复用：每个测试的session在tearDown中关闭，不会被其他测试复用。

测试用例调整 #

测试用例无需修改，保持原有逻辑即可，因为现在每个测试的session都是独立的，提交的数据会在tearDown时被回滚，不会影响其他测试。

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