Async frameworks like Tornado scramble our usual unittest strategies: how can you validate the outcome when you do not know when to expect it? Tornado ships with a tornado.testing module that provides two solutions: the wait / stop pattern, and gen_test.
Wait / Stop
To begin, let us say we are writing an async application with feature like Gmail’s undo send: when I click “send”, Gmail delays a few seconds before actually sending the email. It is a funny phenomenon, that during the seconds after clicking “sending” I experience a special clarity about my email. It was too angry, or I forgot an attachment, most often both. If I click the “undo” button in time, the email reverts to a draft and I can tone it down, add the attachment, and send it again.
To write an application with this feature, we will need an asynchronous “delay” function, and we must test it. If we were testing a normal blocking delay function we could use unittest.TestCase from the standard library:
import time
import unittest
from my_application import delay
class MyTestCase(unittest.TestCase):
def test_delay(self):
start = time.time()
delay(1)
duration = time.time() - start
self.assertAlmostEqual(duration, 1, places=2)When we run this, it prints:
Ran 1 test in 1.000s OK
And if we replace delay(1) with delay(2) it fails as expected:
=======================================================
FAIL: test_delay (delay0.MyTestCase)
-------------------------------------------------------
Traceback (most recent call last):
File "delay0.py", line 12, in test_delay
self.assertAlmostEqual(duration, 1, places=2)
AssertionError: 2.000854969024658 != 1 within 2 places
-------------------------------------------------------
Ran 1 test in 2.002s
FAILED (failures=1)Great! What about testing a delay_async(seconds, callback) function?
def test_delay(self):
start = time.time()
delay_async(1, callback=) # What goes here?
duration = time.time() - start
self.assertAlmostEqual(duration, 1, places=2)An asynchronous “delay” function can’t block the caller, so it must take a callback and execute it once the delay is over. (In fact we are just reimplementing Tornado’s call_later, but please pretend for pedagogy’s sake this is a new function that we must test.) To test our delay_async, we will try a series of testing techniques until we have effectively built Tornado’s test framework from scratch—you will see why we need special test tools for async code and how Tornado’s tools work.
So, we define a function done to measure the delay, and pass it as the callback to delay_async:
def test_delay(self):
start = time.time()
def done():
duration = time.time() - start
self.assertAlmostEqual(duration, 1, places=2)
delay_async(1, done)If we run this:
Ran 1 test in 0.001s OK
Success! …right? But why does it only take a millisecond? And what happens if we delay by two seconds instead?
def test_delay(self):
start = time.time()
def done():
duration = time.time() - start
self.assertAlmostEqual(duration, 1, places=2)
delay_async(2, done)Run it again:
Ran 1 test in 0.001s OK
Something is very wrong here. The test appears to pass instantly, regardless of the argument to delay_async, because we neither start the event loop nor wait for it to complete. We have to actually pause the test until the callback has executed:
def test_delay(self):
start = time.time()
io_loop = IOLoop.instance()
def done():
duration = time.time() - start
self.assertAlmostEqual(duration, 1, places=2)
io_loop.stop()
delay_async(1, done)
io_loop.start()Now if we run the test with a delay of one second:
Ran 1 test in 1.002s OK
That looks better. And if we delay for two seconds?
ERROR:tornado.application:Exception in callback
Traceback (most recent call last):
File "site-packages/tornado/ioloop.py", line 568, in _run_callback
ret = callback()
File "site-packages/tornado/stack_context.py", line 275, in null_wrapper
return fn(*args, **kwargs)
File "delay3.py", line 16, in done
self.assertAlmostEqual(duration, 1, places=2)
File "unittest/case.py", line 845, in assertAlmostEqual
raise self.failureException(msg)
AssertionError: 2.001540184020996 != 1 within 2 placesThe test appears to fail, as expected, but there are a few problems. First, notice that it is not the unittest that prints the traceback: it is Tornado’s application logger. We do not get the unittest’s characteristic output. Second, the process is now hung and remains so until I type Control-C. Why?
The bug is here:
def done():
duration = time.time() - start
self.assertAlmostEqual(duration, 1, places=2)
io_loop.stop()Since the failed assertion raises an exception, we never reach the call to io_loop.stop(), so the loop continues running and the process does not exit. We need to register an exception handler. Exception handling with callbacks is convoluted; we have to use a stack context to install a handler with Tornado:
from tornado.stack_context import ExceptionStackContext
class MyTestCase(unittest.TestCase):
def test_delay(self):
start = time.time()
io_loop = IOLoop.instance()
def done():
duration = time.time() - start
self.assertAlmostEqual(duration, 1, places=2)
io_loop.stop()
self.failure = None
def handle_exception(typ, value, tb):
io_loop.stop()
self.failure = value
return True # Stop propagation.
with ExceptionStackContext(handle_exception):
delay_async(2, callback=done)
io_loop.start()
if self.failure:
raise self.failureThe loop can now be stopped two ways: if the test passes, then done stops the loop as before. If it fails, handle_exception stores the error and stops the loop. At the end, if an error was stored we re-raise it to make the test fail:
=======================================================
FAIL: test_delay (delay4.MyTestCase)
-------------------------------------------------------
Traceback (most recent call last):
File "delay4.py", line 31, in test_delay
raise self.failure
File "tornado/ioloop.py", line 568, in _run_callback
ret = callback()
File "tornado/stack_context.py", line 343, in wrapped
raise_exc_info(exc)
File "<string>", line 3, in raise_exc_info
File "tornado/stack_context.py", line 314, in wrapped
ret = fn(*args, **kwargs)
File "delay4.py", line 17, in done
self.assertAlmostEqual(duration, 1, places=2)
AssertionError: 2.0015950202941895 != 1 within 2 places
-------------------------------------------------------
Ran 1 test in 2.004s
FAILED (failures=1)Now the test ends promptly, whether it succeeds or fails, with unittest’s typical output.
This is a lot of tricky code to write just to test a trivial delay function, and it seems hard to get right each time. What does Tornado provide for us? Its AsyncTestCase gives us start and stop methods to control the event loop. If we then move the duration-testing outside the callback we radically simplify our test:
from tornado import testing
class MyTestCase(testing.AsyncTestCase):
def test_delay(self):
start = time.time()
delay_async(1, callback=self.stop)
self.wait()
duration = time.time() - start
self.assertAlmostEqual(duration, 1, places=2)gen_test
But modern async code is not primarily written with callbacks: these days we use coroutines. Let us begin a new example test, one that uses Motor, my asynchronous MongoDB driver for Tornado. Although Motor supports the old callback style, it encourages you to use coroutines and “yield” statements, so we can write some Motor code to demonstrate Tornado coroutines and unittesting.
To begin, say we want to execute find_one and test its return value:
from motor import MotorClient
from tornado import testing
class MyTestCase(testing.AsyncTestCase):
def setUp(self):
super().setUp()
self.client = MotorClient()
def test_find_one(self):
collection = self.client.test.collection
document = yield collection.find_one({'_id': 1})
self.assertEqual({'_id': 1, 'key': 'value'}, document)Notice the “yield” statement: whenever you call a Motor method that does I/O, you must use “yield” to pause the current function and wait for the returned Future object to be resolved to a value. Including a yield statement makes this function a generator. But now there is a problem:
TypeError: Generator test methods should be decorated with tornado.testing.gen_test
Tornado smartly warns us that our test method is merely a generator—we must decorate it with gen_test. Otherwise the test method simply stops at the first yield, and never reaches the assert. It needs a coroutine driver to run it to completion:
from tornado.testing import gen_test
class MyTestCase(testing.AsyncTestCase):
# ... same setup ...
@gen_test
def test_find_one(self):
collection = self.client.test.collection
document = yield collection.find_one({'_id': 1})
self.assertEqual({'_id': 1, 'key': 'value'}, document)But now when I run the test, it fails:
AssertionError: {'key': 'value', '_id': 1} != NoneWe need to insert some data in setUp so that find_one can find it! Since Motor is asynchronous, we cannot call its insert method directly from setUp, we must run the insertion in a coroutine as well:
from tornado import gen, testing
class MyTestCase(testing.AsyncTestCase):
def setUp(self):
super().setUp()
self.client = MotorClient()
self.setup_coro()
@gen.coroutine
def setup_coro(self):
collection = self.client.test.collection
# Clean up from prior runs:
yield collection.remove()
yield collection.insert({'_id': 0})
yield collection.insert({'_id': 1, 'key': 'value'})
yield collection.insert({'_id': 2})Now when I run the test:
AssertionError: {'key': 'value', '_id': 1} != NoneIt still fails! When I check in the mongo shell whether my data was inserted, only two of the three expected documents are there:
> db.collection.find()
{ "_id" : 0 }
{ "_id" : 1, "key" : "value" }Why is it incomplete? Furthermore, since the document I actually query is there, why did the test fail?
When I called self.setup_coro() in setUp, I launched it as a concurrent coroutine. It began running, but I did not wait for it to complete before beginning the test, so the test may reach its find_one statement before the second document is inserted. Furthermore, test_find_one can fail quickly enough that setup_coro does not insert its third document before the whole test suite finishes, stopping the event loop and preventing the final document from ever being inserted.
Clearly I must wait for the setup coroutine to complete before beginning the test. Tornado’s run_sync method is designed for uses like this:
class MyTestCase(testing.AsyncTestCase):
def setUp(self):
super().setUp()
self.client = MotorClient()
self.io_loop.run_sync(self.setup_coro)With my setup coroutine correctly executed, now test_find_one passes.
Further Study
Now we have seen two techniques that make async testing with Tornado as convenient and reliable as standard unittests. To learn more, see my page of links related to this article.
Plus, stay tuned for the next book in the Architecture of Open Source Applications series. It will be called “500 Lines or Less”, and my chapter is devoted to the implementation of coroutines in asyncio and Python 3.
| Reference: | Eventually Correct: Async Testing With Tornado from our WCG partner Jesse Davis at the A. Jesse Jiryu Davis blog. |
