noseresult

nosetests --exe  --nocapture -vv
nose.config: INFO: Ignoring files matching ['^\\.', '^_', '^setup\\.py$']
test_connection (test.WebSocketTest) ... ok
test_no_action (test.WebSocketTest) ... ERROR
test_websocket_http_fail (test.WebSocketTest) ... ok

======================================================================
ERROR: test_no_action (test.WebSocketTest)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/Users/eloycotopereiro/.virtualenvs/zendesk/lib/python2.7/site-packages/tornado/testing.py", line 427, in wrapper
    functools.partial(f, self), timeout=timeout)
  File "/Users/eloycotopereiro/.virtualenvs/zendesk/lib/python2.7/site-packages/tornado/ioloop.py", line 369, in run_sync
    raise TimeoutError('Operation timed out after %s seconds' % timeout)
TimeoutError: Operation timed out after 5 seconds

----------------------------------------------------------------------
Ran 3 tests in 5.106s

FAILED (errors=1)

server.py

import tornado.ioloop
import tornado.websocket
import json


class WSHandler(tornado.websocket.WebSocketHandler):
    def open(self):
        self.write_message("Welcome")

    def on_message(self, message):
        try:
            message = json.loads(message)
            if not isinstance(message, dict):
                return None
        except:
            return None

        if not message.get('action'):
            return None

        if message.get('action') == 'ping':
            self.write_message(json.dumps({'action': 'pong'}))

    def on_close(self):
        return None

application = tornado.web.Application([
    (r'/ws', WSHandler),
])

if __name__ == "__main__":
    application.listen(9090)
    tornado.ioloop.IOLoop.instance().start()

test.py

from server import application
from tornado.concurrent import Future
from tornado.httpclient import HTTPError
from tornado.ioloop import TimeoutError
from tornado.testing import AsyncHTTPTestCase, gen_test
from tornado.websocket import websocket_connect
import json


class WebSocketTest(AsyncHTTPTestCase):

    def setUp(self):
        super(WebSocketTest, self).setUp()

    def get_app(self):
        self.close_future = Future()
        return application

    @gen_test
    def test_connection(self):
        ws = yield websocket_connect(
            'ws://localhost:%d/ws' % self.get_http_port(),
            io_loop=self.io_loop)
        response = yield ws.read_message()
        self.assertEqual(response, 'Welcome')

    @gen_test
    def test_websocket_http_fail(self):
        with self.assertRaises(HTTPError) as cm:
            yield websocket_connect(
                'ws://localhost:%d/notfound' % self.get_http_port(),
                io_loop=self.io_loop)
        self.assertEqual(cm.exception.code, 404)

    @gen_test
    def test_no_action(self):

        ws = yield websocket_connect(
            'ws://localhost:%d/ws' % self.get_http_port(),
            io_loop=self.io_loop)
        response = yield ws.read_message()
        self.assertEqual(response, 'Welcome')

        ws.write_message(json.dumps({'action': 'ping'}))
        response = yield ws.read_message()
        response = json.loads(response)
        self.assertEqual(response.get('action'), 'pong')

        ws.write_message(json.dumps('noaction'))
        with self.assertRaises(TimeoutError):
            yield ws.read_message()