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