Cutlery/calibre
1
0
Fork 0
mirror of https://github.com/kovidgoyal/calibre.git synced 2025-07-09 03:04:10 -04:00
Code Issues Packages Projects Releases Wiki Activity

py3: port websocket tests

Browse Source
This commit is contained in:
Kovid Goyal 2019-04-15 13:58:50 +05:30
parent 440d174284
commit 51ae5ee86f
No known key found for this signature in database
GPG Key ID: 06BC317B515ACE7C
1 changed files with 9 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
src/calibre/srv/tests/web_sockets.py
View File

@ -16,7 +16,7 @@ from calibre.srv.web_socket import (
from calibre.utils.monotonic import monotonic
from calibre.utils.socket_inheritance import set_socket_inherit
from polyglot.builtins import range, unicode_type
from polyglot.binary import as_base64_bytes, as_base64_unicode
from polyglot.binary import as_base64_unicode
HANDSHAKE_STR = '''\
GET / HTTP/1.1\r
@ -35,7 +35,7 @@ class WSClient(object):
self.timeout = timeout
self.socket = socket.create_connection(('localhost', port), timeout)
set_socket_inherit(self.socket, False)
self.key = as_base64_bytes(os.urandom(8))
self.key = as_base64_unicode(os.urandom(8))
self.socket.sendall(HANDSHAKE_STR.format(self.key).encode('ascii'))
self.read_buf = deque()
self.read_upgrade_response()
@ -64,7 +64,7 @@ class WSClient(object):
if rl != b'HTTP/1.1 101 Switching Protocols\r\n':
raise ValueError('Server did not respond with correct switching protocols line')
headers = read_headers(partial(next, lines))
key = as_base64_unicode(sha1(self.key + GUID_STR.encode('ascii')).digest())
key = as_base64_unicode(sha1((self.key + GUID_STR).encode('ascii')).digest())
if headers.get('Sec-WebSocket-Accept') != key:
raise ValueError('Server did not respond with correct key in Sec-WebSocket-Accept: {} != {}'.format(
key, headers.get('Sec-WebSocket-Accept')))
@ -215,7 +215,7 @@ class WebSocketTest(BaseTest):
for q in (b'', b'\xfe' * 125, b'\xfe' * 126, b'\xfe' * 127, b'\xfe' * 128, b'\xfe' * 65535, b'\xfe' * 65536):
simple_test([q], [q])
for payload in ['', 'ping', b'\x00\xff\xfe\xfd\xfc\xfb\x00\xff', b"\xfe" * 125]:
for payload in [b'', b'ping', b'\x00\xff\xfe\xfd\xfc\xfb\x00\xff', b"\xfe" * 125]:
simple_test([(PING, payload)], [(PONG, payload)])
with server.silence_log:
@ -260,7 +260,8 @@ class WebSocketTest(BaseTest):
isf_test(frags, close_code=INCONSISTENT_DATA, send_close=False)
frags, q = [], b'\xce\xba\xe1\xbd\xb9\xcf\x83\xce\xbc\xce\xb5\xed\xa0\x80\x80\x65\x64\x69\x74\x65\x64'
for i, b in enumerate(q):
for i in range(len(q)):
b = q[i:i+1]
frags.append({'opcode':(TEXT if i == 0 else CONTINUATION), 'fin':1 if i == len(q)-1 else 0, 'payload':b})
isf_test(frags, close_code=INCONSISTENT_DATA, send_close=False, ignore_send_failures=True)
@ -289,9 +290,10 @@ class WebSocketTest(BaseTest):
simple_test([
{'opcode':TEXT, 'fin':0}, {'opcode':CONTINUATION, 'fin':0, 'payload':'x'}, {'opcode':CONTINUATION},], ['x'])
for q in (b'\xc2\xb5', b'\xce\xba\xe1\xbd\xb9\xcf\x83\xce\xbc\xce\xb5', "Hello-µ@ßöäüàá-UTF-8!!".encode('utf-8')):
for q in (b'\xc2\xb5', b'\xce\xba\xe1\xbd\xb9\xcf\x83\xce\xbc\xce\xb5', "Hello-µ@ßöäüàá-UTF-8!!".encode('utf-8')):
frags = []
for i, b in enumerate(q):
for i in range(len(q)):
b = q[i:i+1]
frags.append({'opcode':(TEXT if i == 0 else CONTINUATION), 'fin':1 if i == len(q)-1 else 0, 'payload':b})
simple_test(frags, [q.decode('utf-8')])