如何在没有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选项可以帮助这一点。它会导致试用在每个测试方法之间调用垃圾收集器。这会显着降低你的测试速度，但它应该会导致错误被记录在实际触发它的测试中，而不是任意后来的测试。