如何测试龙卷风 read_message 没有什么可读的
How to test that tornado read_message got nothing to read
我有一个 Tornado 聊天,我正在做一些测试,大多数客户端消息都会从服务器生成回复,但其他人不能生成任何回复。
我用这段代码成功做到了,等待读取超时发生,有更好的方法吗?
import json
import tornado
from tornado.httpclient import HTTPRequest
from tornado.web import Application
from tornado.websocket import websocket_connect
from tornado.testing import AsyncHTTPTestCase, gen_test
class RealtimeHandler(tornado.websocket.WebSocketHandler):
def on_message(self, message):
if message != 'Hi':
self.write_message('Hi there')
return
class ChatTestCase(AsyncHTTPTestCase):
def get_app(self):
return Application([
('/rt', RealtimeHandler),
])
@gen_test
def test_no_reply(self):
request = HTTPRequest('ws://127.0.0.1:%d/rt' % self.get_http_port())
ws = yield websocket_connect(request)
ws.write_message('Hi')
with self.assertRaises(tornado.ioloop.TimeoutError):
response = yield ws.read_message()
测试结束也有问题
======================================================================
ERROR: test_no_reply (myproj.tests.ChatTestCase)
----------------------------------------------------------------------
Traceback (most recent call last):
File "/home/ubuntu/my_env/local/lib/python2.7/site-packages/tornado/testing.py", line 120, in __call__
result = self.orig_method(*args, **kwargs)
File "/home/ubuntu/my_env/local/lib/python2.7/site-packages/tornado/testing.py", line 506, in post_coroutine
self._test_generator.throw(e)
StopIteration
一般来说,很难测试是否定的:您要等多久才能断定您正在测试的事情永远不会发生?最好重新安排事情,以便可以用积极的方式表达测试。在这个玩具示例中很难做到这一点,但请考虑以下处理程序:
class RealtimeHandler(tornado.websocket.WebSocketHandler):
def on_message(self, message):
if int(message) % 2 == 1:
self.write_message('%s is odd' % message)
在这种情况下,您可以通过发送消息 1、2 和 3 来测试它,并断言您收到两个响应,“1 是奇数”和“3 是奇数”。
你看到的 StopIteration
失败让我有点吃惊:我不希望在 @gen_test
方法中捕获超时,所以这样做可能会产生意想不到的结果,但我不会没想到会变成StopIteration
。无论如何,最好重构测试,这样您就不必依赖超时。如果您确实需要超时,请使用 gen.with_timeout
这样您就可以从测试内部控制超时,而不是依赖 @gen_test
.
中的外部超时
只是为了说明@Ben Darnell 。
from tornado import gen
class ChatTestCase(AsyncHTTPTestCase):
def get_app(self):
return Application([
('/rt', RealtimeHandler),
])
@gen_test
def test_no_reply(self):
request = HTTPRequest('ws://127.0.0.1:%d/rt' % self.get_http_port())
ws = yield websocket_connect(request)
ws.write_message('Hi')
with self.assertRaises(gen.TimeoutError):
response = yield gen.with_timeout(datetime.timedelta(seconds=4), ws.read_message()
我有一个 Tornado 聊天,我正在做一些测试,大多数客户端消息都会从服务器生成回复,但其他人不能生成任何回复。
我用这段代码成功做到了,等待读取超时发生,有更好的方法吗?
import json
import tornado
from tornado.httpclient import HTTPRequest
from tornado.web import Application
from tornado.websocket import websocket_connect
from tornado.testing import AsyncHTTPTestCase, gen_test
class RealtimeHandler(tornado.websocket.WebSocketHandler):
def on_message(self, message):
if message != 'Hi':
self.write_message('Hi there')
return
class ChatTestCase(AsyncHTTPTestCase):
def get_app(self):
return Application([
('/rt', RealtimeHandler),
])
@gen_test
def test_no_reply(self):
request = HTTPRequest('ws://127.0.0.1:%d/rt' % self.get_http_port())
ws = yield websocket_connect(request)
ws.write_message('Hi')
with self.assertRaises(tornado.ioloop.TimeoutError):
response = yield ws.read_message()
测试结束也有问题
======================================================================
ERROR: test_no_reply (myproj.tests.ChatTestCase)
----------------------------------------------------------------------
Traceback (most recent call last):
File "/home/ubuntu/my_env/local/lib/python2.7/site-packages/tornado/testing.py", line 120, in __call__
result = self.orig_method(*args, **kwargs)
File "/home/ubuntu/my_env/local/lib/python2.7/site-packages/tornado/testing.py", line 506, in post_coroutine
self._test_generator.throw(e)
StopIteration
一般来说,很难测试是否定的:您要等多久才能断定您正在测试的事情永远不会发生?最好重新安排事情,以便可以用积极的方式表达测试。在这个玩具示例中很难做到这一点,但请考虑以下处理程序:
class RealtimeHandler(tornado.websocket.WebSocketHandler):
def on_message(self, message):
if int(message) % 2 == 1:
self.write_message('%s is odd' % message)
在这种情况下,您可以通过发送消息 1、2 和 3 来测试它,并断言您收到两个响应,“1 是奇数”和“3 是奇数”。
你看到的 StopIteration
失败让我有点吃惊:我不希望在 @gen_test
方法中捕获超时,所以这样做可能会产生意想不到的结果,但我不会没想到会变成StopIteration
。无论如何,最好重构测试,这样您就不必依赖超时。如果您确实需要超时,请使用 gen.with_timeout
这样您就可以从测试内部控制超时,而不是依赖 @gen_test
.
只是为了说明@Ben Darnell
from tornado import gen
class ChatTestCase(AsyncHTTPTestCase):
def get_app(self):
return Application([
('/rt', RealtimeHandler),
])
@gen_test
def test_no_reply(self):
request = HTTPRequest('ws://127.0.0.1:%d/rt' % self.get_http_port())
ws = yield websocket_connect(request)
ws.write_message('Hi')
with self.assertRaises(gen.TimeoutError):
response = yield gen.with_timeout(datetime.timedelta(seconds=4), ws.read_message()