我有两个测试类(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
第一次测试后 react 堆似乎没有正确关闭。有谁知道问题出在哪里?似乎 tearDown()
被提前调用了(在第二次测试中的 _called_by_deffered1
之前),也许解决方法是使用 deferTearDown
(未记录试用单元测试的方法)。
编辑
建议的解决方案之一是删除 reactor.run()
和 reactor.stop()
,因为 react 堆不可重启并且您只有一个 react 堆实例用于默认全部测试:
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()>
如果我只想使用一个在测试之间共享的 react 器实例,_called_by_deffered
方法如何成为测试的一部分(即 在tearDown之前执行)
) ?
最佳答案
react 堆不可重启。有两个明显的选择可供您选择编写测试。
一种是使用全局 react 堆。 trial 启动和停止它——您的测试永远不必调用 reactor.run
或 reactor.stop
(它们永远不应该)。它可以通过通常的方式访问,from twisted.internet import reactor
。
另一种是每次测试使用一个新的 react 器实例。 twisted.test.proto_helpers
中有一些面向测试的 react 器实例(顺便说一下,这是 twisted.test
中唯一受支持的公共(public)接口(interface))。 MemoryReactor
和 StringTransport
使您能够测试网络交互。 twisted.internet.task.Clock
帮助您测试基于时间的事件。
关于python - ReactorNotRestartable 在使用 twisted 和 trial 启动两个等效的单元测试时,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/18910912/