Chapter 11. Testing

Because Twisted programs are event-driven and use Deferreds to wait for and handle events, we can’t easily use standard testing frameworks like Python’s unittest to write tests for them.

To handle this, Twisted comes with an extension of Python’s unittest framework for testing event-driven Twisted programs, and a command-line utility for running them. These components comprise Trial, Twisted’s testing framework.

Writing and Running Twisted Unit Tests with Trial

Tests that don’t exercise event-driven logic import twisted.trial.unittest instead of unittest but otherwise look identical to traditional Python unittest tests.

Example 11-1 shows a single test case class called MyFirstTestCase, containing a single test called test_something, which makes an assertion using the Twisted version of unittest’s TestCase.assertTrue. Most unittest assertions have Twisted versions, and Trial has additional assertions for exercising Failures.

Example 11-1. test_foo.py
from twisted.trial import unittest

class MyFirstTestCase(unittest.TestCase):
    def test_something(self):
        self.assertTrue(True)

We can use the trial command-line utility that ships with Twisted to run the test file:

$ trial test_foo.py
test_foo
  MyFirstTestCase
    test_something ...                                                     [OK]

-------------------------------------------------------------------------------
Ran 1 tests in 0.002s

PASSED (successes=1)

We can run individual test classes by specifying the class name, as in:

trial test_foo.MyFirstTestCase

and run individual tests by specifying the path to the test, as in:

trial test_foo.MyFirstTestCase.test_something

Testing Protocols

Let’s say we wanted to write a unit test suite for our echo protocol logic from Chapter 2 , reproduced Example 11-2 in for convenience.

Example 11-2. echo.py
from twisted.internet import protocol, reactor

class Echo(protocol.Protocol):
    def dataReceived(self, data):
        self.transport.write(data)

class EchoFactory(protocol.Factory):
    def buildProtocol(self, addr):
        return Echo()

These are unit tests; they shouldn’t rely on making network connections. But how do we fake making a client connection?

Twisted provides helper modules in twisted.test for unit-testing clients and servers. Chief amongst them is proto_helpers, which has a StringTransport class for mocking transports. When a protocol uses an instance of StringTransport, instead of pushing bytes out through a network connection, they are written to a string which can easily be inspected.

Example 11-3 has a test case for the Echo protocol. It creates an instance of EchoFactory, uses that factory to build an instance of the Echo protocol, and sets the protocol’s transport to an instance of proto_helpers.StringTransport. The protocol’s makeConnection method is called to simulate a client connection, and dataReceived is called to simulate receiving client data. At that point, the transport should contain the echoed version of the fake client data, so we make an assertion on transport.value().

Example 11-3.  test_echo.py 
from twisted.test import proto_helpers
from twisted.trial import unittest

from echo import EchoFactory

class EchoServerTestCase(unittest.TestCase):
    def test_echo(self):
        factory = EchoFactory()
        self.proto = factory.buildProtocol(("localhost", 0))
        self.transport = proto_helpers.StringTransport()

        self.proto.makeConnection(self.transport)
        self.proto.dataReceived("test\r\n")
        self.assertEqual(self.transport.value(), "test\r\n")

This idiom of:

  1. Building a protocol instance

  2. Giving it a mock transport

  3. Faking client communication

  4. Inspecting the mocked transport data

is very common when testing server functionality.

A handy feature built into trial is the generation of coverage information. If we pass --coverage to trial, it will generate coverage data for every Python module exercised during the test run and (by default) store it in _trial_temp/. Re-running the echo tests with trial --coverage test_echo.py and inspecting the resulting _trial_temp/coverage/echo.cover, we can see that we have full coverage of the echo module with this test:

$ cat _trial_temp/coverage/echo.cover
    1: from twisted.internet import protocol, reactor

    2: class Echo(protocol.Protocol):
    1:     def dataReceived(self, data):
    1:         self.transport.write(data)

    2: class EchoFactory(protocol.Factory):
    1:     def buildProtocol(self, addr):
    1:         return Echo()

As another example of mocking transports using proto_helpers.StringTransport, how about some unit tests for the chat protocol from Chapter 2 (reproduced in Example 11-4).

Example 11-4. chatserver.py
from twisted.internet.protocol import Factory
from twisted.protocols.basic import LineReceiver

