Dashboard > CI Development > ... > System Development Environment > Unit testing with Trial and Twisted
Log In   View a printable version of the current page.
CI Development
Unit testing with Trial and Twisted
Added by Paul Hubbard , last edited by Paul Hubbard on Apr 05, 2010  (view change)
Labels: 

Introduction

Writing unit and integration tests with an asycronous framework like Twisted can be quite challenging. This page presents some code and references to get you started.

The basic idea is that of test driven development; namely that writing the code and its test should be done at the same time. However, when the code is Twisted, it gets more complicated and testing is tricker. This attempts to explain how to write tests for protocols, services, factories and clients using the facilites in Trial and Twisted.

References

Requirements and imports

For most of these tests, these imports are required:

1
2
3
4
5
6
7
8
9
from twisted.trial import unittest
from twisted.internet import reactor
from twisted.internet.defer import inlineCallbacks, DeferredQueue
import logging

from ooici.protocol import ApplicationProtocol, Message

from magnet.preactor import Preactor
from magnet.protocol import ClientCreator as MCC

Testing a protocol

When I first wrote the attribute store, I factored the code into protocol and factory. The protocol was actually protocol plus service, so the StringIO tests worked well. You could do something nice and compact like

 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
class ASProtoTestCase(unittest.TestCase):
    def setUp(self):
        logging.basicConfig(level=logging.ERROR, \
                format='%(asctime)s %(levelname)s [%(funcName)s] %(message)s')

        factory = ASFactory()
        self.proto = factory.buildProtocol(('127.0.0.1', 0))
        self.tr = proto_helpers.StringTransport()
        self.proto.makeConnection(self.tr)

    def tearDown(self):
        self.tr.loseConnection()

    def _test(self, op, key, value, expectedRC=200):
        # Clear any old data in the pipe
        self.tr.clear()
        # Push into fake-net and see if the answer is as expected
        self.proto.dataReceived('%s %s %s\r\n' % (op, key, value))

        # First chunk should be http return code
        rc = int(self.tr.value().split()[0])
        self.assertEquals(rc, expectedRC)

    def test_write(self):
        return self._test('write', uuid.uuid4(), uuid.uuid4())

and have a single-line test. However, once you separate protocol from service, this fails because your protocol will reference

1
self.factory.function

which is undefined until instantiated via the factory. However, if you have a simpler program where the protocol does the work, please see this code.

Testing a protocol+service+factory

In this case, we instantiate a server object, and replace its sendLine method, with the effect that we exercise the factory, service and protocol, but not the transport or layers below it. (e.g. Magnet, AMQP)

Here's the code from the presentation:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
from twisted.internet import unittest

class ServerTestCase(unittest.TestCase):
    def testServer(self):
        l = []; f = PresenceFactory()
        p = f.buildProtocol(None)
        p.sendLine = l.append
        p.connectionMade()
        p.lineReceived('ISONLINE foo')
        self.assertEquals(l[-1], 'ERROR')

The main trick here is to instantiate the factory (which invokes the component registry, service and protocol) then manually invoke the steps in a real connection.

The other cute trick is this line:

1
p.sendLine = l.append

This reaches into the object and replaces its sendLine method with a hook to append our local array. So when we send it data, it replies straight into our local array!

You can see this used here, in the test_server_only method:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
l = []
logging.debug('Creating server')
preactor = yield Preactor()
serv = AttributeStoreService()
factory = IASFactory(serv)
p = factory.buildProtocol(None)
p.sendLine = l.append
p.connectionMade()

# Start of read/write tests
p.lineReceived('read trial:%s' % uuid.uuid4())
self.assertEquals(l[-1], '404 Key not found')

Testing just the factory

