Test-driven development with Twisted

  1. Introductory example of Python unit testing
  2. Creating an API and writing tests
  3. Making the tests pass
  4. Twisted specific testing
  5. Testing a protocol
  6. More good pratices
  7. Resolve a bug
  8. Code coverage
  9. Conclusion

Writing good code is hard, or at least it can be. A major challenge is to ensure that your code remains correct as you add new functionality.

Unit testing is a modern, light-weight testing methodology in widespread use in many programming languages. Development that relies on unit tests is often referred to as Test-Driven Development (TDD). Most Twisted code is tested using TDD.

To gain a solid understanding of unit testing in Python, you should read the unittest -- Unit testing framework chapter of the Python Library Reference. There is also a ton of information available online and in books.

Introductory example of Python unit testing

This document is principally a guide to Trial, Twisted's unit testing framework. Trial is based on Python's unit testing framework. While we do not aim to give a comprehensive guide to general Python unit testing, it will be helpful to consider a simple non-networked example before expanding to cover a networking code that requires the special capabilities of Trial. If you are already familiar with unit test in Python, jump straight to the section specific to testing Twisted code.

Note: In what follows we will make a series of refinements to some simple classes. In order to keep the examples and source code links complete and to allow you to run Trial on the intermediate results at every stage, I add _N (where the N are successive integers) to file names to keep them separate. This is a minor visual distraction that should be ignored.

Creating an API and writing tests

We'll create a library for arithmetic calculation. First, create a project structure with a directory called calculus containing an empty __init__.py file.

Then put the following simple class definition API into calculus/base_1.py:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

# -*- test-case-name: calculus.test.test_base_1 -*-



class Calculation(object):
    def add(self, a, b):
        pass

    def subtract(self, a, b):
        pass

    def multiply(self, a, b):
        pass

    def divide(self, a, b):
        pass
Source listing - listings/trial/calculus/base_1.py

(Ignore the test-case-name comment for now. You'll see why that's useful below.)

We've written the interface, but not the code. Now we'll write a set of tests. At this point of development, we'll be expecting all tests to fail. Don't worry, that's part of the point. Once we have a test framework functioning, and we have some decent tests written (and failing!), we'll go and do the actual development of our calculation API. This is the preferred way to work for many people using TDD - write tests first, make sure they fail, then do development. Others are not so strict and write tests after doing the development.

Create a test directory beneath calculus, with an empty __init__.py file. In a calculus/test/test_base_1.py, put the following:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

from calculus.base_1 import Calculation
from twisted.trial import unittest

class CalculationTestCase(unittest.TestCase):
    def test_add(self):
        calc = Calculation()
        result = calc.add(3, 8)
        self.assertEqual(result, 11)

    def test_subtract(self):
        calc = Calculation()
        result = calc.subtract(7, 3)
        self.assertEqual(result, 4)

    def test_multiply(self):
        calc = Calculation()
        result = calc.multiply(12, 5)
        self.assertEqual(result, 60)

    def test_divide(self):
        calc = Calculation()
        result = calc.divide(12, 5)
        self.assertEqual(result, 2)
Source listing - listings/trial/calculus/test/test_base_1.py

You should now have the following 4 files: 

    calculus/__init__.py
    calculus/base_1.py
    calculus/test/__init__.py
    calculus/test/test_base_1.py

To run the tests, there are two things you must get set up. Make sure you get these both done - nothing below will work unless you do.

First, make sure that the directory that contains your calculus directory is in your Python load path. If you're using the Bash shell on some form of unix (e.g., Linux, Mac OS X), run PYTHONPATH="$PYTHONPATH:`pwd`/.." at the command line in the calculus directory. Once you have your Python path set up correctly, you should be able to run Python from the command line and import calculus without seeing an import error.

Second, make sure you can run the trial command. That is, make sure the directory containing the trial program on you system is in your shell's PATH. The easiest way to check if you have this is to try running trial --help at the command line. If you see a list of invocation options, you're in business. If your shell reports something like trial: command not found, make sure you have Twisted installed properly, and that the Twisted bin directory is in your PATH. If you don't know how to do this, get some local help, or figure it out by searching online for information on setting and changing environment variables for you operating system.

With those (one-time) preliminary steps out of the way, let's perform the tests. Run trial calculus.test.test_base_1 from the command line from the calculus directory. You should see the following output (though your files are probably not in /tmp:

$ trial calculus.test.test_base_1
calculus.test.test_base_1
  CalculationTestCase
    test_add ...                                                         [FAIL]
    test_divide ...                                                      [FAIL]
    test_multiply ...                                                    [FAIL]
    test_subtract ...                                                    [FAIL]

