Mocking Tornado Asynchronous Code and Writing Decorators • quietin

mock is well suited to unit tests: patch the return value of the network request.

async_func.py:

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import signal
import errno
import tornado.gen
import tornado.ioloop
import tornado.web
import tornado.httpclient
import tornado.httpserver

HOST = 'baidu.com'

@tornado.gen.coroutine
def search(keyword):
    client = tornado.httpclient.AsyncHTTPClient()
    url = 'http://%s/s?wd=%s' % (HOST, keyword)
    resp = yield client.fetch(url)
    raise tornado.gen.Return(resp.body)

class FooHandler(tornado.web.RequestHandler):
    @tornado.gen.coroutine
    def get(self):
        keyword = self.get_argument('wd')
        result = yield search(keyword)
        self.finish(result)

def handle_signal_kill(sig, frame):
    print 'Catch signal: %s' % errno.errorcode(sig)
    tornado.ioloop.IOLoop.instance().stop()

if __name__ == '__main__':
    app = tornado.web.Application(
        handlers=[
            (r"/", FooHandler),
        ]
    )
    http_server = tornado.httpserver.HTTPServer(app)
    http_server.listen(8888)

    signal.signal(signal.SIGINT, handle_signal_kill)
    signal.signal(signal.SIGQUIT, handle_signal_kill)
    signal.signal(signal.SIGTERM, handle_signal_kill)
    signal.signal(signal.SIGHUP, handle_signal_kill)

    # test url: http://127.0.0.1:8888/?wd=nmb
    tornado.ioloop.IOLoop.current().start()

The test code in test.py:

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import mock
import tornado.gen
import tornado.ioloop
import tornado.testing
import tornado.concurrent
import unittest
import tornado.testing

from async_func import search

class AsyncTestCase(tornado.testing.AsyncTestCase):
    def setUp(self):
        super(AsyncTestCase, self).setUp()

    @mock.patch('tornado.httpclient.AsyncHTTPClient')
    @tornado.testing.gen_test
    def test_fetch(self, AsyncHTTPClient):
        AsyncHTTPClient.return_value = mock.MagicMock()
        future = tornado.concurrent.Future()
        future.set_result(mock.MagicMock(body='mycontent'))
        x = mock.MagicMock()
        x.fetch.return_value = future
        AsyncHTTPClient.return_value = x
        result = yield search('nmb48')
        self.assertIn(result, 'mycontent test')

unittest.main()

To decorate FooHandler, place the decorator above coroutine. Typical uses include caching or timing. Because the asynchronous function is a generator, the innermost wrapper must also use coroutine and return its value through gen:

def cache_it(func):
    @tornado.gen.coroutine
    def _deco(self):
        print 'decrator work'
        # save cache or other...
        result = yield func(self)
        raise tornado.gen.Return(result)
    return _deco

Originally published on SegmentFault.