sqlalchemy.ex c .ResourceClosedError: Ошибка подключения закрыта - PullRequest
Новая
       18

sqlalchemy.ex c .ResourceClosedError: Ошибка подключения закрыта

0 голосов
/ 26 марта 2020

Я покрываю проект тестами, используя pytest.

. В каждом приложении (module) я создал tests folder, внутри которого помещались файлы с application tests.

.

В каждой папке тестов есть conftest fixtures для каждого приложения.

Когда я запускаю тесты отдельно для каждого приложения ( как pytest apps/users), все работает нормально.

Но когда я запускаю тесты полностью для всего проекта ( просто pytest) для первого приложения tests pass, но затем оно выдает sqlalchemy.ex c. ResourceClosedError: Это соединение закрывается с ошибкой для другого приложения

Пример 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

1 Ответ

1 голос
/ 01 апреля 2020

Вы не можете иметь два одновременных подключения к базе данных sqlite. Также у вас есть два соединения здесь, одно явное в фикстуре сеанса, вы открываете и закрываете его сами, а второе неявное в db фикстуре (_db.session), вероятно, закрытие здесь не происходит. Итак, попробуйте использовать неявное соединение и только один раз, вместо того, чтобы базы данных и фиксации сеанса создавали только фиксации сессии: 

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

, потому что сессия Приспособление здесь является функцией-областью, в каждом тесте у вас будет одна база данных. Будет более практичным не заполнять базу данных каждый раз полностью, а разбивать этот prepare_data на несколько отдельных приборов, каждое для одного объекта, и использовать их в тестах там, где они точно необходимы.

...