===============================================================================
[FAIL]
Traceback (most recent call last):
  File "/tmp/calculus/test/test_base_1.py", line 8, in test_add
    self.assertEqual(result, 11)
twisted.trial.unittest.FailTest: not equal:
a = None
b = 11


calculus.test.test_base_1.CalculationTestCase.test_add
===============================================================================
[FAIL]
Traceback (most recent call last):
  File "/tmp/calculus/test/test_base_1.py", line 23, in test_divide
    self.assertEqual(result, 2)
twisted.trial.unittest.FailTest: not equal:
a = None
b = 2


calculus.test.test_base_1.CalculationTestCase.test_divide
===============================================================================
[FAIL]
Traceback (most recent call last):
  File "/tmp/calculus/test/test_base_1.py", line 18, in test_multiply
    self.assertEqual(result, 60)
twisted.trial.unittest.FailTest: not equal:
a = None
b = 60


calculus.test.test_base_1.CalculationTestCase.test_multiply
===============================================================================
[FAIL]
Traceback (most recent call last):
  File "/tmp/calculus/test/test_base_1.py", line 13, in test_subtract
    self.assertEqual(result, 4)
twisted.trial.unittest.FailTest: not equal:
a = None
b = 4


calculus.test.test_base_1.CalculationTestCase.test_subtract
-------------------------------------------------------------------------------
Ran 4 tests in 0.042s

FAILED (failures=4)

How to interpret this output? You get a list of the individual tests, each followed by its result. By default, failures are printed at the end, but this can be changed with the -e (or --rterrors) option.

One very useful thing in this output is the fully-qualified name of the failed tests. This appears at the bottom of each =-delimited area of the output. This allows you to copy and paste it to just run a single test you're interested in. In our example, you could run trial calculus.test.test_base_1.CalculationTestCase.test_subtract from the shell.

Note that trial can use different reporters to modify its output. Run trial --help-reporters to see a list of reporters.

The tests can be run by trial in multiple ways:

The first 3 versions using full qualified names are strongly encouraged: they are much more reliable and they allow you to easily be more selective in your test runs.

You'll notice that Trial create a _trial_temp directory in the directory where you run the tests. This has a file called test.log which contains the log output of the tests (created using log.msg or log.err functions). Examine this file if you add logging to your tests.

Making the tests pass

Now that we have a working test framework in place, and our tests are failing (as expected) we can go and try to implement the correct API. We'll do that in a new version of the above base_1 module, calculus/base_2.py:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

# -*- test-case-name: calculus.test.test_base_2 -*-

class Calculation(object):
    def add(self, a, b):
        return a + b

    def subtract(self, a, b):
        return a - b

    def multiply(self, a, b):
        return a * b

    def divide(self, a, b):
        return a / b
Source listing - listings/trial/calculus/base_2.py

We'll also create a new version of test_base_1 which imports and tests this new implementation, in calculus/test_base_2.py:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

from calculus.base_2 import Calculation
from twisted.trial import unittest



class CalculationTestCase(unittest.TestCase):

    def test_add(self):
        calc = Calculation()
        result = calc.add(3, 8)
        self.assertEqual(result, 11)


    def test_subtract(self):
        calc = Calculation()
        result = calc.subtract(7, 3)
        self.assertEqual(result, 4)


    def test_multiply(self):
        calc = Calculation()
        result = calc.multiply(12, 5)
        self.assertEqual(result, 60)


    def test_divide(self):
        calc = Calculation()
        result = calc.divide(12, 5)
        self.assertEqual(result, 2)
test_base_2 - listings/trial/calculus/test/test_base_2.py
is a copy of test_base_1, but with the import changed. Run trial again as above, and your tests should now pass:

$ trial calculus.test.test_base_2

Running 4 tests.
calculus.test.test_base
  CalculationTestCase
    test_add ...                                                           [OK]
    test_divide ...                                                        [OK]
    test_multiply ...                                                      [OK]
    test_subtract ...                                                      [OK]

-------------------------------------------------------------------------------
Ran 4 tests in 0.067s

PASSED (successes=4)

Factoring out common test logic

You'll notice that our test file contains redundant code. Let's get rid of that. Python's unit testing framework allows your test class to define a setUp method that is called before each test method in the class. This allows you to add attributes to self that can be used in tests methods. We'll also add a parameterized test method to further simplify the code.

Note that a test class may also provide the counterpart of setUp, named tearDown, which will be called after each test (whether successful or not). tearDown is mainly used for post-test cleanup purposes. We will not use tearDown until later.

