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

Turn on bufferring for the test suite

Browse Source 
Allows for more verbose server logging
This commit is contained in:
Kovid Goyal 2020-12-08 09:23:32 +05:30
parent beadc4384f
commit 1ee3c0fdd6
No known key found for this signature in database
GPG Key ID: 06BC317B515ACE7C
4 changed files with 46 additions and 43 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
src/calibre/srv/tests/base.py
View File

@ -6,7 +6,7 @@ __license__ = 'GPL v3'
__copyright__ = '2011, Kovid Goyal <kovid@kovidgoyal.net>'
__docformat__ = 'restructuredtext en'
import unittest, time, shutil, gc, tempfile, atexit, os
import unittest, time, shutil, gc, tempfile, atexit, os, sys
from io import BytesIO
from functools import partial
from threading import Thread
@ -86,10 +86,11 @@ class TestServer(Thread):
create_http_handler(handler),
opts=Options(**kwargs),
plugins=plugins,
log=ServerLog(level=ServerLog.WARN),
log=ServerLog(level=ServerLog.DEBUG),
)
self.log = self.loop.log
self.silence_log = self.log
# allow unittest's bufferring to work
self.log.outputs[0].stream = sys.stdout
specialize(self)
def setup_defaults(self, kwargs):
@ -145,8 +146,10 @@ class LibraryServer(TestServer):
create_http_handler(self.handler.dispatch),
opts=opts,
plugins=plugins,
log=ServerLog(level=ServerLog.WARN),
log=ServerLog(level=ServerLog.DEBUG),
)
# allow unittest's bufferring to work
self.loop.log.outputs[0].stream = sys.stdout
self.handler.set_log(self.loop.log)
specialize(self)

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

@ -222,8 +222,7 @@ class WebSocketTest(BaseTest):
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:
simple_test([(PING, 'a'*126)], close_code=PROTOCOL_ERROR, send_close=False)
simple_test([(PING, 'a'*126)], close_code=PROTOCOL_ERROR, send_close=False)
for payload in (b'', b'pong'):
simple_test([(PONG, payload)], [])
@ -231,46 +230,45 @@ class WebSocketTest(BaseTest):
fragments = 'Hello-µ@ßöä üàá-UTF-8!!'.split()
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 range(1, 7):
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):
isf_test([{'opcode':opcode}], [], close_code=PROTOCOL_ERROR, send_close=False)
for opcode in (PING, PONG):
isf_test([
{'opcode':opcode, 'payload':'f1', 'fin':0}, {'opcode':opcode, 'payload':'f2'}
], close_code=PROTOCOL_ERROR, send_close=False)
isf_test([(CLOSE, nc + b'x'*124)], send_close=False, close_code=PROTOCOL_ERROR)
for fin in (0, 1):
isf_test([{'opcode':0, 'fin': fin, 'payload':b'non-continuation frame'}, 'some text'], close_code=PROTOCOL_ERROR, send_close=False)
# 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 range(1, 7):
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):
isf_test([{'opcode':opcode}], [], close_code=PROTOCOL_ERROR, send_close=False)
for opcode in (PING, PONG):
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)
isf_test([
{'opcode':TEXT, 'payload':fragments[0], 'fin':0}, {'opcode':TEXT, 'payload':fragments[1]},
{'opcode':opcode, 'payload':'f1', 'fin':0}, {'opcode':opcode, 'payload':'f2'}
], close_code=PROTOCOL_ERROR, send_close=False)
isf_test([(CLOSE, nc + b'x'*124)], send_close=False, close_code=PROTOCOL_ERROR)
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})
isf_test(frags, close_code=INCONSISTENT_DATA, send_close=False)
for fin in (0, 1):
isf_test([{'opcode':0, 'fin': fin, 'payload':b'non-continuation frame'}, 'some text'], close_code=PROTOCOL_ERROR, 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 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)
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)
for q in (b'\xce', b'\xce\xba\xe1'):
isf_test([{'opcode':TEXT, 'payload':q}], close_code=INCONSISTENT_DATA, send_close=False)
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})
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 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)
for q in (b'\xce', b'\xce\xba\xe1'):
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]}

4
src/calibre/utils/logging.py
View File

@ -26,7 +26,6 @@ class Stream(object):
stream = io.StringIO()
self.stream = stream
self.encoding = getattr(self.stream, 'encoding', None) or 'utf-8'
self._prints = partial(prints, file=self.stream)
def write(self, text):
self._prints(text, end='')
@ -37,6 +36,9 @@ class Stream(object):
def prints(self, level, *args, **kwargs):
self._prints(*args, **kwargs)
def _prints(self, *args, **kwargs):
prints(*args, file=self.stream, **kwargs)
class ANSIStream(Stream):

4
src/calibre/utils/run_tests.py
View File

@ -142,10 +142,10 @@ def run_tests(find_tests, verbosity=4):
run_cli(tests, verbosity)
def run_cli(suite, verbosity=4):
def run_cli(suite, verbosity=4, buffer=True):
r = unittest.TextTestRunner
r.resultclass = unittest.TextTestResult if verbosity < 2 else TestResult
init_env()
result = r(verbosity=verbosity).run(suite)
result = r(verbosity=verbosity, buffer=buffer).run(suite)
if not result.wasSuccessful():
raise SystemExit(1)