SQLAlchemy scoped_session在pytest函数级Fixture中未重建致测试失败

尽管将数据库Fixture设置为function作用域，但参数化测试的三个用例共享同一个scoped_session对象，导致数据库数据清理不彻底，其中一个用例失败。三个参数化用例的预期数据状态：

• CHOOSE_GROUP：数据库表需有1行数据
• CHOOSE_PLACE：数据库表需有多行数据
• ADD_NEW_PLACE：数据库表需为空

应用与数据库工厂代码 #

init.py #

from sqlalchemy.orm import Session

def create_app():
    app = Flask(__name__)
    # some config here
    with app.app_context():
        from database.database_factory import init_db
        # db variable is set in the class with database models like below
        # db = SQLAlchemy(model_class=Base, session_options={'autoflush': True})
        init_db(db)
    # some config here
    return app

database_factory.py #

from flask_sqlalchemy import SQLAlchemy

Engine = None
Session = None

def init_db(db_object: SQLAlchemy):
    db_object.create_all()
    global Session
    global Engine
    Session = db_object.session
    Engine = db_object.engine

测试Fixture代码（conftest.py） #

@pytest.fixture(scope="function")
def app():
    app = create_app()
    yield app

@pytest.fixture(scope="function")
def client(app):
    with app.test_client() as test_client:
        return test_client

@pytest.fixture(scope="function")
def app_ctx(app):
    with app.app_context():
        set_up_db(app)
        yield
        teardown_db(app)

def set_up_db(test_app):
    from database.database_factory import Session, Engine
    test_app.Session = Session
    test_app.Engine = Engine
    test_user = User(email=USER_EMAIL, password=ph.hash(USER_PASSWORD), phone=USER_PHONE)
    test_app.Session.add(test_user)
    test_app.Session.commit()

def teardown_db(test_app):
    test_app.Session.remove()
    test_app.Engine.dispose()
    shutil.rmtree('./instance')

测试用例代码（test.py） #

@pytest.mark.usefixtures("app_ctx")
@pytest.mark.parametrize("content", [CHOOSE_GROUP, CHOOSE_PLACE, ADD_NEW_PLACE])
def test_login_view_with_correct_credentials1(self, app, client, content):
    add_temp_places(content, app.Session)
    app.Session.commit()
    response = client.post(path='/login', data={'email': USER_EMAIL, 'password': USER_PASSWORD, 'submit': True},
                           follow_redirects=True)
    assert response.status_code == 200
    assert content in response.data

def add_temp_places(content, session: scoped_session):
    try:
        if content == CHOOSE_PLACE:
            session.add_all([MOCK_PLACE_1, MOCK_PLACE_2])
        if content == CHOOSE_GROUP:
            session.add(MOCK_PLACE_1)
    except Exception:
        raise

问题根源 #

核心问题出在全局Session变量的复用：

1. database_factory.py中的Session是全局变量，每次调用create_app()时，init_db会将同一个db.session（scoped_session实例）赋值给这个全局变量。Flask-SQLAlchemy的scoped_session默认基于线程/上下文隔离，但pytest参数化测试的多个用例会在同一线程的不同上下文执行，导致Session未被正确重置。
2. app_ctx Fixture的清理逻辑仅调用Session.remove()，这只是清除当前上下文的Session实例，并未销毁全局的scoped_session对象，下一个用例会继续复用。
3. create_app()在app.app_context()中初始化数据库，导致每次创建app时重复创建数据库结构，但Session的全局引用未重置。

修正方案 #

1. 移除全局Session变量，绑定DB到App上下文 #

修改database_factory.py，去掉全局变量：

from flask_sqlalchemy import SQLAlchemy

def init_db(db_object: SQLAlchemy):
    db_object.create_all()

同时修改create_app()，将DB实例绑定到App：

def create_app():
    app = Flask(__name__)
    # 从models模块导入预先定义的db实例
    from database.models import db
    db.init_app(app)
    with app.app_context():
        from database.database_factory import init_db
        init_db(db)
        app.db = db  # 将db绑定到App，方便测试调用
    # some config here
    return app

2. 重构Fixture，确保每个用例完全隔离 #

改用内存数据库替代文件数据库，同时在Fixture中彻底清理Session和数据库：

@pytest.fixture(scope="function")
def app():
    app = create_app()
    # 使用内存数据库，避免文件IO且隔离性更强
    app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///:memory:'
    with app.app_context():
        app.db.create_all()
    yield app

@pytest.fixture(scope="function")
def app_ctx(app):
    with app.app_context():
        # 初始化测试用户
        test_user = User(email=USER_EMAIL, password=ph.hash(USER_PASSWORD), phone=USER_PHONE)
        app.db.session.add(test_user)
        app.db.session.commit()
        yield
        # 彻底清理
        app.db.session.remove()
        app.db.drop_all()

3. 修改测试用例的Session调用方式 #

@pytest.mark.usefixtures("app_ctx")
@pytest.mark.parametrize("content", [CHOOSE_GROUP, CHOOSE_PLACE, ADD_NEW_PLACE])
def test_login_view_with_correct_credentials1(self, app, client, content):
    add_temp_places(content, app.db.session)
    app.db.session.commit()
    response = client.post(path='/login', data={'email': USER_EMAIL, 'password': USER_PASSWORD, 'submit': True},
                           follow_redirects=True)
    assert response.status_code == 200
    assert content in response.data

可选：用事务回滚提升测试速度 #

如果需要更快的测试执行效率，可以用事务回滚替代每次重建数据库：

@pytest.fixture(scope="function")
def db_session(app):
    with app.app_context():
        connection = app.db.engine.connect()
        transaction = connection.begin()
        session = app.db.session(bind=connection)
        session.begin_nested()

        # 事务结束后自动重启嵌套事务，避免影响后续操作
        @app.db.event.listens_for(session, "after_transaction_end")
        def restart_savepoint(session, transaction):
            if transaction.nested and not transaction._parent.nested:
                session.begin_nested()

        yield session

        # 清理资源
        session.close()
        transaction.rollback()
        connection.close()

之后在测试用例中直接使用db_session Fixture即可。

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