[py] тестирование приложения Flask с вложенными транзакциями SQLAlchemy во вложенных областях - PullRequest
Новая
       25

[py] тестирование приложения Flask с вложенными транзакциями SQLAlchemy во вложенных областях

1 голос
/ 30 октября 2019

Я хочу иметь вложенные области для доступа к базе данных в моих тестах pytest, чтобы я мог использовать область действия module для создания объектов базы данных, которые в значительной степени статичны и занимают много времени, и область действия function длявещи, которые нужно создать и откатить для каждого теста. В полупсевдокоде что-то вроде этого: 

@pytest.fixture(scope='module')
def module_db_session():
    # create db session
    session = db.create()

    yield session

    # roll back all the changes after the module is tested
    session.rollback()

@pytest.fixture(scope='function')
def function_db_session(module_db_session):
    # Create a nested transaction that is used for this function scope
    nested_transaction = module_db_session.create()

    yield nested_transaction

    # roll back all the changes in the nested transaction after the function is tested
    nested_transaction.rollback()


# Theoretical usage
@pytest.fixture(scope='module')
def createstuff(module_db_session):
    # Creates a lot of stuff that takes a lot of time to generate
    # and is static data in the database so generating it once for module is nice
    for i in range(1000000000):
        module_db_session.create_thing()

@pytest.fixture(scope='function')
def creatething(createstuff, function_db_session):
    # Creates much less stuff that is actively tested and changes from test to test
    # so the scope should be function here

    function_db_session.create_another_thing()  # this should be rolled back after the test runs

Так что, очевидно, я пытался сделать то же самое в реальном коде, но транзакции и сеансы SQLAlchemy все еще слишком незнакомы для меня. Вот что у меня сейчас есть: 

@pytest.fixture(scope='module')
def module_db_session(app, db):
    """
    Returns function-scoped session.
    """
    with app.app_context():
        conn = db.engine.connect()
        transaction = conn.begin()

        options = {'bind': conn, 'binds': {}}
        sess = database.create_scoped_session(options=options)

        # establish a SAVEPOINT just before beginning the test
        # (http://docs.sqlalchemy.org/en/latest/orm/session_transaction.html#using-savepoint)
        sess.begin_nested()

        @event.listens_for(sess(), 'after_transaction_end')
        def restart_savepoint(sess2, trans):    
            # Detecting whether this is indeed the nested transaction of the test
            if trans.nested and not trans._parent.nested:
                # The test should have normally called session.commit(),
                # but to be safe we explicitly expire the session
                sess2.expire_all()
                sess.begin_nested()

        db.session = sess
        yield sess

        # Cleanup
        # This instruction rolls back any commit that was executed in the tests.

        transaction.rollback()

        sess.remove()

        conn.close()


@pytest.fixture(scope='function')
def db_session(app, module_db_session):
    """
    Returns function-scoped session.
    """

    with module_db_session.begin(subtransactions=True) as parent_transaction:
        conn = module_db_session.connection()
        options = {'bind': conn, 'binds': {}}
        sess = database.create_scoped_session(options=options)

        with sess.begin_nested() as nested_transaction:
            @event.listens_for(sess(), 'after_transaction_end')
            def restart_savepoint(sess2, trans):    
                if trans.nested and trans._parent.nested and not trans._parent._parent.nested:
                    # The test should have normally called session.commit(),
                    # but to be safe we explicitly expire the session
                    sess2.expire_all()
                    sess.begin_nested()

            yield sess

        parent_transaction.rollback()

Я пробовал это разными способами, но сейчас почти для каждого из них я получаю sqlalchemy.exc.ResourceClosedError: This transaction is closed для parent_transaction в функциональной сессии, но я не совсем уверен, почему. Изначально у меня была только одна вложенная транзакция в функциональном приборе, но поскольку db.session.commit() закрывает субтранзакцию, если вы в ней, то я создал два уровня вложенных транзакций и возвращал новую вложенную транзакцию каждый раз, когда .commit() закрывает одну ... Так что я не совсем понимаю, почему моя родительская транзакция закрывается в конце, когда я пытаюсь .rollback(). .commit() закрывает все субтранзакции тогда? Как работает модульная арматура?

Большое спасибо, если вы можете поделиться некоторым пониманием этого.

PS. Мой оригинальный прибор с модульной областью в значительной степени основан на этом: https://github.com/pytest-dev/pytest-flask/issues/70#issuecomment-361005780. Должен дать реквизит, где он должен.

...