<s> """<STR_LIT>""" <EOL> from urlparse import urlparse <EOL> from warnings import warn <EOL> from werkzeug . datastructures import Headers <EOL> from werkzeug . http import is_entity_header <EOL> from werkzeug . wsgi import FileWrapper <EOL> from werkzeug . _compat import string_types <EOL> class WSGIWarning ( Warning ) : <EOL> """<STR_LIT>""" <EOL> class HTTPWarning ( Warning ) : <EOL> """<STR_LIT>""" <EOL> def check_string ( context , obj , stacklevel = <NUM_LIT:3> ) : <EOL> if type ( obj ) is not str : <EOL> warn ( WSGIWarning ( '<STR_LIT>' % <EOL> ( context , obj . __class__ . __name__ ) ) ) <EOL> class InputStream ( object ) : <EOL> def __init__ ( self , stream ) : <EOL> self . _stream = stream <EOL> def read ( self , * args ) : <EOL> if len ( args ) == <NUM_LIT:0> : <EOL> warn ( WSGIWarning ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) , <EOL> stacklevel = <NUM_LIT:2> ) <EOL> elif len ( args ) != <NUM_LIT:1> : <EOL> warn ( WSGIWarning ( '<STR_LIT>' ) , <EOL> stacklevel = <NUM_LIT:2> ) <EOL> return self . _stream . read ( * args ) <EOL> def readline ( self , * args ) : <EOL> if len ( args ) == <NUM_LIT:0> : <EOL> warn ( WSGIWarning ( '<STR_LIT>' <EOL> '<STR_LIT>' ) , <EOL> stacklevel = <NUM_LIT:2> ) <EOL> elif len ( args ) == <NUM_LIT:1> : <EOL> warn ( WSGIWarning ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) , <EOL> stacklevel = <NUM_LIT:2> ) <EOL> else : <EOL> raise TypeError ( '<STR_LIT>' ) <EOL> return self . _stream . readline ( * args ) <EOL> def __iter__ ( self ) : <EOL> try : <EOL> return iter ( self . _stream ) <EOL> except TypeError : <EOL> warn ( WSGIWarning ( '<STR_LIT>' ) , stacklevel = <NUM_LIT:2> ) <EOL> return iter ( ( ) ) <EOL> def close ( self ) : <EOL> warn ( WSGIWarning ( '<STR_LIT>' ) , <EOL> stacklevel = <NUM_LIT:2> ) <EOL> self . _stream . close ( ) <EOL> class ErrorStream ( object ) : <EOL> def __init__ ( self , stream ) : <EOL> self . _stream = stream <EOL> def write ( self , s ) : <EOL> check_string ( '<STR_LIT>' , s ) <EOL> self . _stream . write ( s ) <EOL> def flush ( self ) : <EOL> self . _stream . flush ( ) <EOL> def writelines ( self , seq ) : <EOL> for line in seq : <EOL> self . write ( seq ) <EOL> def close ( self ) : <EOL> warn ( WSGIWarning ( '<STR_LIT>' ) , <EOL> stacklevel = <NUM_LIT:2> ) <EOL> self . _stream . close ( ) <EOL> class GuardedWrite ( object ) : <EOL> def __init__ ( self , write , chunks ) : <EOL> self . _write = write <EOL> self . _chunks = chunks <EOL> def __call__ ( self , s ) : <EOL> check_string ( '<STR_LIT>' , s ) <EOL> self . _write . write ( s ) <EOL> self . _chunks . append ( len ( s ) ) <EOL> class GuardedIterator ( object ) : <EOL> def __init__ ( self , iterator , headers_set , chunks ) : <EOL> self . _iterator = iterator <EOL> self . _next = iter ( iterator ) . next <EOL> self . closed = False <EOL> self . headers_set = headers_set <EOL> self . chunks = chunks <EOL> def __iter__ ( self ) : <EOL> return self <EOL> def next ( self ) : <EOL> if self . closed : <EOL> warn ( WSGIWarning ( '<STR_LIT>' ) , <EOL> stacklevel = <NUM_LIT:2> ) <EOL> rv = self . _next ( ) <EOL> if not self . headers_set : <EOL> warn ( WSGIWarning ( '<STR_LIT>' <EOL> '<STR_LIT>' ) , stacklevel = <NUM_LIT:2> ) <EOL> check_string ( '<STR_LIT>' , rv ) <EOL> self . chunks . append ( len ( rv ) ) <EOL> return rv <EOL> def close ( self ) : <EOL> self . closed = True <EOL> if hasattr ( self . _iterator , '<STR_LIT>' ) : <EOL> self . _iterator . close ( ) <EOL> if self . headers_set : <EOL> status_code , headers = self . headers_set <EOL> bytes_sent = sum ( self . chunks ) <EOL> content_length = headers . get ( '<STR_LIT>' , type = int ) <EOL> if status_code == <NUM_LIT> : <EOL> for key , value in headers : <EOL> key = key . lower ( ) <EOL> if key not in ( '<STR_LIT>' , '<STR_LIT>' ) and is_entity_header ( key ) : <EOL> warn ( HTTPWarning ( '<STR_LIT>' <EOL> '<STR_LIT>' % key ) ) <EOL> if bytes_sent : <EOL> warn ( HTTPWarning ( '<STR_LIT>' ) ) <EOL> elif <NUM_LIT:100> <= status_code < <NUM_LIT:200> or status_code == <NUM_LIT> : <EOL> if content_length != <NUM_LIT:0> : <EOL> warn ( HTTPWarning ( '<STR_LIT>' <EOL> '<STR_LIT>' ) % status_code ) <EOL> if bytes_sent : <EOL> warn ( HTTPWarning ( '<STR_LIT>' % <EOL> status_code ) ) <EOL> elif content_length is not None and content_length != bytes_sent : <EOL> warn ( WSGIWarning ( '<STR_LIT>' <EOL> '<STR_LIT>' ) ) <EOL> def __del__ ( self ) : <EOL> if not self . closed : <EOL> try : <EOL> warn ( WSGIWarning ( '<STR_LIT>' <EOL> '<STR_LIT>' ) ) <EOL> except Exception : <EOL> pass <EOL> class LintMiddleware ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , app ) : <EOL> self . app = app <EOL> def check_environ ( self , environ ) : <EOL> if type ( environ ) is not dict : <EOL> warn ( WSGIWarning ( '<STR_LIT>' ) , <EOL> stacklevel = <NUM_LIT:4> ) <EOL> for key in ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ) : <EOL> if key not in environ : <EOL> warn ( WSGIWarning ( '<STR_LIT>' <EOL> % key ) , stacklevel = <NUM_LIT:3> ) <EOL> if environ [ '<STR_LIT>' ] != ( <NUM_LIT:1> , <NUM_LIT:0> ) : <EOL> warn ( WSGIWarning ( '<STR_LIT>' ) , <EOL> stacklevel = <NUM_LIT:3> ) <EOL> script_name = environ . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> if script_name and script_name [ : <NUM_LIT:1> ] != '<STR_LIT:/>' : <EOL> warn ( WSGIWarning ( '<STR_LIT>' <EOL> % script_name ) , stacklevel = <NUM_LIT:3> ) <EOL> path_info = environ . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> if path_info [ : <NUM_LIT:1> ] != '<STR_LIT:/>' : <EOL> warn ( WSGIWarning ( '<STR_LIT>' <EOL> % path_info ) , stacklevel = <NUM_LIT:3> ) <EOL> def check_start_response ( self , status , headers , exc_info ) : <EOL> check_string ( '<STR_LIT:status>' , status ) <EOL> status_code = status . split ( None , <NUM_LIT:1> ) [ <NUM_LIT:0> ] <EOL> if len ( status_code ) != <NUM_LIT:3> or not status_code . isdigit ( ) : <EOL> warn ( WSGIWarning ( '<STR_LIT>' ) , stacklevel = <NUM_LIT:3> ) <EOL> if len ( status ) < <NUM_LIT:4> or status [ <NUM_LIT:3> ] != '<STR_LIT:U+0020>' : <EOL> warn ( WSGIWarning ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) , stacklevel = <NUM_LIT:3> ) <EOL> status_code = int ( status_code ) <EOL> if status_code < <NUM_LIT:100> : <EOL> warn ( WSGIWarning ( '<STR_LIT>' ) , stacklevel = <NUM_LIT:3> ) <EOL> if type ( headers ) is not list : <EOL> warn ( WSGIWarning ( '<STR_LIT>' ) , stacklevel = <NUM_LIT:3> ) <EOL> for item in headers : <EOL> if type ( item ) is not tuple or len ( item ) != <NUM_LIT:2> : <EOL> warn ( WSGIWarning ( '<STR_LIT>' ) , <EOL> stacklevel = <NUM_LIT:3> ) <EOL> name , value = item <EOL> if type ( name ) is not str or type ( value ) is not str : <EOL> warn ( WSGIWarning ( '<STR_LIT>' ) , <EOL> stacklevel = <NUM_LIT:3> ) <EOL> if name . lower ( ) == '<STR_LIT:status>' : <EOL> warn ( WSGIWarning ( '<STR_LIT>' <EOL> '<STR_LIT>' ) , <EOL> stacklevel = <NUM_LIT:3> ) <EOL> if exc_info is not None and not isinstance ( exc_info , tuple ) : <EOL> warn ( WSGIWarning ( '<STR_LIT>' ) , stacklevel = <NUM_LIT:3> ) <EOL> headers = Headers ( headers ) <EOL> self . check_headers ( headers ) <EOL> return status_code , headers <EOL> def check_headers ( self , headers ) : <EOL> etag = headers . get ( '<STR_LIT>' ) <EOL> if etag is not None : <EOL> if etag . startswith ( '<STR_LIT>' ) : <EOL> etag = etag [ <NUM_LIT:2> : ] <EOL> if not ( etag [ : <NUM_LIT:1> ] == etag [ - <NUM_LIT:1> : ] == '<STR_LIT:">' ) : <EOL> warn ( HTTPWarning ( '<STR_LIT>' ) , stacklevel = <NUM_LIT:4> ) <EOL> location = headers . get ( '<STR_LIT:location>' ) <EOL> if location is not None : <EOL> if not urlparse ( location ) . netloc : <EOL> warn ( HTTPWarning ( '<STR_LIT>' ) , <EOL> stacklevel = <NUM_LIT:4> ) <EOL> def check_iterator ( self , app_iter ) : <EOL> if isinstance ( app_iter , string_types ) : <EOL> warn ( WSGIWarning ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) , stacklevel = <NUM_LIT:3> ) <EOL> def __call__ ( self , * args , ** kwargs ) : <EOL> if len ( args ) != <NUM_LIT:2> : <EOL> warn ( WSGIWarning ( '<STR_LIT>' ) , stacklevel = <NUM_LIT:2> ) <EOL> if kwargs : <EOL> warn ( WSGIWarning ( '<STR_LIT>' ) , <EOL> stacklevel = <NUM_LIT:2> ) <EOL> environ , start_response = args <EOL> self . check_environ ( environ ) <EOL> environ [ '<STR_LIT>' ] = InputStream ( environ [ '<STR_LIT>' ] ) <EOL> environ [ '<STR_LIT>' ] = ErrorStream ( environ [ '<STR_LIT>' ] ) <EOL> environ [ '<STR_LIT>' ] = FileWrapper <EOL> headers_set = [ ] <EOL> chunks = [ ] <EOL> def checking_start_response ( * args , ** kwargs ) : <EOL> if len ( args ) not in ( <NUM_LIT:2> , <NUM_LIT:3> ) : <EOL> warn ( WSGIWarning ( '<STR_LIT>' <EOL> '<STR_LIT>' % len ( args ) , stacklevel = <NUM_LIT:2> ) ) <EOL> if kwargs : <EOL> warn ( WSGIWarning ( '<STR_LIT>' ) ) <EOL> status , headers = args [ : <NUM_LIT:2> ] <EOL> if len ( args ) == <NUM_LIT:3> : <EOL> exc_info = args [ <NUM_LIT:2> ] <EOL> else : <EOL> exc_info = None <EOL> headers_set [ : ] = self . check_start_response ( status , headers , <EOL> exc_info ) <EOL> return GuardedWrite ( start_response ( status , headers , exc_info ) , <EOL> chunks ) <EOL> app_iter = self . app ( environ , checking_start_response ) <EOL> self . check_iterator ( app_iter ) <EOL> return GuardedIterator ( app_iter , headers_set , chunks ) </s>
