diff --git a/Lib/wsgiref/validate.py b/Lib/wsgiref/validate.py --- a/Lib/wsgiref/validate.py +++ b/Lib/wsgiref/validate.py @@ -128,16 +128,22 @@ def assert_(cond, *args): raise AssertionError(*args) def check_string_type(value, title): if type (value) is str: return value raise AssertionError( "{0} must be of type str (got {1})".format(title, repr(value))) +def check_bytes_type(value, message): + if type(value) is bytes: + return value + raise AssertionError( + "{}, got {:r}".format(message, type(value).__name__)) + def validator(application): """ When applied between a WSGI server and a WSGI application, this middleware will check for WSGI compliancy on a number of levels. This middleware does not modify the request or response in any way, but will raise an AssertionError if anything seems off (except for a failure to close the application iterator, which @@ -191,31 +197,31 @@ def validator(application): class InputWrapper: def __init__(self, wsgi_input): self.input = wsgi_input def read(self, *args): assert_(len(args) == 1) v = self.input.read(*args) - assert_(type(v) is bytes) + check_bytes_type(v, "read() must return bytes") return v def readline(self, *args): - assert_(len(args) <= 1) + assert_(len(args) <= 1, "readline() takes zero or one arguments (%d given)" % len(args)) v = self.input.readline(*args) - assert_(type(v) is bytes) + check_bytes_type(v, "readline() must return bytes") return v def readlines(self, *args): - assert_(len(args) <= 1) + assert_(len(args) <= 1, "readlines() takes zero or one arguments (%d given)" % len(args)) lines = self.input.readlines(*args) - assert_(type(lines) is list) + assert_(type(lines) is list, "readlines() must return a list, got %r" % (lines,)) for line in lines: - assert_(type(line) is bytes) + check_bytes_type(line, "Each line returned by readlines() must be of type bytes") return lines def __iter__(self): while 1: line = self.readline() if not line: return yield line @@ -224,17 +230,17 @@ class InputWrapper: assert_(0, "input.close() must not be called") class ErrorWrapper: def __init__(self, wsgi_errors): self.errors = wsgi_errors def write(self, s): - assert_(type(s) is str) + check_string_type(s, "argument to write()") self.errors.write(s) def flush(self): self.errors.flush() def writelines(self, seq): for line in seq: self.write(line) @@ -243,17 +249,17 @@ class ErrorWrapper: assert_(0, "errors.close() must not be called") class WriteWrapper: def __init__(self, wsgi_writer): self.writer = wsgi_writer def __call__(self, s): - assert_(type(s) is bytes) + check_bytes_type(s, "write() must be called with bytes") self.writer(s) class PartialIteratorWrapper: def __init__(self, wsgi_iterator): self.iterator = wsgi_iterator def __iter__(self): @@ -390,17 +396,18 @@ def check_headers(headers): assert_(type(headers) is list, "Headers (%r) must be of type list: %r" % (headers, type(headers))) header_names = {} for item in headers: assert_(type(item) is tuple, "Individual headers (%r) must be of type tuple: %r" % (item, type(item))) - assert_(len(item) == 2) + assert_(len(item) == 2, + "Individual headers must only have two items: %r" % (item,)) name, value = item name = check_string_type(name, "Header name") value = check_string_type(value, "Header value") assert_(name.lower() != 'status', "The Status header cannot be used; it conflicts with CGI " "script, and HTTP status is not given through headers " "(value: %r)." % value) header_names[name.lower()] = None