資訊專欄INFORMATION COLUMN
摘要：非常適合寫單元測試用它掉網(wǎng)絡(luò)請求的返回值即可測試用的給上面的加裝飾器放在上面這種一般的使用場景就是加緩存或者計時之類因為異步的里面是個所以最里面包的一層還是要加并且用返回
mock非常適合寫單元測試, 用它patch掉網(wǎng)絡(luò)請求的返回值即可
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
文章版權(quán)歸作者所有，未經(jīng)允許請勿轉(zhuǎn)載,若此文章存在違規(guī)行為，您可以聯(lián)系管理員刪除。
轉(zhuǎn)載請注明本文地址：http://systransis.cn/yun/45483.html
摘要：主要是為了實現(xiàn)系統(tǒng)之間的雙向解耦而實現(xiàn)的。問題及優(yōu)化隊列過長問題使用上述方案的異步非阻塞可能會依賴于的任務(wù)隊列長度，若隊列中的任務(wù)過多，則可能導(dǎo)致長時間等待，降低效率。 Tornado和Celery介紹 1.Tornado Tornado是一個用python編寫的一個強大的、可擴展的異步HTTP服務(wù)器，同時也是一個web開發(fā)框架。tornado是一個非阻塞式web服務(wù)器，其速度相當(dāng)快。...
摘要：譯者說于年月日發(fā)布，該版本正式支持的關(guān)鍵字，并且用舊版本編譯同樣可以使用這兩個關(guān)鍵字，這無疑是一種進步。其次，這是最后一個支持和的版本了，在后續(xù)的版本了會移除對它們的兼容。 譯者說 Tornado 4.3于2015年11月6日發(fā)布，該版本正式支持Python3.5的async/await關(guān)鍵字，并且用舊版本CPython編譯Tornado同樣可以使用這兩個關(guān)鍵字，這無疑是一種進步。其次...
摘要：譯者說于年月日發(fā)布，該版本正式支持的關(guān)鍵字，并且用舊版本編譯同樣可以使用這兩個關(guān)鍵字，這無疑是一種進步。其次，這是最后一個支持和的版本了，在后續(xù)的版本了會移除對它們的兼容。本節(jié)最好直接在或者閱讀，以獲得更好的閱讀體驗格式支持。 譯者說 Tornado 4.3于2015年11月6日發(fā)布，該版本正式支持Python3.5的async/await關(guān)鍵字，并且用舊版本CPython編譯Torn...
閱讀 3567·2021-11-25 09:43
閱讀 3145·2021-10-08 10:04
閱讀 1635·2019-08-26 12:20
閱讀 2067·2019-08-26 12:09
閱讀 609·2019-08-23 18:25
閱讀 3581·2019-08-23 17:54
閱讀 2337·2019-08-23 17:50
閱讀 814·2019-08-23 14:33
