Cutlery/calibre
1
0
Fork 0
mirror of https://github.com/kovidgoyal/calibre.git synced 2025-06-23 15:30:45 -04:00
Code Issues Packages Projects Releases Wiki Activity

Ignore conn closed/reset errors for all invalid data web socket tests

Browse Source
This commit is contained in:
Kovid Goyal 2016-07-10 20:05:49 +05:30
parent c78e78a435
commit 6cd282bced
2 changed files with 19 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
src/calibre/srv/tests/http.py
View File

@ -189,7 +189,7 @@ class TestHTTP(BaseTest):
self.ae('', r.read()) self.ae('', r.read())
server.change_handler(lambda data:data.path[0] + data.read().decode('ascii')) server.change_handler(lambda data:data.path[0] + data.read().decode('ascii'))
conn = server.connect() conn = server.connect(timeout=1)
# Test simple GET # Test simple GET
conn.request('GET', '/test/') conn.request('GET', '/test/')
@ -291,7 +291,7 @@ class TestHTTP(BaseTest):
def handler(conn): def handler(conn):
return conn.generate_static_output('test', lambda : ''.join(conn.path)) return conn.generate_static_output('test', lambda : ''.join(conn.path))
with NamedTemporaryFile(suffix='test.epub') as f, open(P('localization/locales.zip'), 'rb') as lf, \ with NamedTemporaryFile(suffix='test.epub') as f, open(P('localization/locales.zip'), 'rb') as lf, \
TestServer(handler, timeout=0.2, compress_min_size=0) as server: TestServer(handler, timeout=1, compress_min_size=0) as server:
fdata = string.ascii_letters * 100 fdata = string.ascii_letters * 100
f.write(fdata), f.seek(0) f.write(fdata), f.seek(0)

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

@ -214,29 +214,6 @@ class WebSocketTest(BaseTest):
fragments = 'Hello-µ@ßöä üàá-UTF-8!!'.split() fragments = 'Hello-µ@ßöä üàá-UTF-8!!'.split()
nc = struct.pack(b'!H', NORMAL_CLOSE) nc = struct.pack(b'!H', NORMAL_CLOSE)
with server.silence_log:
for rsv in xrange(1, 7):
simple_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)
for opcode in (PING, PONG):
simple_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)
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)
simple_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([
{'opcode':TEXT, 'payload':fragments[0], 'fin':0}, {'opcode':TEXT, 'payload':fragments[1]},
], close_code=PROTOCOL_ERROR, send_close=False)
def ic_test(*args, **kwargs): def ic_test(*args, **kwargs):
try: try:
simple_test(*args, **kwargs) simple_test(*args, **kwargs)
@ -247,6 +224,29 @@ class WebSocketTest(BaseTest):
# writing all data. # writing all data.
raise raise
with server.silence_log:
for rsv in xrange(1, 7):
ic_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):
ic_test([{'opcode':opcode}], [], close_code=PROTOCOL_ERROR, send_close=False)
for opcode in (PING, PONG):
ic_test([
{'opcode':opcode, 'payload':'f1', 'fin':0}, {'opcode':opcode, 'payload':'f2'}
], close_code=PROTOCOL_ERROR, send_close=False)
ic_test([(CLOSE, nc + b'x'*124)], send_close=False, close_code=PROTOCOL_ERROR)
for fin in (0, 1):
ic_test([{'opcode':0, 'fin': fin, 'payload':b'non-continuation frame'}, 'some text'], close_code=PROTOCOL_ERROR, send_close=False)
ic_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)
ic_test([
{'opcode':TEXT, 'payload':fragments[0], 'fin':0}, {'opcode':TEXT, 'payload':fragments[1]},
], close_code=PROTOCOL_ERROR, send_close=False)
frags = [] 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'): 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}) frags.append({'opcode':(CONTINUATION if frags else TEXT), 'fin':1 if len(frags) == 2 else 0, 'payload':payload})