回滾Flask中的測試之間的很多事務

Question

我的測試需要很長時間才能運行，我試圖在測試之間回滾事務，而不是在測試之間刪除和創建表。

問題是，在一些測試中，我做了多次提交。

編輯：我怎麼回退測試之間的事務，以便測試將運行得更快

下面是用於測試的基類。

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

您可以使用Session.begin_nested。只要所有的測試都正常調用commit收出自己的子事務，我想你可以簡單地做

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

，在我眼裏，好像它應該會更快。然而，可能在一定程度上取決於您的數據庫。

Answer 2

雖然這個答案並不技術上回答你的問題，你沒有提到背後滾動回測試的原因是因爲他們需要長時間運行，所以我想提供一個替代的解決方案：

當您開始運行測試套件時創建您的表格，並在所有測試完成後放下它們。然後讓每個測試的tearDown只需empty the tables而不是完全刪除它們。

我花了很長時間試圖找出如何通過回滾加速我的測試，因爲原始海報問，並發現它很混亂，因爲它涉及嵌套事務。但是，一旦我嘗試了上述方法，我的測試套件運行速度就快了一倍，這對我來說已經足夠了。

Answer 3

這是我們用來做這件事的代碼。確保在你的設置中調用__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 4

如果您使用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()