openstack/deb-python-eventlet
Code Issues Proposed changes

Python 3 compat: Improve WebSocket and tests

Browse Source
This commit is contained in:
Jakub Stasiak
2014-11-13 08:30:47 +00:00
parent 6d7633366b
commit 197000511e
2 changed files with 13 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
eventlet/websocket.py
View File

@@ -172,7 +172,7 @@ class WebSocketWSGI(object):
b"HTTP/1.1 101 Web Socket Protocol Handshake\r\n" b"HTTP/1.1 101 Web Socket Protocol Handshake\r\n"
b"Upgrade: WebSocket\r\n" b"Upgrade: WebSocket\r\n"
b"Connection: Upgrade\r\n" b"Connection: Upgrade\r\n"
b"WebSocket-Origin: " + environ.get('HTTP_ORIGIN') + b"\r\n" b"WebSocket-Origin: " + six.b(environ.get('HTTP_ORIGIN')) + b"\r\n"
b"WebSocket-Location: " + six.b(location) + b"\r\n\r\n" b"WebSocket-Location: " + six.b(location) + b"\r\n\r\n"
) )
elif self.protocol_version == 76: elif self.protocol_version == 76:

24
tests/websocket_test.py
View File

@@ -60,7 +60,7 @@ class TestWebSocket(_TestBase):
self.assertEqual(resp.status, 400) self.assertEqual(resp.status, 400)
self.assertEqual(resp.getheader('connection'), 'close') self.assertEqual(resp.getheader('connection'), 'close')
self.assertEqual(resp.read(), '') self.assertEqual(resp.read(), b'')
def test_incomplete_headers_76(self): def test_incomplete_headers_76(self):
# First test: Missing Connection: # First test: Missing Connection:
@@ -77,7 +77,7 @@ class TestWebSocket(_TestBase):
self.assertEqual(resp.status, 400) self.assertEqual(resp.status, 400)
self.assertEqual(resp.getheader('connection'), 'close') self.assertEqual(resp.getheader('connection'), 'close')
self.assertEqual(resp.read(), '') self.assertEqual(resp.read(), b'')
# Now, miss off key2 # Now, miss off key2
headers = dict(kv.split(': ') for kv in [ headers = dict(kv.split(': ') for kv in [
@@ -95,7 +95,7 @@ class TestWebSocket(_TestBase):
self.assertEqual(resp.status, 400) self.assertEqual(resp.status, 400)
self.assertEqual(resp.getheader('connection'), 'close') self.assertEqual(resp.getheader('connection'), 'close')
self.assertEqual(resp.read(), '') self.assertEqual(resp.read(), b'')
def test_correct_upgrade_request_75(self): def test_correct_upgrade_request_75(self):
connect = [ connect = [
@@ -112,13 +112,13 @@ class TestWebSocket(_TestBase):
sock.sendall(six.b('\r\n'.join(connect) + '\r\n\r\n')) sock.sendall(six.b('\r\n'.join(connect) + '\r\n\r\n'))
result = sock.recv(1024) result = sock.recv(1024)
# The server responds the correct Websocket handshake # The server responds the correct Websocket handshake
self.assertEqual(result, '\r\n'.join([ self.assertEqual(result, six.b('\r\n'.join([
'HTTP/1.1 101 Web Socket Protocol Handshake', 'HTTP/1.1 101 Web Socket Protocol Handshake',
'Upgrade: WebSocket', 'Upgrade: WebSocket',
'Connection: Upgrade', 'Connection: Upgrade',
'WebSocket-Origin: http://localhost:%s' % self.port, 'WebSocket-Origin: http://localhost:%s' % self.port,
'WebSocket-Location: ws://localhost:%s/echo\r\n\r\n' % self.port, 'WebSocket-Location: ws://localhost:%s/echo\r\n\r\n' % self.port,
])) ])))
def test_correct_upgrade_request_76(self): def test_correct_upgrade_request_76(self):
connect = [ connect = [
@@ -137,14 +137,14 @@ class TestWebSocket(_TestBase):
sock.sendall(six.b('\r\n'.join(connect) + '\r\n\r\n^n:ds[4U')) sock.sendall(six.b('\r\n'.join(connect) + '\r\n\r\n^n:ds[4U'))
result = sock.recv(1024) result = sock.recv(1024)
# The server responds the correct Websocket handshake # The server responds the correct Websocket handshake
self.assertEqual(result, '\r\n'.join([ self.assertEqual(result, six.b('\r\n'.join([
'HTTP/1.1 101 WebSocket Protocol Handshake', 'HTTP/1.1 101 WebSocket Protocol Handshake',
'Upgrade: WebSocket', 'Upgrade: WebSocket',
'Connection: Upgrade', 'Connection: Upgrade',
'Sec-WebSocket-Origin: http://localhost:%s' % self.port, 'Sec-WebSocket-Origin: http://localhost:%s' % self.port,
'Sec-WebSocket-Protocol: ws', 'Sec-WebSocket-Protocol: ws',
'Sec-WebSocket-Location: ws://localhost:%s/echo\r\n\r\n8jKS\'y:G*Co,Wxa-' % self.port, 'Sec-WebSocket-Location: ws://localhost:%s/echo\r\n\r\n8jKS\'y:G*Co,Wxa-' % self.port,
])) ])))
def test_query_string(self): def test_query_string(self):
# verify that the query string comes out the other side unscathed # verify that the query string comes out the other side unscathed
@@ -163,7 +163,7 @@ class TestWebSocket(_TestBase):
sock.sendall(six.b('\r\n'.join(connect) + '\r\n\r\n^n:ds[4U')) sock.sendall(six.b('\r\n'.join(connect) + '\r\n\r\n^n:ds[4U'))
result = sock.recv(1024) result = sock.recv(1024)
self.assertEqual(result, '\r\n'.join([ self.assertEqual(result, six.b('\r\n'.join([
'HTTP/1.1 101 WebSocket Protocol Handshake', 'HTTP/1.1 101 WebSocket Protocol Handshake',
'Upgrade: WebSocket', 'Upgrade: WebSocket',
'Connection: Upgrade', 'Connection: Upgrade',
@@ -171,7 +171,7 @@ class TestWebSocket(_TestBase):
'Sec-WebSocket-Protocol: ws', 'Sec-WebSocket-Protocol: ws',
'Sec-WebSocket-Location: ' 'Sec-WebSocket-Location: '
'ws://localhost:%s/echo?query_string\r\n\r\n8jKS\'y:G*Co,Wxa-' % self.port, 'ws://localhost:%s/echo?query_string\r\n\r\n8jKS\'y:G*Co,Wxa-' % self.port,
])) ])))
def test_empty_query_string(self): def test_empty_query_string(self):
# verify that a single trailing ? doesn't get nuked # verify that a single trailing ? doesn't get nuked
@@ -190,14 +190,14 @@ class TestWebSocket(_TestBase):
sock.sendall(six.b('\r\n'.join(connect) + '\r\n\r\n^n:ds[4U')) sock.sendall(six.b('\r\n'.join(connect) + '\r\n\r\n^n:ds[4U'))
result = sock.recv(1024) result = sock.recv(1024)
self.assertEqual(result, '\r\n'.join([ self.assertEqual(result, six.b('\r\n'.join([
'HTTP/1.1 101 WebSocket Protocol Handshake', 'HTTP/1.1 101 WebSocket Protocol Handshake',
'Upgrade: WebSocket', 'Upgrade: WebSocket',
'Connection: Upgrade', 'Connection: Upgrade',
'Sec-WebSocket-Origin: http://localhost:%s' % self.port, 'Sec-WebSocket-Origin: http://localhost:%s' % self.port,
'Sec-WebSocket-Protocol: ws', 'Sec-WebSocket-Protocol: ws',
'Sec-WebSocket-Location: ws://localhost:%s/echo?\r\n\r\n8jKS\'y:G*Co,Wxa-' % self.port, 'Sec-WebSocket-Location: ws://localhost:%s/echo?\r\n\r\n8jKS\'y:G*Co,Wxa-' % self.port,
])) ])))
def test_sending_messages_to_websocket_75(self): def test_sending_messages_to_websocket_75(self):
connect = [ connect = [
@@ -215,7 +215,7 @@ class TestWebSocket(_TestBase):
sock.recv(1024) sock.recv(1024)
sock.sendall(b'\x00hello\xFF') sock.sendall(b'\x00hello\xFF')
result = sock.recv(1024) result = sock.recv(1024)
self.assertEqual(result, '\x00hello\xff') self.assertEqual(result, b'\x00hello\xff')
sock.sendall(b'\x00start') sock.sendall(b'\x00start')
eventlet.sleep(0.001) eventlet.sleep(0.001)
sock.sendall(b' end\xff') sock.sendall(b' end\xff')