如何在没有errbacks的情况下对Twisted Deferred错误进行测试?
我有一些Twisted代码可以创建多个Deferreds链。其中一些可能会失败,如果没有errback将它们放回到回调链中。我无法为此代码编写单元测试 - 失败的延迟会导致测试代码完成后测试失败。我如何为这段代码编写一个通过单元测试?预计在正常运行中可能出现故障的每个Deferred应该在链条末端出现错误,并将其放回到回调链上?如何在没有errbacks的情况下对Twisted 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)
大约有审判两个重要的事情涉及到这个问题。
首先,如果在运行时记录失败,测试方法将不会通过。使用故障结果进行垃圾收集的延迟会导致记录失败。
其次,如果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
选项可以帮助这一点。它会导致试用在每个测试方法之间调用垃圾收集器。这会显着降低你的测试速度,但它应该会导致错误被记录在实际触发它的测试中,而不是任意后来的测试。
很好的答案,但你可能还想提及'--force-gc'。 – Glyph 2010-07-16 03:06:17
好的电话,补充。 – 2010-07-16 18:03:31
这在使用失败实例调用log.err时也会发生,对吗? – Chris 2016-01-24 17:00:37