class ChatProtocol(LineReceiver):
    def __init__(self, factory):
        self.factory = factory
        self.name = None
        self.state = "REGISTER"

    def connectionMade(self):
        self.sendLine("What's your name?")

    def connectionLost(self, reason):
        if self.name in self.factory.users:
            del self.factory.users[self.name]
            self.broadcastMessage("%s has left the channel." % (self.name,))

    def lineReceived(self, line):
        if self.state == "REGISTER":
            self.handle_REGISTER(line)
        else:
            self.handle_CHAT(line)

    def handle_REGISTER(self, name):
        if name in self.factory.users:
            self.sendLine("Name taken, please choose another.")
            return
        self.sendLine("Welcome, %s!" % (name,))
        self.broadcastMessage("%s has joined the channel." % (name,))
        self.name = name
        self.factory.users[name] = self
        self.state = "CHAT"

    def handle_CHAT(self, message):
        message = "<%s> %s" % (self.name, message)
        self.broadcastMessage(message)

    def broadcastMessage(self, message):
        for name, protocol in self.factory.users.iteritems():
            if protocol != self:
                protocol.sendLine(message)

class ChatFactory(Factory):
    def __init__(self):
        self.users = {}

    def buildProtocol(self, addr):
        return ChatProtocol(self)

As with the Echo protocol, we first set up an instance of the ChatFactory, build a protocol, and mock the transport. Since this is a more complicated protocol that will need several tests, we can stick the setup work needed by every test in a setUp method, which unittest will run before each test (there is a corresponding tearDown method to clean up after each test).

After that, we can test each part of the state machine in its own unit test by calling lineReceived with the appropriate state-changing data and asserting on the contents of the mocked transport. Example 11-5 shows the start of a chat server test suite.

Example 11-5. Testing chatserver
from twisted.test import proto_helpers
from twisted.trial import unittest

from chatserver import ChatFactory

class ChatServerTestCase(unittest.TestCase):
    def setUp(self):
        self.factory = ChatFactory()
        self.proto = self.factory.buildProtocol(("localhost", 0))
        self.transport = proto_helpers.StringTransport()
        self.proto.makeConnection(self.transport)

    def test_connect(self):
        self.assertEqual(self.transport.value(),
                         "What's your name?\r\n")

    def test_register(self):
        self.assertEqual(self.proto.state, "REGISTER")
        self.proto.lineReceived("jesstess")
        self.assertIn("jesstess", self.proto.factory.users)
        self.assertEqual(self.proto.state, "CHAT")

    def test_chat(self):
        self.proto.lineReceived("jesstess")

        proto2 = self.factory.buildProtocol(("localhost", 0))
        transport2 = proto_helpers.StringTransport()
        proto2.makeConnection(transport2)

        self.transport.clear()
        proto2.lineReceived("adamf")

        self.assertEqual(self.transport.value(),
                         "adamf has joined the channel.\r\n")

To exercise the new user notification logic, we build a second fake client connection in test_chat.

trial --coverage test_foo.py shows a couple of untested code paths:

    1:     def connectionLost(self, reason):
>>>>>>         if self.name in self.factory.users:
>>>>>>             del self.factory.users[self.name]
>>>>>>             self.broadcastMessage("%s has left the channel." % 
                                  (self.name,))

    1:     def lineReceived(self, line):
    3:         if self.state == "REGISTER":
    3:             self.handle_REGISTER(line)
               else:
>>>>>>             self.handle_CHAT(line)

    1:     def handle_REGISTER(self, name):
    3:         if name in self.factory.users:
>>>>>>             self.sendLine("Name taken, please choose another.")
>>>>>>             return
    3:         self.sendLine("Welcome, %s!" % (name,))
    3:         self.broadcastMessage("%s has joined the channel." % (name,))
    3:         self.name = name
    3:         self.factory.users[name] = self
    3:         self.state = "CHAT"

    1:     def handle_CHAT(self, message):
>>>>>>         message = "<%s> %s" % (self.name, message)
>>>>>>         self.broadcastMessage(message)

To have complete test coverage, we’d need to exercise users leaving, nickname collision, and sending a chat message. What would those tests look like?

Tests and the Reactor

Eventually, you will find yourself needing to test something that involves the reactor: typically functions that return Deferreds or use methods like reactor.callLater that register time-based event handlers.

trial runs your test suite in a single thread, with a single reactor. This means that if a test ever leaves an event source (like a timer, socket, or misplaced Deferred) inside the reactor, it can affect future tests. At best, this causes them to fail. At worst, it causes tests to fail apparently randomly and sporadically, leaving you with a nightmare to debug.

This fact forces a basic rule when writing tests:

Leave the reactor as you found it.

This means:

Operations to clean up the reactor often live in the unittest.tearDown test method.

Example 11-6 is a concrete demonstration of what happens when a Deferred is left unfired in the reactor.

Example 11-6. test_deferred.py
from twisted.internet.defer import Deferred
from twisted.internet import reactor
from twisted.trial import unittest

class DeferredTestCase(unittest.TestCase):
    def slowFunction(self):
        d = Deferred()
        reactor.callLater(1, d.callback, ("foo"))
        return d

    def test_slowFunction(self):
        def cb(result):
            self.assertEqual(result, "foo")

        d = self.slowFunction()
        d.addCallback(cb)

