7

我有两个测试类（TrialTest1和TrialTest2）写在两个文件（test_trial1.py和test_trial2.py）中，大部分相同（唯一的区别是类名）：

from twisted.internet import reactor
from twisted.trial import unittest


class TrialTest1(unittest.TestCase):

    def setUp(self):
        print("setUp()")

    def test_main(self):
        print("test_main")
        reactor.callLater(1, self._called_by_deffered1)
        reactor.run()

    def _called_by_deffered1(self):
        print("_called_by_deffered1")
        reactor.callLater(1, self._called_by_deffered2)

    def _called_by_deffered2(self):
        print("_called_by_deffered2")
        reactor.stop()

    def tearDown(self):
        print("tearDown()")

当我一致地运行每个测试时，一切都很好。但是当我启动两者时，我有以下输出：

setUp()
test_main
_called_by_deffered1
_called_by_deffered2
tearDown()
setUp()
test_main
tearDown()

Error
Traceback (most recent call last):
  File "/usr/lib/python2.7/site-packages/twisted/internet/defer.py", line 137, in maybeDeferred
    result = f(*args, **kw)
  File "/usr/lib/python2.7/site-packages/twisted/internet/utils.py", line 203, in runWithWarningsSuppressed
    reraise(exc_info[1], exc_info[2])
  File "/usr/lib/python2.7/site-packages/twisted/internet/utils.py", line 199, in runWithWarningsSuppressed
    result = f(*a, **kw)
  File "/home/kartoch/works/python/netkython/tests/test_twisted_trial2.py", line 13, in test_main
    reactor.run()
  File "/usr/lib/python2.7/site-packages/twisted/internet/base.py", line 1191, in run
    self.startRunning(installSignalHandlers=installSignalHandlers)
  File "/usr/lib/python2.7/site-packages/twisted/internet/base.py", line 1171, in startRunning
    ReactorBase.startRunning(self)
  File "/usr/lib/python2.7/site-packages/twisted/internet/base.py", line 683, in startRunning
    raise error.ReactorNotRestartable()
ReactorNotRestartable


Error
DirtyReactorAggregateError: Reactor was unclean.
DelayedCalls: (set twisted.internet.base.DelayedCall.debug = True to debug)
<DelayedCall 0x8d6482c [0.98535490036s] called=0 cancelled=0 TrialTest2._called_by_deffered1()>


Process finished with exit code 0

第一次测试后反应堆似乎没有正确关闭。有谁知道问题出在哪里？它似乎tearDown()被提前调用（_called_by_deffered1在第二次测试之前），也许可以使用修复deferTearDown（未记录的试用单元测试方法）。

编辑

提出的解决方案之一是删除reactor.run()，reactor.stop()因为反应器不可重新启动，并且默认情况下您只有一个反应器实例用于所有测试：

class TrialTest1(unittest.TestCase):

    def setUp(self):
        print("setUp()")

    def test_main(self):
        print("test_main")
        reactor.callLater(1, self._called_by_deffered1)

    def _called_by_deffered1(self):
        print("_called_by_deffered1")
        reactor.callLater(1, self._called_by_deffered2)

    def _called_by_deffered2(self):
        print("_called_by_deffered2")

    def tearDown(self):
        print("tearDown()")

但是，当删除对此类方法的调用时，我的测试在不执行_called_by_deffered方法的情况下失败：

setUp()
test_main
tearDown()

Error
DirtyReactorAggregateError: Reactor was unclean.
DelayedCalls: (set twisted.internet.base.DelayedCall.debug = True to debug)
<DelayedCall 0x94967ec [0.99936413765s] called=0 cancelled=0 TrialTest1._called_by_deffered1()>

setUp()
test_main
tearDown()

Error
DirtyReactorAggregateError: Reactor was unclean.
DelayedCalls: (set twisted.internet.base.DelayedCall.debug = True to debug)
<DelayedCall 0x94968cc [0.99958896637s] called=0 cancelled=0 TrialTest2._called_by_deffered1()>

如果我只想使用一个在测试之间共享的反应器实例， _called_by_deffered方法如何成为测试的一部分（即之前执行tearDown）？

4

2 回答 2

4

在 Jean-Paul、this page和this question的帮助下，我已经能够使用twisted.internet.task.deferLater(). 总结一下我一直在寻找的要点：如果测试方法返回延迟，则只有在所有延迟都被触发时才会调用“tearDown()”方法。

这是代码：

from twisted.trial import unittest
from twisted.internet import reactor, task


class TrialTest1(unittest.TestCase):

    def setUp(self):
        print("setUp()")

    def test_main(self):
        print("test_main()")
        return task.deferLater(reactor, 1, self._called_by_deffered1)

    def _called_by_deffered1(self):
        print("_called_by_deffered1()")
        return task.deferLater(reactor, 1, self._called_by_deffered2)

    def _called_by_deffered2(self):
        print("_called_by_deffered2()")

    def tearDown(self):
        print("tearDown()")

输出：

setUp() 
test_main()

// (1s wait)

_called_by_deffered1()

// (1s wait)

_called_by_deffered2()
tearDown()
于 2013-09-20T19:29:55.817 回答
4

反应堆不可重新启动。编写测试有两个明显的选择。

一是使用全局反应堆。试用开始和停止它 - 你的测试永远不必调用reactor.run或reactor.stop（他们永远不应该）。它可以通过通常的方式访问，from twisted.internet import reactor.

另一种是每次测试使用一个新的反应器实例。有一些面向测试的反应器实例twisted.test.proto_helpers（顺便说一下，其中唯一的一部分twisted.test是公共的、受支持的接口）。 MemoryReactor并StringTransport让您能够测试网络交互。 twisted.internet.task.Clock帮助您测试基于时间的事件。

于 2013-09-20T11:54:42.910 回答