使用扭曲和试用版启动两个等效的单元测试时,ReactorNotRestartable


问题内容

我在两个文件(TrialTest1TrialTest2)中编写了两个相同的测试类(和)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)?


问题答案:

在让·保罗(Jean-Paul),此页面这个问题的帮助下,我已经能够使用解决问题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()