如何在沒有errbacks的情況下對Twisted Deferred錯誤進行測試？

Question

我有一些Twisted代碼可以創建多個Deferreds鏈。其中一些可能會失敗，如果沒有errback將它們放回到回調鏈中。我無法爲此代碼編寫單元測試 - 失敗的延遲會導致測試代碼完成後測試失敗。我如何爲這段代碼編寫一個通過單元測試？預計在正常運行中可能出現故障的每個Deferred應該在鏈條末端出現錯誤，並將其放回到回調鏈上？

當DeferredList中Deferred失敗時會發生同樣的事情，除非我使用consumeErrors創建DeferredList。即使在使用fireOnOneErrback創建DeferredList並給出了將其放回到回調鏈上的errback時，情況也是如此。除了抑制測試失敗和錯誤日誌記錄之外，是否還有消耗錯誤的含義？是否每個Deferred可能失敗而沒有errback被放置一個DeferredList？

的示例代碼示例測試：

from twisted.trial import unittest 
from twisted.internet import defer 

def get_dl(**kwargs): 
    "Return a DeferredList with a failure and any kwargs given." 
    return defer.DeferredList(
     [defer.succeed(True), defer.fail(ValueError()), defer.succeed(True)], 
     **kwargs) 

def two_deferreds(): 
    "Create a failing Deferred, and create and return a succeeding Deferred." 
    d = defer.fail(ValueError()) 
    return defer.succeed(True) 


class DeferredChainTest(unittest.TestCase): 

    def check_success(self, result): 
     "If we're called, we're on the callback chain."   
     self.fail() 

    def check_error(self, failure): 
     """ 
     If we're called, we're on the errback chain. 
     Return to put us back on the callback chain. 
     """ 
     return True 

    def check_error_fail(self, failure): 
     """ 
     If we're called, we're on the errback chain. 
     """ 
     self.fail()   

    # This fails after all callbacks and errbacks have been run, with the 
    # ValueError from the failed defer, even though we're 
    # not on the errback chain. 
    def test_plain(self): 
     """ 
     Test that a DeferredList without arguments is on the callback chain. 
     """ 
     # check_error_fail asserts that we are on the callback chain. 
     return get_dl().addErrback(self.check_error_fail) 

    # This fails after all callbacks and errbacks have been run, with the 
    # ValueError from the failed defer, even though we're 
    # not on the errback chain. 
    def test_fire(self): 
     """ 
     Test that a DeferredList with fireOnOneErrback errbacks on failure, 
     and that an errback puts it back on the callback chain. 
     """ 
     # check_success asserts that we don't callback. 
     # check_error_fail asserts that we are on the callback chain. 
     return get_dl(fireOnOneErrback=True).addCallbacks(
      self.check_success, self.check_error).addErrback(
      self.check_error_fail) 

    # This succeeds. 
    def test_consume(self): 
     """ 
     Test that a DeferredList with consumeErrors errbacks on failure, 
     and that an errback puts it back on the callback chain. 
     """ 
     # check_error_fail asserts that we are on the callback chain. 
     return get_dl(consumeErrors=True).addErrback(self.check_error_fail) 

    # This succeeds. 
    def test_fire_consume(self): 
     """ 
     Test that a DeferredList with fireOnOneCallback and consumeErrors 
     errbacks on failure, and that an errback puts it back on the 
     callback chain. 
     """ 
     # check_success asserts that we don't callback. 
     # check_error_fail asserts that we are on the callback chain. 
     return get_dl(fireOnOneErrback=True, consumeErrors=True).addCallbacks(
      self.check_success, self.check_error).addErrback(
      self.check_error_fail) 

    # This fails after all callbacks and errbacks have been run, with the 
    # ValueError from the failed defer, even though we're 
    # not on the errback chain. 
    def test_two_deferreds(self): 
     # check_error_fail asserts that we are on the callback chain.   
     return two_deferreds().addErrback(self.check_error_fail) 

Answer 1

大約有審判兩個重要的事情涉及到這個問題。

首先，如果在運行時記錄失敗，測試方法將不會通過。使用故障結果進行垃圾收集的延遲會導致記錄失敗。

其次，如果Deferred觸發失敗，則返回Deferred的測試方法將不會通過。

這意味着既不這些測試可以通過：

def test_logit(self): 
    defer.fail(Exception("oh no")) 

def test_returnit(self): 
    return defer.fail(Exception("oh no")) 

這是重要的，因爲第一種情況下，具有失效結果收集遞延被垃圾的情況下，意味着發生了沒有之一的錯誤處理。這有點類似於如果異常到達程序的頂層時Python會報告堆棧跟蹤的方式。

同樣，第二種情況是由試驗提供的安全網。如果同步測試方法引發異常，則測試不通過。因此，如果試用測試方法返回延遲，則延遲必須具有成功結果才能通過測試。

雖然有處理這些情況的工具。畢竟，如果你不能通過一個API測試來返回一個Deferred，有時候這個Deferred有時會失敗，那麼你永遠不能測試你的錯誤代碼。這將是一個相當悲傷的情況。 :)

因此，這兩種工具處理這個更有用的是TestCase.assertFailure。這對於想返回遞延那將觸發一個故障測試一個幫手：

def test_returnit(self): 
    d = defer.fail(ValueError("6 is a bad value")) 
    return self.assertFailure(d, ValueError) 

此測試將通過，因爲d確實火了故障包裹一個ValueError。如果d已成功結果或帶有失敗包裝某種其他異常類型，則測試仍會失敗。

接下來，有TestCase.flushLoggedErrors。這是爲了當你測試一個的應該是的API來記錄一個錯誤。畢竟，有時您確實想通知管理員存在問題。

def test_logit(self): 
    defer.fail(ValueError("6 is a bad value")) 
    gc.collect() 
    self.assertEquals(self.flushLoggedErrors(ValueError), 1) 

這讓您檢查記錄的故障以確保您的記錄代碼正常工作。它還告訴審判不要擔心你沖洗的東西，所以他們不會再讓測試失敗。 （gc.collect()調用是因爲直到Deferred被垃圾收集才記錄錯誤，在CPython上，由於引用計數GC的行爲，它將立即被垃圾收集，但是，在Jython或PyPy或任何其他Python運行時）

此外，由於垃圾收集可能隨時發生，因此您可能會發現其中一個測試失敗，因爲錯誤是由Deferred創建的前測試在執行後期測試期間被垃圾收集。這幾乎總是意味着你的錯誤處理代碼在某種程度上是不完整的 - 你錯過了一個errback，或者你沒有將兩個Deferred連接在一起，或者你讓測試方法在它開始的任務實際完成之前完成 - 但錯誤報告的方式有時很難追蹤有問題的代碼。試用--force-gc選項可以幫助這一點。它會導致試用在每個測試方法之間調用垃圾收集器。這會顯着降低你的測試速度，但它應該會導致錯誤被記錄在實際觸發它的測試中，而不是任意後來的測試。