我有一些创建多个延迟链的 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)

最佳答案

与这个问题相关的试验有两件重要的事情。

首先，如果在运行时记录了失败，测试方法将不会通过。 Deferreds 是垃圾收集的失败结果导致失败被记录。

其次，如果 Deferred 因失败而触发，则返回 Deferred 的测试方法将不会通过。

这意味着这些测试都不能通过:

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

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

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

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

虽然有处理这些情况的工具。毕竟，如果您不能通过测试返回一个 Deferred 的 API，而这个 Deferred 有时会引发 Failure，那么您永远无法测试您的错误代码。那将是一个非常可悲的情况。 :)

因此，处理此问题的两个工具中更有用的是 TestCase.assertFailure。这是一个帮助程序，用于希望返回将因失败而触发的 Deferred 的测试:

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

此测试将通过，因为 d 确实触发了包含 ValueError 的 Failure。如果 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 运行时，你不能依赖它。)

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

关于python - 没有 errbacks 的 Twisted Deferred 错误如何通过试验进行测试？，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/3250168/