使用扭曲和试用版启动两个等效的单元测试时,ReactorNotRestartable [英] ReactorNotRestartable when launching two equivalent unittest with twisted and trial
问题描述
我在两个文件(test_trial1.py
和test_trial2.py
)中编写了两个测试类(TrialTest1
和TrialTest2
),它们几乎完全相同(唯一的区别是类名):
I've two test classes (TrialTest1
and TrialTest2
) written in two files (test_trial1.py
and test_trial2.py
) mostly identical (the only difference is the class name):
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()")
当我独立运行每个测试时,一切都很好.但是,当我同时启动两者时,将得到以下输出:
When I run each test idepently, everything is fine. But when I launch both I've the following output:
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
之前),也许一种解决方法是使用
It seems the reactor is not switched off correctly after the first test. Does anyone know where is the problem ? It seems tearDown()
is called to early (before _called_by_deffered1
in the second test), maybe a fix would be to use deferTearDown
(not documented method of trial unittest).
建议的解决方案之一是删除reactor.run()
和reactor.stop()
,因为反应堆不可重新启动,并且默认情况下,您只有一个反应堆实例可用于所有测试:
One of the solution proposed was to remove reactor.run()
and reactor.stop()
because a reactor is not restartable and you have only one reactor instance for all test by default:
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
方法,我的测试就会失败:
But when removing calls to such methods, my tests fail without executing _called_by_deffered
methods:
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()>
如果我只想使用测试之间共享的一个Reactor实例,那么_called_by_deffered
方法如何成为测试的一部分(在tearDown
之前执行即)?
If I want to use only one instance of reactor shared between tests, how _called_by_deffered
methods could be part of the test (i.e. executed before tearDown
) ?
推荐答案
在让·保罗(Jean-Paul)的帮助下,此问题,我已经能够使用twisted.internet.task.deferLater()
解决此问题.总结一下我一直在寻找的要点:如果一个测试方法返回一个延迟,则仅当所有延迟被触发时才调用'tearDown()'方法.
With the help of Jean-Paul, this page and this question, I've been able to fix the problem using twisted.internet.task.deferLater()
. To summarize the point i was looking for: if a test method returning a deferred, the 'tearDown()' method will be called only when all deferreds are fired.
这是代码:
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()
这篇关于使用扭曲和试用版启动两个等效的单元测试时,ReactorNotRestartable的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持IT屋!