When applied between a WSGI server and a WSGI application, this middleware will check for WSGI compliance 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 cl
(application)
| 134 | "{0} must be of type str (got {1})".format(title, repr(value))) |
| 135 | |
| 136 | def validator(application): |
| 137 | |
| 138 | """ |
| 139 | When applied between a WSGI server and a WSGI application, this |
| 140 | middleware will check for WSGI compliance on a number of levels. |
| 141 | This middleware does not modify the request or response in any |
| 142 | way, but will raise an AssertionError if anything seems off |
| 143 | (except for a failure to close the application iterator, which |
| 144 | will be printed to stderr -- there's no way to raise an exception |
| 145 | at that point). |
| 146 | """ |
| 147 | |
| 148 | def lint_app(*args, **kw): |
| 149 | assert_(len(args) == 2, "Two arguments required") |
| 150 | assert_(not kw, "No keyword arguments allowed") |
| 151 | environ, start_response = args |
| 152 | |
| 153 | check_environ(environ) |
| 154 | |
| 155 | # We use this to check if the application returns without |
| 156 | # calling start_response: |
| 157 | start_response_started = [] |
| 158 | |
| 159 | def start_response_wrapper(*args, **kw): |
| 160 | assert_(len(args) == 2 or len(args) == 3, ( |
| 161 | "Invalid number of arguments: %s" % (args,))) |
| 162 | assert_(not kw, "No keyword arguments allowed") |
| 163 | status = args[0] |
| 164 | headers = args[1] |
| 165 | if len(args) == 3: |
| 166 | exc_info = args[2] |
| 167 | else: |
| 168 | exc_info = None |
| 169 | |
| 170 | check_status(status) |
| 171 | check_headers(headers) |
| 172 | check_content_type(status, headers) |
| 173 | check_exc_info(exc_info) |
| 174 | |
| 175 | start_response_started.append(None) |
| 176 | return WriteWrapper(start_response(*args)) |
| 177 | |
| 178 | environ['wsgi.input'] = InputWrapper(environ['wsgi.input']) |
| 179 | environ['wsgi.errors'] = ErrorWrapper(environ['wsgi.errors']) |
| 180 | |
| 181 | iterator = application(environ, start_response_wrapper) |
| 182 | assert_(iterator is not None and iterator != False, |
| 183 | "The application must return an iterator, if only an empty list") |
| 184 | |
| 185 | check_iterator(iterator) |
| 186 | |
| 187 | return IteratorWrapper(iterator, start_response_started) |
| 188 | |
| 189 | return lint_app |
| 190 | |
| 191 | class InputWrapper: |
| 192 |
no outgoing calls
searching dependent graphs…