sqlalchemy.exc.ResourceClosedError: This Connection is closed error

user_12884304 0
Viewed 1048

I cover the project with tests using pytest.

In each application (module), I created the tests folder, inside which placed the files with the application tests.

In each tests folder there are conftest fixtures for each application.

When I run the tests separately for each application (like pytest apps/users) everything works fine.

But when I run the tests entirely for the whole project (just pytest) for the first application the tests pass, but then it throws the sqlalchemy.exc.ResourceClosedError: This Connection is closed error for other application

Example of 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()

structure of project

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
flask-sqlalchemy pytest flask sqlalchemy python
1 Answers

You can not have two simultaneous connections to sqlite database. Also you have two connections here, one explicit in session fixture, you open and close it by your self, and second implicit in db fixture (_db.session), probably closing not happen here. So, try use connection implicit and only once, instead db and session fixtures make only 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()

here prepare_data is your data filling of new db:

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()

because session fixture here is function-scope, in each test you will have your one database. It will be more practical dont fill database fully each time, but split this prepare_data to few separate fixtures, each for one object, and use them in tests where they exactly needed.

user_2729330
Related