Откат многих транзакций между тестами во Flask

Выполнение моих тестов занимает много времени, и я пытаюсь откатить транзакции между тестами вместо того, чтобы отбрасывать и создавать таблицы между тестами.

Проблема в том, что в некоторых тестах я делаю несколько коммитов.

РЕДАКТИРОВАТЬ: как выполнить откат транзакций между тестами, чтобы тесты выполнялись быстрее

Вот базовый класс, используемый для тестирования.

import unittest
from app import create_app
from app.core import db
from test_client import TestClient, TestResponse


class TestBase(unittest.TestCase):
    def setUp(self):
        self.app = create_app('testing')
        self.app_context = self.app.app_context()
        self.app_context.push()
        self.app.response_class = TestResponse
        self.app.test_client_class = TestClient
        db.create_all()

    def tearDown(self):
        db.session.remove()
        db.drop_all()
        db.get_engine(self.app).dispose()
        self.app_context.pop()

Вот моя попытка отката транзакций.

class TestBase(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.app = create_app('testing')
        cls.app_context = cls.app.app_context()
        cls.app_context.push()
        cls.app.response_class = TestResponse
        cls.app.test_client_class = TestClient

        db.create_all()

    @classmethod
    def tearDown(cls):
        db.session.remove()
        db.drop_all()
        db.get_engine(cls.app).dispose()

    def setUp(self):
        self.app_content = self.app.app_context()
        self.app_content.push()
        db.session.begin(subtransactions=True)

    def tearDown(self):
        db.session.rollback()
        db.session.close()

        self.app_context.pop()

person Siecje    schedule 06.10.2014    source источник
comment
Хотя многие с этим будут спорить. Вам действительно не нужно тестировать запущенные команды базы данных. Модульные тесты предназначены для бизнес-логики, и вы затем можете создать имитацию базы данных, чтобы избежать подобных проблем, и у вас не будет риска испортить свою базу данных.   -  person CodeLikeBeaker    schedule 11.10.2014
comment
Вы используете БД в памяти для тестирования? В противном случае это могло бы значительно ускорить тесты.   -  person jsnjack    schedule 14.10.2014
comment
Я использую тестовую базу данных в postgreSQL.   -  person Siecje    schedule 15.10.2014


Ответы (4)


Вы можете использовать Session.begin_nested. Пока все ваши тесты правильно вызывают commit, чтобы закрыть свои суб-транзакции, я думаю, вы можете просто сделать

session.begin_nested()
run_test(session)
session.rollback()

Что, на мой взгляд, должно быть быстрее. Однако, вероятно, в некоторой степени зависит от вашей базы данных.

person vgel    schedule 16.10.2014
comment
Я заменил db.session.begin (subtransactions = True) на db.session.begin_nested (). Я получаю sqlalchemy.exc.InvalidRequestError: транзакция этого сеанса была отменена из-за предыдущего исключения во время сброса. Чтобы начать новую транзакцию с этим сеансом, сначала выполните Session.rollback (). Исходное исключение: (IntegrityError) повторяющееся значение ключа нарушает уникальное ограничение ix_users_email ДЕТАЛИ: Ключ (электронная почта) = ([email protected]) уже существует. - person Siecje; 16.10.2014
comment
@Siecje, эта ошибка связана с другой проблемой, скорее всего, вам сначала нужно очистить свою БД, прежде чем начинать тесты. Если во время выполнения теста произойдет исключение, оно остановит работу до того, как произойдет откат. Тогда в следующий раз, когда вы запустите тест, таблица начнет уже частично заполненную данными, вызывающими эту ошибку. - person Jeff Widman; 13.01.2016

Это код, который мы используем для этого. Убедитесь, что __start_transaction вызывается в вашей настройке, а __close_transaction - в вашем teardown (с контекстом приложения, если вы используете flask-sqlalchemy). В качестве дополнительной подсказки, наследуйте этот код только в тестовых примерах, которые попадают в базу данных, и отделите код, который проверяет функцию вашей базы данных, от кода, который проверяет вашу бизнес-логику, потому что они по-прежнему будут работать НАМНОГО быстрее.

def __start_transaction(self):
    # Create a db session outside of the ORM that we can roll back
    self.connection = db.engine.connect()
    self.trans = self.connection.begin()

    # bind db.session to that connection, and start a nested transaction
    db.session = db.create_scoped_session(options={'bind': self.connection})
    db.session.begin_nested()

    # sets a listener on db.session so that whenever the transaction ends-
    # commit() or rollback() - it restarts the nested transaction
    @event.listens_for(db.session, "after_transaction_end")
    def restart_savepoint(session, transaction):
        if transaction.nested and not transaction._parent.nested:
            session.begin_nested()

    self.__after_transaction_end_listener = restart_savepoint

def __close_transaction(self):
    # Remove listener
    event.remove(db.session, "after_transaction_end", self.__after_transaction_end_listener)

    # Roll back the open transaction and return the db connection to
    # the pool
    db.session.close()

    # The app was holding the db connection even after the session was closed.
    # This caused the db to run out of connections before the tests finished.
    # Disposing of the engine from each created app handles this.
    db.get_engine(self.app).dispose()

    self.trans.rollback()
    self.connection.invalidate()
person Paul Becotte    schedule 08.04.2016

Если вы используете pytest, вы можете создать следующие приборы:

@pytest.fixture(scope='session')
def app():
    app = create_app('config.TestingConfig')
    log.info('Initializing Application context.')

    ctx = app.app_context()
    ctx.push()

    yield app
    log.info('Destroying Application context.')
    ctx.pop()

@pytest.fixture(scope='session')
def db():
    log.info('Initializating the database')

    _db.drop_all()
    _db.create_all()

    session = _db.session
    seed_data_if_not_exists(session)
    session.commit()

    yield _db

    log.info('Destroying the database')
    session.rollback()
    #_db.drop_all() #if necessary

@pytest.fixture(scope='function')
def session(app, db):
    log.info("Creating database session")

    session = db.session
    session.begin_nested()

    yield session

    log.info("Rolling back database session")
    session.rollback()
person ffleandro    schedule 20.03.2017

Хотя этот ответ технически не отвечает на ваш вопрос, вы упомянули, что причина отката тестов заключается в том, что их выполнение занимает много времени, поэтому я хотел бы предложить альтернативное решение:

Создавайте таблицы, когда вы запускаете свой набор тестов, и отбрасывайте их, когда все ваши тесты будут завершены. Затем сделайте tearDown каждого теста просто пустым таблицы вместо того, чтобы полностью их отбрасывать.

Я потратил много времени, пытаясь выяснить, как ускорить мои тесты с помощью отката, как просил исходный плакат, и нашел это очень запутанным, поскольку он включал вложенные транзакции. Однако, как только я попробовал описанный выше подход, мой набор тестов работал примерно в два раза быстрее, что было для меня достаточно хорошо.

person Atul Varma    schedule 13.01.2016
comment
Это не полностью сбрасывает все, в частности, последовательности (например, нумерация идентификаторов PK) не сбрасываются. - person Jeff Widman; 13.01.2016