Reactor stops between tests when using twisted.trial.unittest

Question

Is there a way to keep connection online between tests?

Im trying to keep users logged when running a multiple tests set, but noticed unexpected behaviour: reactor stops after each test (so once established connection becomes unusable).

To explore this, I prepared example test-file (below).

This creates simple listening echo-server and stores it in class fields. 3 tests are run. Server expected to properly run during all tests (actualy it does only until 1st test ends).

Example:

#! /usr/bin/python
# -*- coding: utf-8 -*-

import logging

from twisted.python import log
from twisted.internet import defer, base, reactor
from twisted.trial import unittest
from twisted.internet.protocol import Protocol, Factory
from twisted.internet.endpoints import TCP4ServerEndpoint

observer = log.PythonLoggingObserver()
observer.start()
logging.basicConfig(level=logging.DEBUG)

class Echo(Protocol):
    '''Protocol from twistedmatrix examples'''
    def connectionMade(self):
        log.msg('Got incomming connection')
        self.transport.write("An apple a day keeps the doctor away\r\n")

    def connectionLost(self, reason):
        log.msg('Connection lost due to: %s' % reason)

    def dataReceived(self, data):
        self.transport.write(data)
        log.msg('Got some data: %s' % data)


class EchoFactory(Factory):
    '''Factory from twistedmatrix examples'''
    def buildProtocol(self, addr):
        return Echo()


class SampleTest(unittest.TestCase):
    '''Sample test case class derived straight from twisteds TestCase'''
    is_a_first_test = True
    endppoint = None
    def logLater(self, msgg = None):
        log.msg('called later message')

    @defer.inlineCallbacks
    def setUp(self):
        if self.__class__.is_a_first_test:
            self.__class__.endpoint = TCP4ServerEndpoint(reactor, 8007)
            self.__class__.endpoint.listen(EchoFactory())
            self.__class__.is_a_first_test = False

        log.msg('setting Up ... You may try (re)connecting now!!!')
        log.msg('We have endpoint: %s' % self.endpoint)
        yield reactor.callLater(5, self.logLater)
        log.msg('setUp done')

    def tearDown(self):
        log.msg('tearDown started')
        result = defer.Deferred()
        result.addCallback(self.logLater)
        reactor.callLater(5, result.callback, 'tearDown msg')
        log.msg('leaving tearDown')
        return result

    @defer.inlineCallbacks
    def test_00(self):
        log.msg('00 test body')
        sample_defer = defer.Deferred()
        sample_defer.addCallback(self.logLater)
        reactor.callLater(5, sample_defer.callback, 'Some sample action 00')
        log.msg('waiting reactor deferred')
        yield sample_defer
        log.msg('done with test body')

    @defer.inlineCallbacks
    def test_01(self):
        log.msg('01 test body')
        sample_defer = defer.Deferred()
        sample_defer.addCallback(self.logLater)
        reactor.callLater(5, sample_defer.callback, 'Some sample action 01')
        log.msg('waiting reactor deferred')
        yield sample_defer

    @defer.inlineCallbacks
    def test_02(self):
        log.msg('02 test body')
        sample_defer = defer.Deferred()
        sample_defer.addCallback(self.logLater)
        reactor.callLater(5, sample_defer.callback, 'Some sample action 02')
        log.msg('waiting reactor deferred')
        yield sample_defer

Running above file with:

trial test-file.py

shows "main loop terminated" at the end of each test. After that port remains listening in all tests (according to netstat -n4lt). But there is no echo when telnet connecting during 2nd and 3rd test body (only for the 1st one).

Twisted doc http://twistedmatrix.com/documents/current/core/howto/testing.html#auto3 says: "Trial runs the entire test suite (over four thousand tests) in a single process, with a single reactor" (unfortunately my twistedmatrix registration request is still not verified, so unable to place questions there).

In my real case connection to server takes too long to repeat it for each test, so I'd like to do this at least once for each class of cases.

So, is there a way to keep connection online between tests?

PS Using python 2.7.1, python-twisted 10.2.0-1, Ubuntu 11.04

Answer 1

Unit tests are meant to be isolated and independent. Connections set up in one should not be used by another.

The global reactor is an unfortunate piece of shared mutable state. Unit tests should avoid using it at all, when possible. As an aid to isolation and independence, trial resets it (as much as possible) between test methods. This feature cannot be disabled.

Since you have a connection that is expensive to set up, you should investigate creating a verified fake for that connection. A verified fake is an alternate implementation of some API which is (typically) more test-friendly (eg, creating your fake should be fast) with its own set of unit tests which demonstrate it behaves the same way as your real implementation.

Your unit tests can use this verified fake, creating a new one for each test method. This avoids the problems of violating isolation and independence and also lets your unit tests run quickly.