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

Ignore failures to send all packets when testing sending malformed data to server

Browse Source
This commit is contained in:
Kovid Goyal 2016-07-20 09:03:13 +05:30
parent 56506f7173
commit 2c783fd62d
1 changed files with 23 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -161,13 +161,17 @@ class WSTestServer(TestServer):
class WebSocketTest(BaseTest):
def simple_test(self, server, msgs, expected=(), close_code=NORMAL_CLOSE, send_close=True, close_reason=b'NORMAL CLOSE'):
def simple_test(self, server, msgs, expected=(), close_code=NORMAL_CLOSE, send_close=True, close_reason=b'NORMAL CLOSE', ignore_send_failures=False):
client = server.connect()
for msg in msgs:
try:
if isinstance(msg, dict):
client.write_frame(**msg)
else:
client.write_message(msg)
except Exception:
if not ignore_send_failures:
raise
expected_messages, expected_controls = [], []
for ex in expected:
@ -215,40 +219,44 @@ class WebSocketTest(BaseTest):
nc = struct.pack(b'!H', NORMAL_CLOSE)
with server.silence_log:
# It can happen that the server detects bad data and closes the
# connection before the client has finished sending all
# messages, so ignore failures to send packets.
isf_test = partial(simple_test, ignore_send_failures=True)
for rsv in xrange(1, 7):
simple_test([{'rsv':rsv, 'opcode':BINARY}], [], close_code=PROTOCOL_ERROR, send_close=False)
isf_test([{'rsv':rsv, 'opcode':BINARY}], [], close_code=PROTOCOL_ERROR, send_close=False)
for opcode in (3, 4, 5, 6, 7, 11, 12, 13, 14, 15):
simple_test([{'opcode':opcode}], [], close_code=PROTOCOL_ERROR, send_close=False)
isf_test([{'opcode':opcode}], [], close_code=PROTOCOL_ERROR, send_close=False)
for opcode in (PING, PONG):
simple_test([
isf_test([
{'opcode':opcode, 'payload':'f1', 'fin':0}, {'opcode':opcode, 'payload':'f2'}
], close_code=PROTOCOL_ERROR, send_close=False)
simple_test([(CLOSE, nc + b'x'*124)], send_close=False, close_code=PROTOCOL_ERROR)
isf_test([(CLOSE, nc + b'x'*124)], send_close=False, close_code=PROTOCOL_ERROR)
for fin in (0, 1):
simple_test([{'opcode':0, 'fin': fin, 'payload':b'non-continuation frame'}, 'some text'], close_code=PROTOCOL_ERROR, send_close=False)
isf_test([{'opcode':0, 'fin': fin, 'payload':b'non-continuation frame'}, 'some text'], close_code=PROTOCOL_ERROR, send_close=False)
simple_test([
isf_test([
{'opcode':TEXT, 'payload':fragments[0], 'fin':0}, {'opcode':CONTINUATION, 'payload':fragments[1]}, {'opcode':0, 'fin':0}
], [''.join(fragments)], close_code=PROTOCOL_ERROR, send_close=False)
simple_test([
isf_test([
{'opcode':TEXT, 'payload':fragments[0], 'fin':0}, {'opcode':TEXT, 'payload':fragments[1]},
], close_code=PROTOCOL_ERROR, send_close=False)
frags = []
for payload in (b'\xce\xba\xe1\xbd\xb9\xcf\x83\xce\xbc\xce\xb5', b'\xed\xa0\x80', b'\x80\x65\x64\x69\x74\x65\x64'):
frags.append({'opcode':(CONTINUATION if frags else TEXT), 'fin':1 if len(frags) == 2 else 0, 'payload':payload})
simple_test(frags, close_code=INCONSISTENT_DATA, send_close=False)
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):
frags.append({'opcode':(TEXT if i == 0 else CONTINUATION), 'fin':1 if i == len(q)-1 else 0, 'payload':b})
simple_test(frags, close_code=INCONSISTENT_DATA, send_close=False)
isf_test(frags, close_code=INCONSISTENT_DATA, send_close=False, ignore_send_failures=True)
for q in (b'\xce', b'\xce\xba\xe1'):
simple_test([{'opcode':TEXT, 'payload':q}], close_code=INCONSISTENT_DATA, send_close=False)
isf_test([{'opcode':TEXT, 'payload':q}], close_code=INCONSISTENT_DATA, send_close=False)
simple_test([
{'opcode':TEXT, 'payload':fragments[0], 'fin':0}, {'opcode':CONTINUATION, 'payload':fragments[1]}