tornado异步的mock以及装饰器

摘要：非常适合写单元测试用它掉网络请求的返回值即可测试用的给上面的加装饰器放在上面这种一般的使用场景就是加缓存或者计时之类因为异步的里面是个所以最里面包的一层还是要加并且用返回

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