摘要：非常適合寫單元測試用它掉網絡請求的返回值即可測試用的給上面的加裝飾器放在上面這種一般的使用場景就是加緩存或者計時之類因為異步的里面是個所以最里面包的一層還是要加并且用返回

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