tornado异步的mock以及装饰器
mock非常适合写单元测试, 用它patch掉网络请求的返回值即可async_func.py#!/usr/bin/env python# -*- coding:utf-8 -*-import signalimport errnoimport tornado.genimport tornado.ioloopimport tornado.webimport tornado.h
·
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
来自:https://segmentfault.com/a/1190000005647942
更多推荐
已为社区贡献2条内容
所有评论(0)