Как правильно настроить один сеанс SQLAlchemy для каждого модульного теста?

При тестировании моего приложения Pyramid с помощью WebTest я не смог создать/использовать отдельный сеанс в своих тестах без получения предупреждений об уже существующем сеансе с ограниченной областью действия.

Вот функция main() приложения Pyramid, где настраивается база данных.

# __init__.py of Pyramid application

from pyramid_sqlalchemy import init_sqlalchemy
from sqlalchemy import create_engine


def main(global_config, **settings):
    ...
    db_url = 'some-url'
    engine = create_engine(db_url)
    init_sqlalchemy(engine)  # Warning thrown here.

Вот тестовый код.

# test.py (Functional tests)

import transaction
from unittest import TestCase
from pyramid.paster import get_appsettings
from pyramid_sqlalchemy import init_sqlalchemy, Session
from sqlalchemy import create_engine
from webtest import TestApp

from app import main
from app.models.users import User


class BaseTestCase(TestCase):
    def base_set_up(self):
        # Create app using WebTest
        settings = get_appsettings('test.ini', name='main')
        app = main({}, **settings)
        self.test_app = TestApp(app)

        # Create session for tests.
        db_url = 'same-url-as-above'
        engine = create_engine(db_url)
        init_sqlalchemy(engine)
        # Note: I've tried both using pyramid_sqlalchemy approach here and 
        # creating a "plain, old" SQLAlchemy session here using sessionmaker.

    def base_tear_down(self):
        Session.remove()


class MyTests(BaseTestCase):
    def setUp(self):
        self.base_set_up()

        with transaction.manager:
            self.user = User('[email protected]', 'John', 'Smith')
            Session.add(self.user)
            Session.flush()

            Session.expunge_all()
        ...

    def tearDown(self):
        self.base_tear_down()

    def test_1(self):
        # This is a typical workflow on my tests.
        response = self.test_app.patch_json('/users/{0}'.format(self.user.id), {'email': '[email protected]')
        self.assertEqual(response.status_code, 200)

        user = Session.query(User).filter_by(id=self.user.id).first()
        self.assertEqual(user.email, '[email protected]')
    ...
    def test_8(self):
        ...

Запуск тестов дает мне 8 пройденных, 7 предупреждений, где каждый тест, кроме первого, дает следующее предупреждение:

Из приложения Pyramid: __init__.py -> main -> init_sqlalchemy(engine): sqlalchemy.exc.SAWarning: По крайней мере один сеанс с заданной областью уже присутствует. configure() не может влиять на уже созданные сеансы.

Если это имеет какое-либо значение, я полагаю, что вижу здесь ту же проблему, за исключением того, что я использую пирамиду_sqlalchemy, а не создаю свою собственную DBSession.

https://github.com/Pylons/webtest/issues/5


person latetojoin    schedule 21.08.2018    source источник
comment
Можете ли вы дать нам код одного из ваших тестов, где появляется предупреждение   -  person bboumend    schedule 27.08.2018
comment
Обновлен пост, чтобы включить больше информации, которую я с тех пор нашел.   -  person latetojoin    schedule 27.08.2018
comment
Я не могу воспроизвести предупреждение с кодом, который вы связали, вы явно делаете вызов Session.configure() дважды подряд, но этот вызов не появляется на том, что вы нам показали, и в init_sqlalchemy такого вызова нет, я также не понимает, почему вы вызываете свой основной для настройки движка и сеанса, а затем настраиваете второй движок на тот же db и сеанс в своей базовой настройке, поскольку вы просто переопределяете первый, возможно, попробуйте удалить конфигурацию сеанса в своем основном или просто не называйте это в своем тесте.   -  person bboumend    schedule 28.08.2018
comment
@bboumend Спасибо за ваши усилия. Я считаю, что для воспроизведения этой проблемы вам нужно было бы использовать сеанс в приложении, а также тесты. Первоначально я хотел настроить отдельные сеансы для проверки фактических изменений в базе данных, поскольку состояние в рамках сеанса может отличаться от того, что фактически зафиксировано в некоторых (неожиданных) случаях. Однако я последовал вашему совету и повторно использую тот же сеанс и использую Session.expire_all() для очистки состояния по мере необходимости.   -  person latetojoin    schedule 30.08.2018


Ответы (1)


Отвечая на мой собственный вопрос: я не уверен, что это лучший подход, но тот, который сработал для меня.

Вместо того, чтобы пытаться создать отдельный сеанс в своих тестах, я использую фабрику сеансов пирамиды_sqlalchemy, которая настроена в приложении. Насколько я могу судить, вызовы Session как в тестовом коде, так и в коде приложения возвращают один и тот же зарегистрированный scoped_session.

Моей первоначальной целью создания отдельного сеанса для моих тестов было подтверждение того, что записи записываются в базу данных, а не просто обновляются в активном сеансе SQLAlchemy. Благодаря этому новому подходу мне удалось избежать этих проблем с «кэшированием», выдав Session.expire_all() в точках тестов, где я перехожу между тестовыми транзакциями и транзакциями приложения.

# test.py (Functional tests)

import transaction
from unittest import TestCase
from pyramid.paster import get_appsettings
from pyramid_sqlalchemy import Session
from webtest import TestApp

from app import main
from app.models.users import User


class BaseTestCase(TestCase):
    def base_set_up(self):
        # Create app using WebTest
        settings = get_appsettings('test.ini', name='main')
        app = main({}, **settings)
        self.test_app = TestApp(app)

        # Don't set up an additional session in the tests. Instead import
        # and use pyramid_sqlalchemy.Session, which is set up in the application.

    def base_tear_down(self):
        Session.remove()


class MyTests(BaseTestCase):
    def setUp(self):
        self.base_set_up()

        with transaction.manager:
            self.user = User('[email protected]', 'John', 'Smith')
            Session.add(self.user)
            Session.flush()

            Session.expunge_all()
            Session.expire_all()  # "Reset" your session.

    def tearDown(self):
        self.base_tear_down()
person latetojoin    schedule 30.08.2018