Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
Application to demonstrate testing tornado websockets
'''
Application to demonstrate testing tornado websockets.
Try it wiith: python -m tornado.testing discover
'''
from tornado import testing, httpserver, gen, websocket
from ws import APP
class TestChatHandler(testing.AsyncTestCase):
def setUp(self):
super(TestChatHandler, self).setUp()
server = httpserver.HTTPServer(APP)
socket, self.port = testing.bind_unused_port()
server.add_socket(socket)
def _mk_connection(self):
return websocket.websocket_connect(
'ws://localhost:{}/'.format(self.port)
)
@gen.coroutine
def _mk_client(self):
c = yield self._mk_connection()
# Discard the hello
# This could be any initial handshake, which needs to be generalized
# for most of the tests.
_ = yield c.read_message()
raise gen.Return(c)
@testing.gen_test
def test_hello(self):
c = yield self._mk_connection()
# Get the initial hello from the server.
response = yield c.read_message()
# Make sure that we got a 'hello' not 'bye'
self.assertEqual('hello', response)
@testing.gen_test
def test_echo(self):
# A client with the hello taken care of.
c = yield self._mk_client()
# Send a 'foo' to the server.
c.write_message("foo")
# Get the 'foo' back.
response = yield c.read_message()
# Make sure that we got a 'foo' back and not 'bar'.
self.assertEqual('foo', response)
'''
Application to demonstrate testing tornado websockets
'''
from tornado import web, websocket
# We don't override `data_received`
# pylint: disable=W0223
class Echo(websocket.WebSocketHandler):
# Open allows for any number arguments, unlike what pylint thinks.
# pylint: disable=W0221
def open(self):
self.write_message('hello')
def on_message(self, message):
self.write_message(message)
def on_close(self):
self.write_message('bye')
APP = web.Application([
(r"/", Echo),
])
if __name__ == "__main__":
APP.listen(5000)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.