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=nmbtornado.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_testdef 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

关键字：Python

版权声明

本文来自互联网用户投稿，文章观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处。如若内容有涉嫌抄袭侵权/违法违规/事实不符，请点击 举报 进行投诉反馈！