slowFunction is a stand-in for any function that returns a Deferred. test_slowFunction is an attempt to test that when slowFunction’s callback chain is fired, it is with the result “foo”.

Try running this test suite. You’ll get something like:

$ trial test_deferred.DeferredTestCase
test_foo
  DeferredTestCase
    test_slowFunction ...                                               [ERROR]

===============================================================================
[ERROR]
Traceback (most recent call last):
Failure: twisted.trial.util.DirtyReactorAggregateError: Reactor was unclean.
DelayedCalls: (set twisted.internet.base.DelayedCall.debug = True to debug)
<DelayedCall 0x1010e1560 [0.9989798069s] called=0 cancelled=0 Deferred
 .callback(('foo',))>

test_slowFunction broke the rule: it invoked a function that returned a Deferred without returning the Deferred, causing the test to fail with a DirtyReactorAggregateError: Reactor was unclean.

To fix this test so it doesn’t leave stray event sources in the reactor, return d.

DBCredentialsChecker.requestAvatarId from Example 9-2 is a method that returns a Deferred. Example 11-7 reproduces the full DBCredentialsChecker class for context. What would a test suite for requestAvatarId look like?

Example 11-7. db_checker.py
class DBCredentialsChecker(object):
    implements(ICredentialsChecker)

    credentialInterfaces = (IUsernameHashedPassword,)

    def __init__(self, runQuery, query):
        self.runQuery = runQuery
        self.query = query

    def requestAvatarId(self, credentials):
        for interface in self.credentialInterfaces:
            if interface.providedBy(credentials):
                break
            else:
                raise error.UnhandledCredentials()

        dbDeferred = self.runQuery(self.query, (credentials.username,))
        deferred = Deferred()
        dbDeferred.addCallbacks(self._cbAuthenticate, self._ebAuthenticate,
                                callbackArgs=(credentials, deferred),
                                errbackArgs=(credentials, deferred))
        return deferred

    def _cbAuthenticate(self, result, credentials, deferred):
        if not result:
            deferred.errback(error.UnauthorizedLogin('User not in database'))
        else:
            username, password = result[0]
            if credentials.checkPassword(password):
                deferred.callback(credentials.username)
            else:
                deferred.errback(error.UnauthorizedLogin('Password mismatch'))

    def _ebAuthenticate(self, failure, credentials, deferred):
        deferred.errback(error.LoginFailed(failure))

Some good candidates for unit tests are:

In lieu of setting up a test database as part of this test suite, we can mock the runQuery and query attributes to return fixed results.

Example 11-8 shows one possible implementation of the success test case. It instantiates a DBCredentialsChecker with a fakeRunqueryMatchingPassword that returns hard-coded correct credentials. A callback is attached to the Deferred returned by requestAvatarId to assert that the username in the supplied credentials is returned on a password match, and the Deferred is returned for Trial to ensure that it has time to fire.

Example 11-8. Testing DBCredentialsChecker
from twisted.trial import unittest
from twisted.cred import credentials
from twisted.cred.error import UnauthorizedLogin
from twisted.internet import reactor
from twisted.internet.defer import Deferred

from  db_checker import DBCredentialsChecker 

class DBCredentialsCheckerTestCase(unittest.TestCase):

    def test_requestAvatarIdGoodCredentials(self):
        """
        Calling requestAvatarId with correct credentials returns the
        username.
        """
        def fakeRunqueryMatchingPassword(query, username):
            d = Deferred()
            reactor.callLater(0, d.callback, (("user", "pass"),))
            return d

        creds = credentials.UsernameHashedPassword("user", "pass")
        checker = DBCredentialsChecker(fakeRunqueryMatchingPassword,
                                       "fake query")
        d = checker.requestAvatarId(creds)

        def checkRequestAvatarCb(result):
            self.assertEqual(result, "user")
        d.addCallback(checkRequestAvatarCb)
        return d

Example 11-9 shows the two error test cases, which are structured quite similarly. They use a Twisted extension to Python’s unittest assertions: assertFailure, which asserts that a Deferred fires with a Failure wrapping a particular type of Exception.

Example 11-9. Testing errors in DBCredentialsChecker
def test_requestAvatarIdBadCredentials(self):
        """
        Calling requestAvatarId with invalid credentials raises an
        UnauthorizedLogin error.
        """
        def fakeRunqueryBadPassword(query, username):
            d = Deferred()
            reactor.callLater(0, d.callback, (("user", "badpass"),)) 
            return d

        creds = credentials.UsernameHashedPassword("user", "pass")
        checker = DBCredentialsChecker(fakeRunqueryBadPassword, "fake query")
        d = checker.requestAvatarId(creds)

        def checkError(result):
            self.assertEqual(result.message, "Password mismatch")
        return self.assertFailure(d, UnauthorizedLogin).addCallback(checkError)

    def test_requestAvatarIdNoUser(self):
        """
        Calling requestAvatarId with credentials for an unknown user
        raises an UnauthorizedLogin error.
        """
        def fakeRunqueryMissingUser(query, username):
            d = Deferred()
            reactor.callLater(0, d.callback, ())
            return d

        creds = credentials.UsernameHashedPassword("user", "pass")
        checker = DBCredentialsChecker(fakeRunqueryMissingUser, "fake query")
        d = checker.requestAvatarId(creds)

        def checkError(result):
            self.assertEqual(result.message, "User not in database")
        return self.assertFailure(d, UnauthorizedLogin).addCallback(checkError)

