andpp/mopidy
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

http: Remove conditional only needed for Tornado < 3.2

Browse Source
This commit is contained in:
Stein Magnus Jodal 2018-04-13 13:05:23 +02:00
parent 3e91f9819d
commit 41882c6395
1 changed files with 37 additions and 39 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

76
tests/http/test_handlers.py
View File

@ -41,48 +41,46 @@ class StaticFileHandlerTest(tornado.testing.AsyncHTTPTestCase):
response.headers['Cache-Control'], 'no-cache')
# We aren't bothering with skipIf as then we would need to "backport" gen_test
if hasattr(tornado.websocket, 'websocket_connect'):
class WebSocketHandlerTest(tornado.testing.AsyncHTTPTestCase):
class WebSocketHandlerTest(tornado.testing.AsyncHTTPTestCase):
def get_app(self):
self.core = mock.Mock()
return tornado.web.Application([
(r'/ws/?', handlers.WebSocketHandler, {'core': self.core})
])
def get_app(self):
self.core = mock.Mock()
return tornado.web.Application([
(r'/ws/?', handlers.WebSocketHandler, {'core': self.core})
])
def connection(self):
url = self.get_url('/ws').replace('http', 'ws')
return tornado.websocket.websocket_connect(url, self.io_loop)
def connection(self):
url = self.get_url('/ws').replace('http', 'ws')
return tornado.websocket.websocket_connect(url, self.io_loop)
@tornado.testing.gen_test
def test_invalid_json_rpc_request_doesnt_crash_handler(self):
# An uncaught error would result in no message, so this is just a
# simplistic test to verify this.
conn = yield self.connection()
conn.write_message('invalid request')
message = yield conn.read_message()
self.assertTrue(message)
@tornado.testing.gen_test
def test_invalid_json_rpc_request_doesnt_crash_handler(self):
# An uncaught error would result in no message, so this is just a
# simplistic test to verify this.
conn = yield self.connection()
conn.write_message('invalid request')
message = yield conn.read_message()
self.assertTrue(message)
@tornado.testing.gen_test
def test_broadcast_makes_it_to_client(self):
conn = yield self.connection()
handlers.WebSocketHandler.broadcast('message')
message = yield conn.read_message()
self.assertEqual(message, 'message')
@tornado.testing.gen_test
def test_broadcast_makes_it_to_client(self):
conn = yield self.connection()
handlers.WebSocketHandler.broadcast('message')
message = yield conn.read_message()
self.assertEqual(message, 'message')
@tornado.testing.gen_test
def test_broadcast_to_client_that_just_closed_connection(self):
conn = yield self.connection()
conn.stream.close()
handlers.WebSocketHandler.broadcast('message')
@tornado.testing.gen_test
def test_broadcast_to_client_that_just_closed_connection(self):
conn = yield self.connection()
conn.stream.close()
handlers.WebSocketHandler.broadcast('message')
@tornado.testing.gen_test
def test_broadcast_to_client_without_ws_connection_present(self):
yield self.connection()
# Tornado checks for ws_connection and raises WebSocketClosedError
# if it is missing, this test case simulates winning a race were
# this has happened but we have not yet been removed from clients.
for client in handlers.WebSocketHandler.clients:
client.ws_connection = None
handlers.WebSocketHandler.broadcast('message')
@tornado.testing.gen_test
def test_broadcast_to_client_without_ws_connection_present(self):
yield self.connection()
# Tornado checks for ws_connection and raises WebSocketClosedError
# if it is missing, this test case simulates winning a race were
# this has happened but we have not yet been removed from clients.
for client in handlers.WebSocketHandler.clients:
client.ws_connection = None
handlers.WebSocketHandler.broadcast('message')