ReactorNotRestartable when launching two equivalent unittest with twisted and trial

Question

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

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).

EDIT

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()")

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()>

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) ?

Answer 1

The reactor is not restartable. There are two obvious options for your to pursue for writing your tests.

One is to use the global reactor. trial starts and stops it - your tests never have to call reactor.run or reactor.stop (and they never should). It is accessible in the usual way, from twisted.internet import reactor.

The other is to use a new reactor instance per test. There are some test-oriented reactor instances in twisted.test.proto_helpers (that is the only part of twisted.test that is a public, supported interface by the way). MemoryReactor and StringTransport get you most of the way towards being able to test network interactions. twisted.internet.task.Clock helps you out with testing time-based events.

Answer 2

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.

This is the code:

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()")

Output:

setUp() 
test_main()

// (1s wait)

_called_by_deffered1()

// (1s wait)

_called_by_deffered2()
tearDown()