Bite-sized Twisted: Asynchronous Testing and the Clock

In this, the seventh Bite-sized Twisted post, I'll go over a little asynchronous testing and the Clock.

Impatient runners

Let's test that a lit bomb explodes with a message when it should. Or in other words, that a function returns a Deferred which fires after a time. For convenience, I'm going to put the code to be tested in the same file as the test.

We'll need to import some things:

from unittest import TestCase
from twisted.internet import reactor
from twisted.internet.defer import Deferred

test_bomb1.py:1-3

Here's the test:

class BombTest(TestCase):

    def test_explodes(self):
        def check(result):
            self.assertEqual(result, 'exploded')
        d = lightBomb(2, 'exploded')
        d.addCallback(check)

test_bomb1.py:5-11

And here's the function:

def lightBomb(fuse, message):
    defer = Deferred()
    reactor.callLater(fuse, defer.callback, message)
    return defer

test_bomb1.py:12-16

When we run it, we see that it passes the test:

test_bomb1
  BombTest
    test_explodes ...                                                      [OK]

-------------------------------------------------------------------------------
Ran 1 tests in 0.001s

PASSED (successes=1)

trial test_bomb1.py

Yay!

No, not yay. Change the test to this:

    def test_explodes(self):
        def check(result):
            print "I was executed with %r" % result     
        d = lightBomb(2, 'exploded')
        d.addCallback(check)

test_bomb2.py:7-11

And see that check is never called:

test_bomb2
  BombTest
    test_explodes ...                                                      [OK]

-------------------------------------------------------------------------------
Ran 1 tests in 0.001s

PASSED (successes=1)

trial test_bomb2.py

What gives?

Wait up!

Does it seem a little suspicious that the bomb is supposed to go off in 2 seconds and the test runs in less than 1 second? The problem is that the test runner is not waiting for the Deferred to be called back. To wait for the Deferred, we'll have to use twisted.trial.unittest.TestCase and return the Deferred from the test method:

from twisted.trial.unittest import TestCase
from twisted.internet import reactor
from twisted.internet.defer import Deferred

test_bomb3.py:1-3

    def test_explodes(self):
        def check(result):
            self.assertEqual(result, 'exploded')
        d = lightBomb(2, 'exploded')
        d.addCallback(check)
        return d                                    

test_bomb3.py:7-12

test_bomb3
  BombTest
    test_explodes ...                                                      [OK]

-------------------------------------------------------------------------------
Ran 1 tests in 2.012s

PASSED (successes=1)

trial test_bomb3.py

See how the test took more than 2 seconds to run this time?

Timeout

What if lightBomb looked like this instead?

def lightBomb(fuse, message):
    defer = Deferred()
    return defer

test_hang.py:14-16

Run the test and it will hang. Add a timeout to fail any test that takes too long:

class BombTest(TestCase):

    timeout = 3

test_timeout.py:5-7

test_timeout
  BombTest
    test_explodes ...                                                   [ERROR]

===============================================================================
[ERROR]
Traceback (most recent call last):
Failure: twisted.internet.defer.TimeoutError: <test_timeout.BombTest testMethod=test_explodes> (test_explodes) still running at 3.0 secs

test_timeout.BombTest.test_explodes
-------------------------------------------------------------------------------
Ran 1 tests in 3.015s

FAILED (errors=1)

trial test_timeout.py

Hello, Clock

If you have a lot of tests dealing with scheduled events and timing, running all of them will take a long time. To avoid that, we can simulate the passage of time with the Clock.

Here's the same test as test_bomb3.py using the Clock (changed lines are highlighted):

from twisted.trial.unittest import TestCase
from twisted.internet.task import Clock
from twisted.internet import reactor
from twisted.internet.defer import Deferred

class BombTest(TestCase):

    def test_explodes(self):
        def check(result):
            self.assertEqual(result, 'exploded')
        clock = Clock()
        d = lightBomb(2, 'exploded', reactor=clock)
        d.addCallback(check)
        clock.advance(2)
        return d                                    

def lightBomb(fuse, message, reactor=reactor):
    defer = Deferred()
    reactor.callLater(fuse, defer.callback, message)
    return defer

test_clock.py

test_clock
  BombTest
    test_explodes ...                                                      [OK]

-------------------------------------------------------------------------------
Ran 1 tests in 0.005s

PASSED (successes=1)

trial test_clock.py

Yes! Much faster! Here are the Clock docs if you are interested. Note in test_clock.py how we had to make lightBomb accept an optional reactor.

Boom tests

All of the code for the current version of boom was test-driven. Perhaps reading some of the tests will be enlightening -- though I make no claim at being a testing expert. The test code is available on GitHub.