Create calculus/test/test_base_2b.py as follows:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

from calculus.base_2 import Calculation
from twisted.trial import unittest



class CalculationTestCase(unittest.TestCase):
    def setUp(self):
        self.calc = Calculation()


    def _test(self, operation, a, b, expected):
        result = operation(a, b)
        self.assertEqual(result, expected)


    def test_add(self):
        self._test(self.calc.add, 3, 8, 11)


    def test_subtract(self):
        self._test(self.calc.subtract, 7, 3, 4)


    def test_multiply(self):
        self._test(self.calc.multiply, 6, 9, 54)


    def test_divide(self):
        self._test(self.calc.divide, 12, 5, 2)
Source listing - listings/trial/calculus/test/test_base_2b.py

Much cleaner, no?

We'll now add some additional error tests. Testing just for successful use of the API is generally not enough, especially if you expect your code to be used by others. Let's make sure the Calculation class raises exceptions if someone tries to call its methods with arguments that cannot be converted to integers.

We arrive at calculus/test/test_base_3.py:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52

from calculus.base_3 import Calculation
from twisted.trial import unittest



class CalculationTestCase(unittest.TestCase):
    def setUp(self):
        self.calc = Calculation()


    def _test(self, operation, a, b, expected):
        result = operation(a, b)
        self.assertEqual(result, expected)


    def _test_error(self, operation):
        self.assertRaises(TypeError, operation, "foo", 2)
        self.assertRaises(TypeError, operation, "bar", "egg")
        self.assertRaises(TypeError, operation, [3], [8, 2])
        self.assertRaises(TypeError, operation, {"e": 3}, {"r": "t"})


    def test_add(self):
        self._test(self.calc.add, 3, 8, 11)


    def test_subtract(self):
        self._test(self.calc.subtract, 7, 3, 4)


    def test_multiply(self):
        self._test(self.calc.multiply, 6, 9, 54)


    def test_divide(self):
        self._test(self.calc.divide, 12, 5, 2)


    def test_errorAdd(self):
        self._test_error(self.calc.add)


    def test_errorSubtract(self):
        self._test_error(self.calc.subtract)


    def test_errorMultiply(self):
        self._test_error(self.calc.multiply)


    def test_errorDivide(self):
        self._test_error(self.calc.divide)
Source listing - listings/trial/calculus/test/test_base_3.py

We've added four new tests and one general-purpose function, _test_error. This function uses the assertRaises method, which takes an exception class, a function to run and its arguments, and checks that calling the function on the arguments does indeed raise the given exception.

If you run the above, you'll see that not all tests fail. In Python it's often valid to add and multiply objects of different and even differing types, so the code in the add and mutiply tests does not raise an exception and those tests therefore fail. So let's add explicit type conversion to our API class. This brings us to calculus/base_3.py:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

# -*- test-case-name: calculus.test.test_base_3 -*-

class Calculation(object):
    def _make_ints(self, *args):
        try:
            return map(int, args)
        except ValueError:
            raise TypeError("Coudln't coerce arguments to integers: %s" % args)

    def add(self, a, b):
        a, b = self._make_ints(a, b)
        return a + b

    def subtract(self, a, b):
        a, b = self._make_ints(a, b)
        return a - b

    def multiply(self, a, b):
        a, b = self._make_ints(a, b)
        return a * b

    def divide(self, a, b):
        a, b = self._make_ints(a, b)
        return a / b
Source listing - listings/trial/calculus/base_3.py

Here the _make_ints helper function tries to convert a list into a list of equivalent integers, and raises a TypeError in case the conversion goes wrong.

Note: The int conversion can also raise a TypeError if passed something of the wrong type, such as a list. We'll just let that exception go by as TypeError is already what we want in case something goes wrong.

Twisted specific testing

Up to this point we've been doing fairly standard Python unit testing. With only a few cosmetic changes (most importantly, directly importing unittest instead of using Twisted's unittest version) we could make the above tests run using Python's standard library unit testing framework.

Here we will assume a basic familiarity with Twisted's network I/O, timing, and Deferred APIs. If you haven't already read them, you should read the documentation on Writing Servers, Writing Clients, and Deferreds.

Now we'll get to the real point of this tutorial and take advantage of Trial to test Twisted code.

Testing a protocol

We'll now create a custom protocol to invoke our class from within a telnet-like session. We'll remotely call commands with arguments and read back the response. The goal will be to test our network code without creating sockets.

Creating and testing the server

First we'll write the tests, and then explain what they do. The first version of the remote test code is:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

