Tornado + Motor + Unittest

Edit: After working on this a bit more, I’ve found useful tools in Tornado’s testing suite. I’ve written a little bit more in this post!

Tornado Unittesting is a pain in the ass. If you go about writing your unittest the same old way, before long:

(╯°□°)╯︵ ┻━┻

I’m writing up a simple service using Tornado + Motor (MongoDB with Tornado), and in my journeys, I’ve found several issues and gotchas that are worth sharing:

Testing Asynchronous Code

Go ahead and read Davis’ article (below) — the problem is explained pretty well, and it even offers some solutions!

Tornado Unittesting: Eventually Correct
I‘m a fan of Tornado, one of the major async web frameworks for Python, but unittesting async code is a total pain. I’m going to review what the problem is, look at some klutzy solutions, and propose a better way. If you don’t care what I have to say and you just want to steal my code, get it on GitHub.

In summary, if the asynchronous behavior is not addressed, all your tests will always pass — unittest exits its tests before the asynchronous methods can actually run. If you do attempt to handle the asynchronous behavior by playing with Tornado’s IOLoop, chances are the AssertionErrors are thrown while the IOLoop is running and are consumed by the IOLoop. The result? Unittest still passes.

┻━━┻ ︵╰(゜Д゜)╯︵ ┻━━┻

I have no doubt that Davis’ solutions take care of the issue; however, I am unsatisfied with relying on timeouts — something about that irks me.

Somewhat Simpler Solution

import unittest

class AsyncTest(unittest.testCase):
# Ensure IOLoop stops to prevent blocking tests
def callback(self, func):
def wrapper(*args, **kwargs):
func(*args, **kwargs)
except Exception as e:
self.error = e
return wrapper

def wait(self):

def setUp(self):
self.error = None
super(AsyncTest, self).setUp()

def tearDown(self):
if self.error:
super(AsyncTest, self).tearDown()

This custom class wraps TestCase and uses unittest’s own testing life-cycle to re-raise the errors that IOLoop usually consumes on tearDown. It seems a bit silly, but it works! I am no Python / unittest expert, so if this is grossly horrible, please let me know.

Let’s see it in action.
class DBTest(AsyncTest):
def test_bad_calls(self):
def error_callback(result, error):
self.assertIs(result, None)
self.assertIsNot(error, None)
random_object = object()
auth_db.find_user_id(random_object, error_callback)

──┬ ノ( ゜-゜ノ)

For more context, this is from this Github Repo.

Note, this example is very geared towards testing Tornado with Motor, specifically making database queries with callbacks.

Async — Spinning Up Mock Server

Now that I’ve addressed making asynchronous calls to Motor — what about testing Tornado servers? I have similar issues in that I both need to spin up a mock server and get around the IOLoop. This is less of an issue or ‘gotcha’ and more of simply an annoyance.

Thankfully, we can write up nice Python decorators to abstract away all the annoying bits of code.

def server(port, routes):
def decorator(func):
def wrapper(*args):
server, thread = _startServer(port, routes)
except Exception:
_stopServer(server, thread)
return wrapper
return decorator

def _startServer(port, routes):
# Start Server
application = Application(routes, debug = False)
application.logging = "none"
server = HTTPServer(application)
# Start nonblocking IOLoop
thread = threading.Thread(target=IOLoop.instance().start)
return server, thread

def _stopServer(server, thread):

Okay, that was pretty simple. There are a few things to note. First of all, server simply starts and stops a server before and after the test. It gets around IOLoop by doing this on a separate thread.

You might notice a funky try / except / finally block catching a general Exception and simply re-raising it. The key is the finally block which ensures that we kill the server to avoid tests getting stuck, or issues with reusing the same port.

Motor — why you change my data?

For my unittests, I use a file that contains all the mock data I use for consistently. Specifically, I have a user object I use for all my requests that involve passing user data

USER = {
“name”: “Christopher Lee”,
“location”: “Cambridge, MA”,
“phone”: “123456789”

At some point while running unittests, the tests start to fail because json fails to encode USER.

WHAT? (╯°□°)╯︵ ┻━┻

It turns out this guy edits the dictionary you pass through! Talk about an elusive bug.

save(to_save, manipulate=True, safe=None, check_keys=True, callback=None, **kwargs)

Manipulate! That’s definitely a huge ‘gotcha’. Manipulate is literally what it sounds like. Motor, why you got to change my data?

What’s worse is if you pass in False for manipulate, the method will no longer return the id and instead return None. WHAT. Why not simply leave my data alone, and pass back the id anyway? Well, a simple solution — send a copy of the dictionary instead.

save(user.copy(), callback = callback)
Like what you read? Give Chris Lee a round of applause.

From a quick cheer to a standing ovation, clap to show how much you enjoyed this story.