如何通过试验来测试没有 errbacks 的 Twisted Deferred 错误？

Question

我有一些 Twisted 代码，它创建了多个 Deferreds 链。其中一些可能会失败，但不会出现错误返回，从而将它们放回到回调链上。我无法为此代码编写单元测试 - 失败的 Deferred 会导致测试代码完成后测试失败。我怎样才能为这段代码编写一个通过的单元测试？是否期望每个在正常操作中可能失败的 Deferred 都应该在链的末尾有一个 errback ，将其放回到回调链上？

当 DeferredList 中存在失败的 Deferred 时，也会发生同样的情况，除非我使用 ConsumerErrors 创建 DeferredList。即使 DeferredList 是使用 fireOnOneErrback 创建的，并且被赋予一个 errback 并将其放回到回调链上，情况也是如此。除了抑制测试失败和错误日志记录之外，consumerErrors 是否还有其他影响？是否应该将每个可能在没有错误返回的情况下失败的 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 被垃圾收集且结果为 Failure 的情况，意味着发生了无人处理的错误。这有点类似于 Python 在异常到达程序顶层时报告堆栈跟踪的方式。

同样，第二个案例是审判提供的安全网。如果同步测试方法引发异常，则测试不会通过。因此，如果试验测试方法返回 Deferred，则 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 记录，并在后面的测试执行期间进行垃圾收集。这几乎总是意味着您的错误处理代码在某种程度上不完整 - 您缺少错误返回，或者您未能将两个 Deferreds 链接在一起，或者您让测试方法在它开始的任务实际完成之前完成 - 但是报告错误的方式有时会导致很难追踪到有问题的代码。 Trial 的 --force-gc 选项可以帮助解决这个问题。它会导致尝试在每个测试方法之间调用垃圾收集器。这会显着减慢你的测试速度，但它应该会导致错误被记录到实际触发它的测试中，而不是任意的后续测试中。