How to test that tornado read_message got nothing to read

Question

I have a Tornado chat and I'm doing some tests, most client messages generate a reply from the server, but others must not generate any reply.

I managed to do it with this code, waiting for the read timeout to occur, there is a better way to do it?

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()

Also there is a problem when test ends

======================================================================
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

score 1 · Accepted Answer · answered Oct 22 '15 at 01:33

In general, it's difficult to test for a negative: how long do you wait before you conclude that the thing you're testing for will never happen? It's better to rearrange things so that the test can be expressed in positive terms. That's difficult to do in this toy example, but consider the following handler:

class RealtimeHandler(tornado.websocket.WebSocketHandler):
    def on_message(self, message):
        if int(message) % 2 == 1:
            self.write_message('%s is odd' % message)

In this case you could test it by sending the messages 1, 2, and 3, and asserting that you get two responses, "1 is odd" and "3 is odd".

The StopIteration failure you see is slightly surprising to me: I would not expect a timeout to be catchable within the @gen_test method, so doing so may have unexpected results, but I wouldn't have expected it to turn into a StopIteration. In any case, it's better to restructure the test so that you don't have to rely on timeouts. And if you do need a timeout, use gen.with_timeout so you can control the timeout from inside the test instead of relying on the one from outside in @gen_test.

The real code is more complex, I test with multiple clients how some messages are broadcasted to specific users and others not. I agree that although I have mocked async long tasks relying in timeout is not correct, maybe for not overloading tests I can add a helper function and check for positive answer as you mention like self.assertPingPong(ws) — Chemary, Oct 22 '15 at 09:06

score -1 · Answer 2 · answered Jun 07 '17 at 19:31

Just to illustrate the @Ben Darnell answer.

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()

How to test that tornado read_message got nothing to read

2 Answers2