2

我有一个 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
4

2 回答 2

1

一般来说，很难测试是否定的：你要等多久才能得出结论，你正在测试的事情永远不会发生？最好重新排列事物，以便可以用积极的方式表达测试。在这个玩具示例中很难做到这一点，但请考虑以下处理程序：

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。

于 2015-10-22T01:33:02.910 回答
-1

只是为了说明@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()
于 2017-06-07T19:31:40.570 回答