In this case, I'm using the factory to maintain persistent data between connections (though I should perhaps add a service and do it there!), so I want to instantiate the factory sans protocol, connection or transport:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
class PSTest(unittest.TestCase):
    def setUp(self):
        logging.basicConfig(level=logging.INFO, \
                format='%(asctime)s %(levelname)s [%(funcName)s] %(message)s')

        self.f = PSFactory()
        self.f._init_listeners()

    def tearDown(self):
        pass

    def test_sub_once(self):
        self.f.set_listener('http://.+?', 'meee')
        rc = self.f.get_listeners('http://foo.bar.com/')
        self.failUnless(len(rc) > 0)
        self.failUnlessEqual(rc[0], 'meee')

Works fine. No Preactor or reactor calls.

Testing client and server together

This is from the next page, where he shows how to create the server and its client, and then interact via the client API. Note that this adds the transport and client code, making it a full integration test. Here's a snippet from my code:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
logging.debug('Creating server')
preactor = yield Preactor()
serv = AttributeStoreService()
factory = IASFactory(serv)
preactor.listenMS(routingKey, factory)

logging.debug('Creating client')
# Now create a client to match
cc = MCC(reactor, preactor, ASClient)
asc = yield cc.connectMS(routingKey)

# Ready to operate the system now
# Start with a known-404 key
ans = yield asc.read('trial:%s' % uuid.uuid4())
reply = yield ans

rc, body = parseMsg(reply)
self.failUnlessEqual(rc, 404)

Here, you're creating the client and server as you would normally, running both in the reactor. See the test_client_server routine linked above for more tests.

Note that, as of 9/17/09, you still have to call preactor.stop() at the end of the test. This may change as Magnet evolves.

On-disk organization

Before 9/18/09, I was putting a test directory in the top level of the module. After reading the buildbot docs, I learned that it should be underneath the code directory. For the attribute store, here's how it looks:

base dir/attribute_store/test/

That way, you can invoke the tests like a namespace:

cd base_dir
trial attribute_store.test

which will automatically run all the tests.

Skipping tests

Sometimes you want to temporarily skip a failing test, perhaps because you know the code is broken and the errors are driving you mad. Here's how:

1
2
3
4
from twisted.trial import unittest

def test_fail(self):
    raise unittest.SkipTest('Broken right now due to Foo.bar')

this has the benefit of noting the skip in the test report, so that it's not forgotten:

===============================================================================
[SKIPPED]: attribute_store.test.test_service.ServiceTestCase.test_service

known-broken
-------------------------------------------------------------------------------
Ran 5 tests in 0.607s

PASSED (skips=1, successes=4)

Setting timeouts

By default, Twisted will wait 120 seconds for a test to complete. This is often too long for most code, and too short for big tests. To change it, add a call in your setUp method:

1
2
def setUp(self):
        self.timeout = 60

Parameter is seconds.

Waiting for results

In Twisted, calling sleep brings the reactor to a halt, but sometimes you need to wait for an external event or process to complete inside of a test. If you look at the test_service_with_dq test in the attribute store, you'll see the use of DeferredQueue for this purpose. The components' sendLine method is replaced by the queues' put method, and then you yield on get. Very cute. Here's the code:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
serv = AttributeStoreService()
yield serv._connect()
factory = IASFactory(serv)
proto = factory.buildProtocol(None)
# Output should get dropped into DQ...
dq = DeferredQueue()
proto.sendLine = dq.put

key = uuid.uuid4()
value = 'baz'

proto.lineReceived('write trial:%s %s' % (key, value))
logging.debug('Waiting for output...')
msg = yield dq.get()
logging.debug(msg)
rc, body = self.parseMsg(msg)
self.failUnlessEqual(rc, 200)

Integration with buildbot

Once you've got working tests, the next step is to invoke them via our buildbot system. This is something to be determined; right now you have to edit the buildbot configuration directly. If you place your tests as explained above, we should be able to auto-invoke them via a single

trial {packagename}.test

and this will handle the addition or removal of tests without changing buildbot.
Powered by Atlassian Confluence 2.7.1, the Enterprise Wiki. Bug/feature request - Atlassian news - Contact administrators