from calculus.remote_1 import RemoteCalculationFactory
from twisted.trial import unittest
from twisted.test import proto_helpers



class RemoteCalculationTestCase(unittest.TestCase):
    def setUp(self):
        factory = RemoteCalculationFactory()
        self.proto = factory.buildProtocol(('127.0.0.1', 0))
        self.tr = proto_helpers.StringTransport()
        self.proto.makeConnection(self.tr)


    def _test(self, operation, a, b, expected):
        self.proto.dataReceived('%s %d %d\r\n' % (operation, a, b))
        self.assertEqual(int(self.tr.value()), expected)


    def test_add(self):
        return self._test('add', 7, 6, 13)


    def test_subtract(self):
        return self._test('subtract', 82, 78, 4)


    def test_multiply(self):
        return self._test('multiply', 2, 8, 16)


    def test_divide(self):
        return self._test('divide', 14, 3, 4)
Source listing - listings/trial/calculus/test/test_remote_1.py

To fully understand this client, it helps a lot to be comfortable with the Factory/Protocol/Transport pattern used in Twisted.

We first create a protocol factory object. Note that we have yet to see the RemoteCalculationFactory class. It is in calculus/remote_1.py below. We call buildProtocol to ask the factory to build us a protocol object that knows how to talk to our server. We then make a fake network transport, an instance of StringTransport class (note that test packages are generally not part of Twisted's public API; proto_helpers is an exception). This fake transport is the key to the communications. It is used to emulate a network connection without a network. The address and port passed to buildProtocol are typically used by the factory to choose to immediately deny remote connections; since we're using a fake transport, we can choose any value that will be acceptable to the factory. In this case the factory just ignores the address, so we don't need to pick anything in particular.

Testing protocols without the use of real network connections is both simple and recommended when testing Twisted code. Even though there are many tests in Twisted that use the network, most good tests don't. The problem with unit tests and networking is that networks aren't reliable. We cannot know that they will exhibit reasonable behavior all the time. This creates intermittent test failures due to network vagaries. Right now we're trying to test our Twisted code, not network reliability. By setting up and using a fake transport, we can write 100% reliable tests. We can also test network failures in a deterministic manner, another important part of your complete test suite.

The final key to understanding this client code is the _test method. The call to dataReceived simulates data arriving on the network transport. But where does it arrive? It's handed to the lineReceived method of the protocol instance (in calculus/remote_1.py below). So the client is essentially tricking the server into thinking it has received the operation and the arguments over the network. The server (once again, see below) hands the work off to its CalculationProxy object which in turn hands it to its Calculation instance. The result is written back via sendLine (into the fake string transport object), and is then immediately available to the client, who fetches it with tr.value() and checks that it has the expected value. So there's quite a lot going on behind the scenes in the two-line _test method above.

Finally, let's see the implementation of this protocol. Put the following into calculus/remote_1.py:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

# -*- test-case-name: calculus.test.test_remote_1 -*-

from twisted.protocols import basic
from twisted.internet import protocol
from calculus.base_3 import Calculation



class CalculationProxy(object):
    def __init__(self):
        self.calc = Calculation()
        for m in ['add', 'subtract', 'multiply', 'divide']:
            setattr(self, 'remote_%s' % m, getattr(self.calc, m))



class RemoteCalculationProtocol(basic.LineReceiver):
    def __init__(self):
        self.proxy = CalculationProxy()


    def lineReceived(self, line):
        op, a, b = line.split()
        a = int(a)
        b = int(b)
        op = getattr(self.proxy, 'remote_%s' % (op,))
        result = op(a, b)
        self.sendLine(str(result))



class RemoteCalculationFactory(protocol.Factory):
    protocol = RemoteCalculationProtocol



def main():
    from twisted.internet import reactor
    from twisted.python import log
    import sys
    log.startLogging(sys.stdout)
    reactor.listenTCP(0, RemoteCalculationFactory())
    reactor.run()


if __name__ == "__main__":
    main()
Source listing - listings/trial/calculus/remote_1.py

As mentioned, this server creates a protocol that inherits from basic.LineReceiver, and then a factory that uses it as protocol. The only trick is the CalculationProxy object, which calls Calculation methods through remote_* methods. This pattern is used frequently in Twisted, because it is very explicit about what methods you are making accessible.

If you run this test (trial calculus.test.test_remote_1), everything should be fine. You can also run a server to test it with a telnet client. To do that, call python calculus/remote_1.py. You should have the following output:

2008-04-25 10:53:27+0200 [-] Log opened.
2008-04-25 10:53:27+0200 [-] __main__.RemoteCalculationFactory starting on 46194
2008-04-25 10:53:27+0200 [-] Starting factory <__main__.RemoteCalculationFactory instance at 0x846a0cc>

46194 is replaced by a random port. You can then call telnet on it:

$ telnet localhost 46194
Trying 127.0.0.1...
Connected to localhost.
Escape character is '^]'.
add 4123 9423
13546

It works!

Creating and testing the client

Of course, what we build is not particulary useful for now : we'll now build a client to our server, to be able to use it inside a Python program. And it will serve our next purpose.

Create calculus/test/test_client_1.py:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

from calculus.client_1 import RemoteCalculationClient
from twisted.trial import unittest
from twisted.test import proto_helpers



class ClientCalculationTestCase(unittest.TestCase):
    def setUp(self):
        self.tr = proto_helpers.StringTransport()
        self.proto = RemoteCalculationClient()
        self.proto.makeConnection(self.tr)


    def _test(self, operation, a, b, expected):
        d = getattr(self.proto, operation)(a, b)
        self.assertEqual(self.tr.value(), '%s %d %d\r\n' % (operation, a, b))
        self.tr.clear()
        d.addCallback(self.assertEqual, expected)
        self.proto.dataReceived("%d\r\n" % (expected,))
        return d


    def test_add(self):
        return self._test('add', 7, 6, 13)


    def test_subtract(self):
        return self._test('subtract', 82, 78, 4)


    def test_multiply(self):
        return self._test('multiply', 2, 8, 16)


    def test_divide(self):
        return self._test('divide', 14, 3, 4)
Source listing - listings/trial/calculus/test/test_client_1.py

It's really symmetric to the server test cases. The only tricky part is that we don't use a client factory. We're lazy, and it's not very useful in the client part, so we instantiate the protocol directly.

Incidentally, we have introduced a very important concept here: the tests now return a Deferred object, and the assertion is done in a callback. The important thing to do here is to not forget to return the Deferred. If you do, your tests will pass even if nothing is asserted. That's also why it's important to make tests fail first: if your tests pass whereas you know they shouldn't, there is a problem in your tests.

We'll now add the remote client class to produce calculus/client_1.py:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

# -*- test-case-name: calculus.test.test_client_1 -*-

from twisted.protocols import basic
from twisted.internet import defer



class RemoteCalculationClient(basic.LineReceiver):
    def __init__(self):
        self.results = []


    def lineReceived(self, line):
        d = self.results.pop(0)
        d.callback(int(line))


    def _sendOperation(self, op, a, b):
        d = defer.Deferred()
        self.results.append(d)
        line = "%s %d %d" % (op, a, b)
        self.sendLine(line)
        return d


    def add(self, a, b):
        return self._sendOperation("add", a, b)


    def subtract(self, a, b):
        return self._sendOperation("subtract", a, b)


    def multiply(self, a, b):
        return self._sendOperation("multiply", a, b)


    def divide(self, a, b):
        return self._sendOperation("divide", a, b)
Source listing - listings/trial/calculus/client_1.py

More good pratices

Testing scheduling

When testing code that involves the passage of time, waiting e.g. for a two hour timeout to occur in a test is not very realistic. Twisted provides a solution to this, the Clock class that allows one to simulate the passage of time.

As an example we'll test the code for client request timeout: since our client uses TCP it can hang for a long time (firewall, connectivity problems, etc...). So generally we need to implement timeouts on the client side. Basically it's just that we send a request, don't receive a response and expect a timeout error to be triggered after a certain duration. 

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

from calculus.client_2 import RemoteCalculationClient, ClientTimeoutError

from twisted.internet import task
from twisted.trial import unittest
from twisted.test import proto_helpers



class ClientCalculationTestCase(unittest.TestCase):
    def setUp(self):
        self.tr = proto_helpers.StringTransportWithDisconnection()
        self.clock = task.Clock()
        self.proto = RemoteCalculationClient()
        self.tr.protocol = self.proto
        self.proto.callLater = self.clock.callLater
        self.proto.makeConnection(self.tr)


    def _test(self, operation, a, b, expected):
        d = getattr(self.proto, operation)(a, b)
        self.assertEqual(self.tr.value(), '%s %d %d\r\n' % (operation, a, b))
        self.tr.clear()
        d.addCallback(self.assertEqual, expected)
        self.proto.dataReceived("%d\r\n" % (expected,))
        return d


    def test_add(self):
        return self._test('add', 7, 6, 13)


    def test_subtract(self):
        return self._test('subtract', 82, 78, 4)


    def test_multiply(self):
        return self._test('multiply', 2, 8, 16)


    def test_divide(self):
        return self._test('divide', 14, 3, 4)


    def test_timeout(self):
        d = self.proto.add(9, 4)
        self.assertEqual(self.tr.value(), 'add 9 4\r\n')
        self.clock.advance(self.proto.timeOut)
        return self.assertFailure(d, ClientTimeoutError)
Source listing - listings/trial/calculus/test/test_client_2.py

What happens here? We instantiate our protocol as usual, the only trick is to create the clock, and assign proto.callLater to clock.callLater. Thus, every callLater calls in the protocol will finish before clock.advance() returns.

In the new test (test_timeout), we call clock.advance, that simulates and advance in time (logically it's similar to a time.sleep call). And we just have to verify that our Deferred got a timeout error.

Let's implement that in our code.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

# -*- test-case-name: calculus.test.test_client_2 -*-

from twisted.protocols import basic
from twisted.internet import defer, reactor



class ClientTimeoutError(Exception):
    pass



class RemoteCalculationClient(basic.LineReceiver):

    callLater = reactor.callLater
    timeOut = 60

    def __init__(self):
        self.results = []


    def lineReceived(self, line):
        d, callID = self.results.pop(0)
        callID.cancel()
        d.callback(int(line))


    def _cancel(self, d):
        d.errback(ClientTimeoutError())


    def _sendOperation(self, op, a, b):
        d = defer.Deferred()
        callID = self.callLater(self.timeOut, self._cancel, d)
        self.results.append((d, callID))
        line = "%s %d %d" % (op, a, b)
        self.sendLine(line)
        return d


    def add(self, a, b):
        return self._sendOperation("add", a, b)


    def subtract(self, a, b):
        return self._sendOperation("subtract", a, b)


    def multiply(self, a, b):
        return self._sendOperation("multiply", a, b)


    def divide(self, a, b):
        return self._sendOperation("divide", a, b)
Source listing - listings/trial/calculus/client_2.py

The only important thing here is to not forget to cancel our callLater when everything went fine.

Cleaning up after tests

This chapter is mainly intended for people that want to have sockets or processes created in their tests. If it's still not obvious, you must try to avoid that like the plague, because it ends up with a lot of problems, one of them being intermittent failures. And intermittent failures are the plague of automated tests.

To actually test that, we'll launch a server with our protocol.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

from calculus.remote_1 import RemoteCalculationFactory
from calculus.client_2 import RemoteCalculationClient

from twisted.trial import unittest
from twisted.internet import reactor, protocol



class RemoteRunCalculationTestCase(unittest.TestCase):

    def setUp(self):
        factory = RemoteCalculationFactory()
        self.port = reactor.listenTCP(0, factory, interface="127.0.0.1")
        self.client = None


    def tearDown(self):
        if self.client is not None:
            self.client.transport.loseConnection()
        return self.port.stopListening()


    def _test(self, op, a, b, expected):
        creator = protocol.ClientCreator(reactor, RemoteCalculationClient)
        def cb(client):
            self.client = client
            return getattr(self.client, op)(a, b
                ).addCallback(self.assertEqual, expected)
        return creator.connectTCP('127.0.0.1', self.port.getHost().port
            ).addCallback(cb)


    def test_add(self):
        return self._test("add", 5, 9, 14)


    def test_subtract(self):
        return self._test("subtract", 47, 13, 34)


    def test_multiply(self):
        return self._test("multiply", 7, 3, 21)


    def test_divide(self):
        return self._test("divide", 84, 10, 8)
Source listing - listings/trial/calculus/test/test_remote_2.py

Recent versions of trial will fail loudly if you remove the stopListening call, which is good.

Also, you should be aware that tearDown will called in any case, after success or failure. So don't expect that every objects you created in the test method are present, because your tests may have failed in the middle.

Trial also has a addCleanup method, which makes these kind of cleanups easy and removes the need for tearDown . For example, you could remove the code in _test this way:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

def setUp(self):
    factory = RemoteCalculationFactory()
    self.port = reactor.listenTCP(0, factory, interface="127.0.0.1")
    self.addCleanup(self.port.stopListening)

def _test(self, op, a, b, expected):
    creator = protocol.ClientCreator(reactor, RemoteCalculationClient)
    def cb(client):
        self.addCleanup(self.client.transport.loseConnection)
        return getattr(client, op)(a, b).addCallback(self.assertEqual, expected)
    return creator.connectTCP('127.0.0.1', self.port.getHost().port).addCallback(cb)

This remove the need of a tearDown method, and you don't have to check for the value of self.client: you only call addCleanup when the client is created.

Handling logged errors

Currently, if you send an invalid command or invalid arguments to our server, it logs an exception and closes the connection. This is a perfectly valid behavior, but for the sake of this tutorial, we want to return an error to the user if he sends invalid operators, and log any errors on server side. So we'll want a test like this:

1
2
3

def test_invalidParameters(self):
    self.proto.dataReceived('add foo bar\r\n')
    self.assertEqual(self.tr.value(), "error\r\n")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

# -*- test-case-name: calculus.test.test_remote_1 -*-

from twisted.protocols import basic
from twisted.internet import protocol
from twisted.python import log
from calculus.base_3 import Calculation



class CalculationProxy(object):
    def __init__(self):
        self.calc = Calculation()
        for m in ['add', 'subtract', 'multiply', 'divide']:
            setattr(self, 'remote_%s' % m, getattr(self.calc, m))



class RemoteCalculationProtocol(basic.LineReceiver):
    def __init__(self):
        self.proxy = CalculationProxy()


    def lineReceived(self, line):
        op, a, b = line.split()
        op = getattr(self.proxy, 'remote_%s' % (op,))
        try:
            result = op(a, b)
        except TypeError:
            log.err()
            self.sendLine("error")
        else:
            self.sendLine(str(result))



class RemoteCalculationFactory(protocol.Factory):
    protocol = RemoteCalculationProtocol



def main():
    from twisted.internet import reactor
    from twisted.python import log
    import sys
    log.startLogging(sys.stdout)
    reactor.listenTCP(0, RemoteCalculationFactory())
    reactor.run()


if __name__ == "__main__":
    main()
Source listing - listings/trial/calculus/remote_2.py

If you try something like that, it will not work. Here is the output you should have:

trial calculus.test.test_remote_3.RemoteCalculationTestCase.test_invalidParameters
calculus.test.test_remote_3
  RemoteCalculationTestCase
    test_invalidParameters ...                                          [ERROR]

===============================================================================
[ERROR]: calculus.test.test_remote_3.RemoteCalculationTestCase.test_invalidParameters

Traceback (most recent call last):
  File "/tmp/calculus/remote_2.py", line 27, in lineReceived
    result = op(a, b)
  File "/tmp/calculus/base_3.py", line 11, in add
    a, b = self._make_ints(a, b)
  File "/tmp/calculus/base_3.py", line 8, in _make_ints
    raise TypeError
exceptions.TypeError:
-------------------------------------------------------------------------------
Ran 1 tests in 0.004s

FAILED (errors=1)

At first, you could think there is a problem, because you catch this exception. But in fact trial doesn't let you do that without controlling it: you must expect logged errors and clean them. To do that, you have to use the flushLoggedErrors method. You call it with the exception you expect, and it returns the list of exceptions logged since the start of the test. Generally, you'll want to check that this list has the expected length, or possibly that each exception has an expected message. We do the former in our test:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

from calculus.remote_2 import RemoteCalculationFactory
from twisted.trial import unittest
from twisted.test import proto_helpers



class RemoteCalculationTestCase(unittest.TestCase):
    def setUp(self):
        factory = RemoteCalculationFactory()
        self.proto = factory.buildProtocol(('127.0.0.1', 0))
        self.tr = proto_helpers.StringTransport()
        self.proto.makeConnection(self.tr)


    def _test(self, operation, a, b, expected):
        self.proto.dataReceived('%s %d %d\r\n' % (operation, a, b))
        self.assertEqual(int(self.tr.value()), expected)


    def test_add(self):
        return self._test('add', 7, 6, 13)


    def test_subtract(self):
        return self._test('subtract', 82, 78, 4)


    def test_multiply(self):
        return self._test('multiply', 2, 8, 16)


    def test_divide(self):
        return self._test('divide', 14, 3, 4)


    def test_invalidParameters(self):
        self.proto.dataReceived('add foo bar\r\n')
        self.assertEqual(self.tr.value(), "error\r\n")
        errors = self.flushLoggedErrors(TypeError)
        self.assertEqual(len(errors), 1)
Source listing - listings/trial/calculus/test/test_remote_3.py

Resolve a bug

A bug was left over during the development of the timeout (probably several bugs, but that's not the point), concerning the reuse of the protocol when you got a timeout: the connection is not dropped, so you can get timeout forever. Generally an user will come to you saying "I have this strange problem on my crappy network environment. It seems you could solve it with doing XXX at YYY."

Actually, this bug can be corrected several ways. But if you correct it without adding tests, one day you'll face a big problem: regression. So the first step is adding a failing test.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63

from calculus.client_3 import RemoteCalculationClient, ClientTimeoutError

from twisted.internet import task
from twisted.trial import unittest
from twisted.test import proto_helpers



class ClientCalculationTestCase(unittest.TestCase):
    def setUp(self):
        self.tr = proto_helpers.StringTransportWithDisconnection()
        self.clock = task.Clock()
        self.proto = RemoteCalculationClient()
        self.tr.protocol = self.proto
        self.proto.callLater = self.clock.callLater
        self.proto.makeConnection(self.tr)


    def _test(self, operation, a, b, expected):
        d = getattr(self.proto, operation)(a, b)
        self.assertEqual(self.tr.value(), '%s %d %d\r\n' % (operation, a, b))
        self.tr.clear()
        d.addCallback(self.assertEqual, expected)
        self.proto.dataReceived("%d\r\n" % (expected,))
        return d


    def test_add(self):
        return self._test('add', 7, 6, 13)


    def test_subtract(self):
        return self._test('subtract', 82, 78, 4)


    def test_multiply(self):
        return self._test('multiply', 2, 8, 16)


    def test_divide(self):
        return self._test('divide', 14, 3, 4)


    def test_timeout(self):
        d = self.proto.add(9, 4)
        self.assertEqual(self.tr.value(), 'add 9 4\r\n')
        self.clock.advance(self.proto.timeOut)
        return self.assertFailure(d, ClientTimeoutError)


    def test_timeoutConnectionLost(self):
        called = []
        def lost(arg):
            called.append(True)
        self.proto.connectionLost = lost

        d = self.proto.add(9, 4)
        self.assertEqual(self.tr.value(), 'add 9 4\r\n')
        self.clock.advance(self.proto.timeOut)

        def check(ignore):
            self.assertEqual(called, [True])
        return self.assertFailure(d, ClientTimeoutError).addCallback(check)
Source listing - listings/trial/calculus/test/test_client_3.py

What have we done here ?

For doing that, we then use the TimeoutMixin class, that does almost everything we want. The great thing is that it almost changes nothing to our class.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53

# -*- test-case-name: calculus.test.test_client -*-

from twisted.protocols import basic, policies
from twisted.internet import defer



class ClientTimeoutError(Exception):
    pass



class RemoteCalculationClient(object, basic.LineReceiver, policies.TimeoutMixin):

    def __init__(self):
        self.results = []
        self._timeOut = 60

    def lineReceived(self, line):
        self.setTimeout(None)
        d = self.results.pop(0)
        d.callback(int(line))


    def timeoutConnection(self):
        for d in self.results:
            d.errback(ClientTimeoutError())
        self.transport.loseConnection()


    def _sendOperation(self, op, a, b):
        d = defer.Deferred()
        self.results.append(d)
        line = "%s %d %d" % (op, a, b)
        self.sendLine(line)
        self.setTimeout(self._timeOut)
        return d


    def add(self, a, b):
        return self._sendOperation("add", a, b)


    def subtract(self, a, b):
        return self._sendOperation("subtract", a, b)


    def multiply(self, a, b):
        return self._sendOperation("multiply", a, b)


    def divide(self, a, b):
        return self._sendOperation("divide", a, b)
Source listing - listings/trial/calculus/client_3.py

Code coverage

Code coverage is one of the aspects of software testing that shows how much your tests cross (cover) the code of your program. There are different kind of measures: path coverage, condition coverage, statement coverage... We'll only consider statement coverage here, whether a line has been executed or not.

Trial has an option to generate the statement coverage of your tests. This option is --coverage. It creates a coverage directory in _trial_temp, with a file .cover for every modules used during the tests. The ones interesting for us are calculus.base.cover and calculus.remote.cover. In front of each line is the number of times you went through during the tests, or the marker '>>>>>>' if the line was not covered. If you went through all the tutorial to this point, you should have complete coverage :).

Again, this is only another useful pointer, but it doesn't mean your code is perfect: your tests should consider every possibile input and output, to get full coverage (condition, path, etc.) as well .

Conclusion

So what did you learn in this document?

  • How to use the trial command-line tool to run your tests
  • How to use string transports to test individual clients and servers without creating sockets
  • If you really want to create sockets, how to cleanly do it so that it doesn't have bad side effects
  • And some small tips you can't live without.
If one of the topics still looks cloudy to you, please give us your feedback! You can file tickets to improve this document on the Twisted web site.

Index

Version: 12.0.0