mock非常适合写单元测试, 用它patch掉网络请求的返回值即可

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()

测试用的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()

给上面的FooHandler加装饰器(放在coroutine上面), 这种一般的使用场景就是加缓存或者计时之类…
因为异步的里面是个generator, 所以最里面包的一层还是要加coroutine并且用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