When you need to test code scheduled with reactor.callLater, for example protocol timeouts, you need to fake the passage of time. Twisted makes this easy with the twisted.internet.task.Clock class. Clock has its own callLater method, which replaces reactor.callLater in tests and can be advanced manually.

Because Clock.callLater replaces reactor.callLater, and we don’t want to affect the global reactor while running tests, we need to parameterize the reactor (i.e., make the reactor an argument to a class’s __init__ method) so it can easily be replaced for testing.

Example 11-11 shows a test case for EchoProcessProtocol from Example 10-4. That class has been reproduced in Example 11-10 for convenience, with some changes, as discussed after the example code. EchoProcessProtocol terminates itself after 10 seconds using reactor.callLater, and we can use a Clock to exercise this behavior.

Example 11-10. pp.py
from twisted.internet import protocol, reactor

class EchoProcessProtocol(protocol.ProcessProtocol):
    def __init__(self, reactor):
        self.reactor = reactor

    def connectionMade(self):
        print "connectionMade called"
        self.reactor.callLater(10, self.terminateProcess)

    def terminateProcess(self):
        self.transport.signalProcess('TERM')

    def outReceived(self, data):
        print "outReceived called with %d bytes of data:\n%s" % (
            len(data), data)

    def errReceived(self, data):
        print "errReceived called with %d bytes of data:\n%s" % (
            len(data), data)

    def inConnectionLost(self):
        print "inConnectionLost called, stdin closed."

    def outConnectionLost(self):
        print "outConnectionLost called, stdout closed."

    def errConnectionLost(self):
        print "errConnectionLost called, stderr closed."

    def processExited(self, reason):
        print "processExited called with status %d" % (
            reason.value.exitCode,)

    def processEnded(self, reason):
        print "processEnded called with status %d" % (
            reason.value.exitCode,)
        print "All FDs are now closed, and the process has been reaped."
        self.reactor.stop()
Example 11-11. Testing EchoProcessProtocol
from twisted.test import proto_helpers
from twisted.trial import unittest
from twisted.internet import reactor, task

from pp import EchoProcessProtocol

class EchoProcessProtocolTestCase(unittest.TestCase):
    def test_terminate(self):
        """
        EchoProcessProtocol terminates itself after 10 seconds.
        """
        self.terminated = False

        def fakeTerminateProcess():
            self.terminated = True

        clock = task.Clock()
        pp = EchoProcessProtocol(clock)
        pp.terminateProcess = fakeTerminateProcess
        transport = proto_helpers.StringTransport()
        pp.makeConnection(transport)

        self.assertFalse(self.terminated)
        clock.advance(10)
        self.assertTrue(self.terminated)

Before writing this test case, we must parameterize the reactor used by EchoProcessProtocol by adding:

def __init__(self, reactor):
    self.reactor = reactor

Then, in the test case, an instance of EchoProcessProtocol can be instantiated with an instance of task.Clock. A transport is set up, and assertions are made about the state of the protocol before and after a call to clock.advance, which simulates the passage of 10 seconds.

Parameterizing the reactor and using a Clock to simulate the passage of time is a common Twisted Trial idiom.

More Practice and Next Steps

This chapter introduced Twisted’s Trial framework for unit-testing your Twisted applications.

The Twisted Core documentation includes a detailed introduction to test-driven development in Twisted and an overview of trial. trial is, of course, itself written in Twisted, and test result processing can be customized using Twisted’s plugin system. The trial code and tests live in twisted/trial/.

Twisted has a strict test-driven development policy: no code changes get merged without accompanying tests. Consequently, Twisted has an extensive test suite that is a great resource for examples of how to unit-test different scenarios. Tests live in the top-level test/ directory as well as test/ directories for each subproject.

For example, to see how Twisted Web’s Agent interface is tested, including mocking the transport, testing timeouts, and testing errors, have a look at twisted/web/test/test_agent.py. To see how to test a protocol like twisted.words.protocols.irc, check out twisted/words/tests/test_irc.py.

You can read about Twisted’s test-driven development policy in detail on the Twisted website.

Twisted publishes its own coverage information as part of its continuous integration. Help improve Twisted by writing test cases!