ReactorNotRestartable when launching two equivalent unittest with twisted and trial

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

Best answer

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.