没有 errbacks 的 Twisted Deferred 错误如何通过试用进行测试？

Question

我有一些 Twisted 代码可以创建多个延迟链。其中一些可能会失败，而没有将它们放回回调链上的 errback。我无法为此代码编写单元测试 - 失败的 Deferred 导致测试代码完成后测试失败。如何为此代码编写通过单元测试？是否期望每个在正常操作中可能失败的 Deferred 都应该在链的末尾有一个 errback 将其放回回调链上？

当 DeferredList 中的 Deferred 失败时也会发生同样的事情，除非我使用 consumeErrors 创建 DeferredList。即使 DeferredList 是使用 fireOnOneErrback 创建的，并且被赋予了一个 errback 并将其放回回调链上，情况也是如此。除了抑制测试失败和错误记录之外，consumeErrors 是否有任何影响？是否应该将每个可能在没有 errback 的情况下失败的 Deferred 都放入 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"))

这很重要，因为第一种情况，即 Deferred 被垃圾收集并带有失败结果的情况，意味着发生了没有人处理的错误。这有点类似于 Python 在异常到达程序顶层时报告堆栈跟踪的方式。

同样，第二种情况是审判提供的安全网。如果同步测试方法引发异常，则测试不会通过。因此，如果试用测试方法返回 Deferred，则 Deferred 必须有成功的结果才能通过测试。

虽然有处理这些情况的工具。毕竟，如果您无法对返回有时会以失败触发的延迟的 API 进行通过测试，那么您将永远无法测试您的错误代码。那将是一个非常可悲的情况。 :)

所以，这两个工具中更有用的是TestCase.assertFailure。对于想要返回一个会因失败而触发的 Deferred 的测试，这是一个助手：

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)

这使您可以检查记录的故障，以确保您的日志记录代码正常工作。它还告诉 Trial 不要担心你冲洗的东西，所以它们不会再导致测试失败。 （gc.collect() 调用在那里是因为在 Deferred 被垃圾收集之前不会记录错误。在 CPython 上，由于引用计数 GC 行为，它将立即被垃圾收集。但是，在 Jython 或 PyPy 或任何其他没有引用计数的 Python 运行时，你不能依赖它。）

此外，由于垃圾收集几乎可以在任何时候发生，您有时可能会发现您的一个测试失败，因为由 早期 测试创建的 Deferred 记录了错误，该测试在期间被垃圾收集后面的测试的执行。这几乎总是意味着你的错误处理代码在某种程度上是不完整的——你错过了一个 errback，或者你没有在某个地方将两个 Deferred 链接在一起，或者你让你的测试方法在它开始的任务实际完成之前完成——但是报告错误的方式有时会导致难以追踪有问题的代码。试用版的--force-gc 选项可以帮助解决这个问题。它使试验在每个测试方法之间调用垃圾收集器。这会显着减慢您的测试，但它应该会导致错误记录在实际触发它的测试中，而不是以后的任意测试。