python - sqlalchemy.exc.ResourceClosedError : This Connection is closed error

Question

我使用 pytest 覆盖了该项目的测试.

在每个应用程序( module )中，我创建了 tests folder , 其中放置了带有 application tests 的文件.

在每个测试文件夹中有 conftest fixtures对于每个应用程序。

当我为每个应用程序(如 pytest apps/users)分别运行测试时，一切正常。

但是，当我为第一个应用程序的整个项目(仅 pytest)完全运行测试时，tests pass ，但随后它抛出 sqlalchemy.exc.ResourceClosedError: This Connection is closed 错误用于其他应用程序
conftest.py的示例

import os

import pytest


TESTDB = "test.db"
TESTDB_PATH = os.path.join(basedir, TESTDB)


@pytest.fixture(scope="session")
def app(request):
    """Session-wide test `Flask` application."""
    app = create_app("config.TestConfig")
    # Establish an application context before running the tests.
    ctx = app.app_context()
    ctx.push()

    def teardown():
        ctx.pop()

    request.addfinalizer(teardown)
    return app


@pytest.fixture(scope="session")
def db(app, request):
    """Session-wide test database."""
    if os.path.exists(TESTDB_PATH):
        os.unlink(TESTDB_PATH)

    def teardown():
        _db.drop_all()
        try:
            os.unlink(TESTDB_PATH)
        except FileNotFoundError:
            pass

    _db.app = app
    _db.create_all()

    permission = PermissionModel(title="can_search_articles")
    role = RoleModel(title="API User", permissions=[permission])
    tag = TagModel(name="Test tag")
    article = ArticleModel(
        title="Test article",
        legal_language="en",
        abstract="",
        state="Alaska",
        tags=[tag],
    )
    _db.session.add_all([role, permission, tag, article])
    _db.session.commit()

    user1 = UserModel(email="test@gmail.com", role_id=role.id)
    user2 = UserModel(email="test2@gmail.com")
    _db.session.add_all([user1, user2])

    # Commit the changes for the users
    _db.session.commit()

    request.addfinalizer(teardown)
    return _db


@pytest.fixture(scope="function")
def session(db, request):
    """Creates a new database session for a test."""
    connection = db.engine.connect()
    transaction = connection.begin()

    options = dict(bind=connection, binds={})
    session = db.create_scoped_session(options=options)

    db.session = session

    def teardown():
        transaction.rollback()
        connection.close()
        session.remove()

    request.addfinalizer(teardown)
    return session


@pytest.fixture(scope="module")
def client(app):
    client = app.test_client()
    ctx = app.app_context()
    ctx.push()
    yield client
    ctx.pop()

项目结构

proj/
__apps/
____articles/
______models.py, views.py, __init__.py etc
______tests/
________|__init__.py
________test_models.py
________conftest.py
____users/
______models.py, views.py, __init__.py etc
______tests/
________|__init__.py
________test_models.py
________conftest.py
______init__.py  # Here I load my models, register blueprints
__main.py  # Here I run my application

Answer 1

你不能有两个同时连接到 sqlite 数据库。此外，您在这里有两个连接，一个在 session 装置中显式，您自己打开和关闭它，第二个在 db 中隐式。 fixture ( _db.session )，可能不会在这里关闭。因此，尝试使用隐式连接且仅使用一次，而不是 db 和 session fixtures 只生成 session fixture:

@pytest.fixture
def session(app):
    """Creates a new database session for a test."""
    db.app = app
    db.create_all()

    with db.engine.connect() as connection:
        with connection.begin() as transaction:
            options = dict(bind=connection, binds={})
            session = db.create_scoped_session(options=options)

            db.session = session

            prepare_data(session)

            yield session

            transaction.rollback()
            db.drop_all()

这里 prepare_data 是你的新数据库的数据填充:

def prepare_data(session):
    permission = PermissionModel(title="can_search_articles")
    role = RoleModel(title="API User", permissions=[permission])
    tag = TagModel(name="Test tag")
    article = ArticleModel(
        title="Test article",
        legal_language="en",
        abstract="",
        state="Alaska",
        tags=[tag],
    )
    session.add_all([role, permission, tag, article])
    session.commit()

    user1 = UserModel(email="test@gmail.com", role_id=role.id)
    user2 = UserModel(email="test2@gmail.com")
    session.add_all([user1, user2])

    # Commit the changes for the users
    session.commit()

因为这里的 session fixture 是函数作用域，所以在每个测试中你都会有一个数据库。更实用的是不要每次都完全填满数据库，而是将这个 prepare_data 拆分为几个单独的固定装置，每个固定装置用于一个对象，并在他们真正需要的测试中使用它们。