回滚Flask中的测试之间的许多事务

Question

我的测试需要很长时间才能运行,我试图在测试之间回滚事务,而不是在测试之间删除和创建表.

问题是在某些测试中我做了多次提交.

编辑:如何在测试之间回滚事务,以便测试运行得更快

这是用于测试的Base类.

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

Answer 1

这是我们用来执行此操作的代码。确保在您的设置中调用 __start_transaction，并在拆卸中调用 __close_transaction（如果您使用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()

Answer 2

你可以使用Session.begin_nested. 只要您的所有测试都正确调用commit以关闭其子交易，我认为您可以简单地执行以下操作

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

在我看来，这似乎应该更快。然而，可能在某种程度上取决于您的数据库。