test.py (20448B)
1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\carbon\common\lib\werkzeug\test.py 2 import sys 3 import urllib 4 import urlparse 5 import mimetypes 6 from time import time 7 from random import random 8 from itertools import chain 9 from tempfile import TemporaryFile 10 from cStringIO import StringIO 11 from cookielib import CookieJar 12 from urllib2 import Request as U2Request 13 from werkzeug._internal import _empty_stream, _get_environ 14 from werkzeug.wrappers import BaseRequest 15 from werkzeug.urls import url_encode, url_fix, iri_to_uri 16 from werkzeug.wsgi import get_host, get_current_url 17 from werkzeug.datastructures import FileMultiDict, MultiDict, CombinedMultiDict, Headers, FileStorage 18 19 def stream_encode_multipart(values, use_tempfile = True, threshold = 1024 * 500, boundary = None, charset = 'utf-8'): 20 if boundary is None: 21 boundary = '---------------WerkzeugFormPart_%s%s' % (time(), random()) 22 _closure = [StringIO(), 0, False] 23 if use_tempfile: 24 25 def write(string): 26 stream, total_length, on_disk = _closure 27 if on_disk: 28 stream.write(string) 29 else: 30 length = len(string) 31 if length + _closure[1] <= threshold: 32 stream.write(string) 33 else: 34 new_stream = TemporaryFile('wb+') 35 new_stream.write(stream.getvalue()) 36 new_stream.write(string) 37 _closure[0] = new_stream 38 _closure[2] = True 39 _closure[1] = total_length + length 40 41 else: 42 write = _closure[0].write 43 if not isinstance(values, MultiDict): 44 values = MultiDict(values) 45 for key, values in values.iterlists(): 46 for value in values: 47 write('--%s\r\nContent-Disposition: form-data; name="%s"' % (boundary, key)) 48 reader = getattr(value, 'read', None) 49 if reader is not None: 50 filename = getattr(value, 'filename', getattr(value, 'name', None)) 51 content_type = getattr(value, 'content_type', None) 52 if content_type is None: 53 content_type = filename and mimetypes.guess_type(filename)[0] or 'application/octet-stream' 54 if filename is not None: 55 write('; filename="%s"\r\n' % filename) 56 else: 57 write('\r\n') 58 write('Content-Type: %s\r\n\r\n' % content_type) 59 while 1: 60 chunk = reader(16384) 61 if not chunk: 62 break 63 write(chunk) 64 65 else: 66 if isinstance(value, unicode): 67 value = value.encode(charset) 68 write('\r\n\r\n' + value) 69 write('\r\n') 70 71 write('--%s--\r\n' % boundary) 72 length = int(_closure[0].tell()) 73 _closure[0].seek(0) 74 return (_closure[0], length, boundary) 75 76 77 def encode_multipart(values, boundary = None, charset = 'utf-8'): 78 stream, length, boundary = stream_encode_multipart(values, use_tempfile=False, boundary=boundary, charset=charset) 79 return (boundary, stream.read()) 80 81 82 def File(fd, filename = None, mimetype = None): 83 from warnings import warn 84 warn(DeprecationWarning('werkzeug.test.File is deprecated, use the EnvironBuilder or FileStorage instead')) 85 return FileStorage(fd, filename=filename, content_type=mimetype) 86 87 88 class _TestCookieHeaders(object): 89 90 def __init__(self, headers): 91 self.headers = headers 92 93 def getheaders(self, name): 94 headers = [] 95 name = name.lower() 96 for k, v in self.headers: 97 if k.lower() == name: 98 headers.append(v) 99 100 return headers 101 102 103 class _TestCookieResponse(object): 104 105 def __init__(self, headers): 106 self.headers = _TestCookieHeaders(headers) 107 108 def info(self): 109 return self.headers 110 111 112 class _TestCookieJar(CookieJar): 113 114 def inject_wsgi(self, environ): 115 cvals = [] 116 for cookie in self: 117 cvals.append('%s=%s' % (cookie.name, cookie.value)) 118 119 if cvals: 120 environ['HTTP_COOKIE'] = ', '.join(cvals) 121 122 def extract_wsgi(self, environ, headers): 123 self.extract_cookies(_TestCookieResponse(headers), U2Request(get_current_url(environ))) 124 125 126 def _iter_data(data): 127 if isinstance(data, MultiDict): 128 for key, values in data.iterlists(): 129 for value in values: 130 yield (key, value) 131 132 else: 133 for key, values in data.iteritems(): 134 if isinstance(values, list): 135 for value in values: 136 yield (key, value) 137 138 else: 139 yield (key, values) 140 141 142 class EnvironBuilder(object): 143 server_protocol = 'HTTP/1.1' 144 wsgi_version = (1, 0) 145 request_class = BaseRequest 146 147 def __init__(self, path = '/', base_url = None, query_string = None, method = 'GET', input_stream = None, content_type = None, content_length = None, errors_stream = None, multithread = False, multiprocess = False, run_once = False, headers = None, data = None, environ_base = None, environ_overrides = None, charset = 'utf-8'): 148 if query_string is None and '?' in path: 149 path, query_string = path.split('?', 1) 150 self.charset = charset 151 if isinstance(path, unicode): 152 path = iri_to_uri(path, charset) 153 self.path = path 154 if base_url is not None: 155 if isinstance(base_url, unicode): 156 base_url = iri_to_uri(base_url, charset) 157 else: 158 base_url = url_fix(base_url, charset) 159 self.base_url = base_url 160 if isinstance(query_string, basestring): 161 self.query_string = query_string 162 else: 163 if query_string is None: 164 query_string = MultiDict() 165 elif not isinstance(query_string, MultiDict): 166 query_string = MultiDict(query_string) 167 self.args = query_string 168 self.method = method 169 if headers is None: 170 headers = Headers() 171 elif not isinstance(headers, Headers): 172 headers = Headers(headers) 173 self.headers = headers 174 self.content_type = content_type 175 if errors_stream is None: 176 errors_stream = sys.stderr 177 self.errors_stream = errors_stream 178 self.multithread = multithread 179 self.multiprocess = multiprocess 180 self.run_once = run_once 181 self.environ_base = environ_base 182 self.environ_overrides = environ_overrides 183 self.input_stream = input_stream 184 self.content_length = content_length 185 self.closed = False 186 if data: 187 if input_stream is not None: 188 raise TypeError("can't provide input stream and data") 189 if isinstance(data, basestring): 190 self.input_stream = StringIO(data) 191 if self.content_length is None: 192 self.content_length = len(data) 193 else: 194 for key, value in _iter_data(data): 195 if isinstance(value, (tuple, dict)) or hasattr(value, 'read'): 196 self._add_file_from_data(key, value) 197 else: 198 self.form.setlistdefault(key).append(value) 199 200 def _add_file_from_data(self, key, value): 201 if isinstance(value, tuple): 202 self.files.add_file(key, *value) 203 elif isinstance(value, dict): 204 from warnings import warn 205 warn(DeprecationWarning("it's no longer possible to pass dicts as `data`. Use tuples or FileStorage objects instead"), stacklevel=2) 206 args = v 207 value = dict(value) 208 mimetype = value.pop('mimetype', None) 209 if mimetype is not None: 210 value['content_type'] = mimetype 211 self.files.add_file(key, **value) 212 else: 213 self.files.add_file(key, value) 214 215 def _get_base_url(self): 216 return urlparse.urlunsplit((self.url_scheme, 217 self.host, 218 self.script_root, 219 '', 220 '')).rstrip('/') + '/' 221 222 def _set_base_url(self, value): 223 if value is None: 224 scheme = 'http' 225 netloc = 'localhost' 226 scheme = 'http' 227 script_root = '' 228 else: 229 scheme, netloc, script_root, qs, anchor = urlparse.urlsplit(value) 230 if qs or anchor: 231 raise ValueError('base url must not contain a query string or fragment') 232 self.script_root = script_root.rstrip('/') 233 self.host = netloc 234 self.url_scheme = scheme 235 236 base_url = property(_get_base_url, _set_base_url, doc='\n The base URL is a URL that is used to extract the WSGI\n URL scheme, host (server name + server port) and the\n script root (`SCRIPT_NAME`).') 237 del _get_base_url 238 del _set_base_url 239 240 def _get_content_type(self): 241 ct = self.headers.get('Content-Type') 242 if ct is None and not self._input_stream: 243 if self.method in ('POST', 'PUT'): 244 if self._files: 245 return 'multipart/form-data' 246 return 'application/x-www-form-urlencoded' 247 return 248 return ct 249 250 def _set_content_type(self, value): 251 if value is None: 252 self.headers.pop('Content-Type', None) 253 else: 254 self.headers['Content-Type'] = value 255 256 content_type = property(_get_content_type, _set_content_type, doc='\n The content type for the request. Reflected from and to the\n :attr:`headers`. Do not set if you set :attr:`files` or\n :attr:`form` for auto detection.') 257 del _get_content_type 258 del _set_content_type 259 260 def _get_content_length(self): 261 return self.headers.get('Content-Length', type=int) 262 263 def _set_content_length(self, value): 264 if value is None: 265 self.headers.pop('Content-Length', None) 266 else: 267 self.headers['Content-Length'] = str(value) 268 269 content_length = property(_get_content_length, _set_content_length, doc='\n The content length as integer. Reflected from and to the\n :attr:`headers`. Do not set if you set :attr:`files` or\n :attr:`form` for auto detection.') 270 del _get_content_length 271 del _set_content_length 272 273 def form_property(name, storage, doc): 274 key = '_' + name 275 276 def getter(self): 277 if self._input_stream is not None: 278 raise AttributeError('an input stream is defined') 279 rv = getattr(self, key) 280 if rv is None: 281 rv = storage() 282 setattr(self, key, rv) 283 return rv 284 285 def setter(self, value): 286 self._input_stream = None 287 setattr(self, key, value) 288 289 return property(getter, setter, doc) 290 291 form = form_property('form', MultiDict, doc='\n A :class:`MultiDict` of form values.') 292 files = form_property('files', FileMultiDict, doc='\n A :class:`FileMultiDict` of uploaded files. You can use the\n :meth:`~FileMultiDict.add_file` method to add new files to the\n dict.') 293 del form_property 294 295 def _get_input_stream(self): 296 return self._input_stream 297 298 def _set_input_stream(self, value): 299 self._input_stream = value 300 self._form = self._files = None 301 302 input_stream = property(_get_input_stream, _set_input_stream, doc='\n An optional input stream. If you set this it will clear\n :attr:`form` and :attr:`files`.') 303 del _get_input_stream 304 del _set_input_stream 305 306 def _get_query_string(self): 307 if self._query_string is None: 308 if self._args is not None: 309 return url_encode(self._args, charset=self.charset) 310 return '' 311 return self._query_string 312 313 def _set_query_string(self, value): 314 self._query_string = value 315 self._args = None 316 317 query_string = property(_get_query_string, _set_query_string, doc='\n The query string. If you set this to a string :attr:`args` will\n no longer be available.') 318 del _get_query_string 319 del _set_query_string 320 321 def _get_args(self): 322 if self._query_string is not None: 323 raise AttributeError('a query string is defined') 324 if self._args is None: 325 self._args = MultiDict() 326 return self._args 327 328 def _set_args(self, value): 329 self._query_string = None 330 self._args = value 331 332 args = property(_get_args, _set_args, doc='\n The URL arguments as :class:`MultiDict`.') 333 del _get_args 334 del _set_args 335 336 @property 337 def server_name(self): 338 return self.host.split(':', 1)[0] 339 340 @property 341 def server_port(self): 342 pieces = self.host.split(':', 1) 343 if len(pieces) == 2 and pieces[1].isdigit(): 344 return int(pieces[1]) 345 if self.url_scheme == 'https': 346 return 443 347 return 80 348 349 def __del__(self): 350 self.close() 351 352 def close(self): 353 if self.closed: 354 return 355 try: 356 files = self.files.itervalues() 357 except AttributeError: 358 files = () 359 360 for f in files: 361 try: 362 f.close() 363 except Exception as e: 364 pass 365 366 self.closed = True 367 368 def get_environ(self): 369 input_stream = self.input_stream 370 content_length = self.content_length 371 content_type = self.content_type 372 if input_stream is not None: 373 start_pos = input_stream.tell() 374 input_stream.seek(0, 2) 375 end_pos = input_stream.tell() 376 input_stream.seek(start_pos) 377 content_length = end_pos - start_pos 378 elif content_type == 'multipart/form-data': 379 values = CombinedMultiDict([self.form, self.files]) 380 input_stream, content_length, boundary = stream_encode_multipart(values, charset=self.charset) 381 content_type += '; boundary="%s"' % boundary 382 elif content_type == 'application/x-www-form-urlencoded': 383 values = url_encode(self.form, charset=self.charset) 384 content_length = len(values) 385 input_stream = StringIO(values) 386 else: 387 input_stream = _empty_stream 388 result = {} 389 if self.environ_base: 390 result.update(self.environ_base) 391 392 def _path_encode(x): 393 if isinstance(x, unicode): 394 x = x.encode(self.charset) 395 return urllib.unquote(x) 396 397 result.update({'REQUEST_METHOD': self.method, 398 'SCRIPT_NAME': _path_encode(self.script_root), 399 'PATH_INFO': _path_encode(self.path), 400 'QUERY_STRING': self.query_string, 401 'SERVER_NAME': self.server_name, 402 'SERVER_PORT': str(self.server_port), 403 'HTTP_HOST': self.host, 404 'SERVER_PROTOCOL': self.server_protocol, 405 'CONTENT_TYPE': content_type or '', 406 'CONTENT_LENGTH': str(content_length or '0'), 407 'wsgi.version': self.wsgi_version, 408 'wsgi.url_scheme': self.url_scheme, 409 'wsgi.input': input_stream, 410 'wsgi.errors': self.errors_stream, 411 'wsgi.multithread': self.multithread, 412 'wsgi.multiprocess': self.multiprocess, 413 'wsgi.run_once': self.run_once}) 414 for key, value in self.headers.to_list(self.charset): 415 result['HTTP_%s' % key.upper().replace('-', '_')] = value 416 417 if self.environ_overrides: 418 result.update(self.environ_overrides) 419 return result 420 421 def get_request(self, cls = None): 422 if cls is None: 423 cls = self.request_class 424 return cls(self.get_environ()) 425 426 427 class ClientRedirectError(Exception): 428 pass 429 430 431 class Client(object): 432 433 def __init__(self, application, response_wrapper = None, use_cookies = True): 434 self.application = application 435 if response_wrapper is None: 436 response_wrapper = lambda a, s, h: (a, s, h) 437 self.response_wrapper = response_wrapper 438 if use_cookies: 439 self.cookie_jar = _TestCookieJar() 440 else: 441 self.cookie_jar = None 442 self.redirect_client = None 443 444 def open(self, *args, **kwargs): 445 as_tuple = kwargs.pop('as_tuple', False) 446 buffered = kwargs.pop('buffered', False) 447 follow_redirects = kwargs.pop('follow_redirects', False) 448 environ = None 449 if not kwargs and len(args) == 1: 450 if isinstance(args[0], EnvironBuilder): 451 environ = args[0].get_environ() 452 elif isinstance(args[0], dict): 453 environ = args[0] 454 if environ is None: 455 builder = EnvironBuilder(*args, **kwargs) 456 try: 457 environ = builder.get_environ() 458 finally: 459 builder.close() 460 461 if self.cookie_jar is not None: 462 self.cookie_jar.inject_wsgi(environ) 463 rv = run_wsgi_app(self.application, environ, buffered=buffered) 464 if self.cookie_jar is not None: 465 self.cookie_jar.extract_wsgi(environ, rv[2]) 466 redirect_chain = [] 467 status_code = int(rv[1].split(None, 1)[0]) 468 while status_code in (301, 302, 303, 305, 307) and follow_redirects: 469 if not self.redirect_client: 470 self.redirect_client = Client(self.application) 471 self.redirect_client.cookie_jar = self.cookie_jar 472 redirect = dict(rv[2])['Location'] 473 scheme, netloc, script_root, qs, anchor = urlparse.urlsplit(redirect) 474 base_url = urlparse.urlunsplit((scheme, 475 netloc, 476 '', 477 '', 478 '')).rstrip('/') + '/' 479 host = get_host(create_environ('/', base_url, query_string=qs)).split(':', 1)[0] 480 if get_host(environ).split(':', 1)[0] != host: 481 raise RuntimeError('%r does not support redirect to external targets' % self.__class__) 482 redirect_chain.append((redirect, status_code)) 483 redirect_kwargs = {} 484 redirect_kwargs.update({'path': script_root, 485 'base_url': base_url, 486 'query_string': qs, 487 'as_tuple': True, 488 'buffered': buffered, 489 'follow_redirects': False}) 490 environ, rv = self.redirect_client.open(**redirect_kwargs) 491 status_code = int(rv[1].split(None, 1)[0]) 492 if redirect_chain[-1] in redirect_chain[0:-1]: 493 raise ClientRedirectError('loop detected') 494 495 response = self.response_wrapper(*rv) 496 if as_tuple: 497 return (environ, response) 498 return response 499 500 def get(self, *args, **kw): 501 kw['method'] = 'GET' 502 return self.open(*args, **kw) 503 504 def post(self, *args, **kw): 505 kw['method'] = 'POST' 506 return self.open(*args, **kw) 507 508 def head(self, *args, **kw): 509 kw['method'] = 'HEAD' 510 return self.open(*args, **kw) 511 512 def put(self, *args, **kw): 513 kw['method'] = 'PUT' 514 return self.open(*args, **kw) 515 516 def delete(self, *args, **kw): 517 kw['method'] = 'DELETE' 518 return self.open(*args, **kw) 519 520 def __repr__(self): 521 return '<%s %r>' % (self.__class__.__name__, self.application) 522 523 524 def create_environ(*args, **kwargs): 525 builder = EnvironBuilder(*args, **kwargs) 526 try: 527 return builder.get_environ() 528 finally: 529 builder.close() 530 531 532 def run_wsgi_app(app, environ, buffered = False): 533 environ = _get_environ(environ) 534 response = [] 535 buffer = [] 536 537 def start_response(status, headers, exc_info = None): 538 if exc_info is not None: 539 raise exc_info[0], exc_info[1], exc_info[2] 540 response[:] = [status, headers] 541 return buffer.append 542 543 app_iter = app(environ, start_response) 544 if buffered: 545 close_func = getattr(app_iter, 'close', None) 546 try: 547 app_iter = list(app_iter) 548 finally: 549 if close_func is not None: 550 close_func() 551 552 else: 553 while not response: 554 buffer.append(app_iter.next()) 555 556 if buffer: 557 close_func = getattr(app_iter, 'close', None) 558 app_iter = chain(buffer, app_iter) 559 if close_func is not None: 560 app_iter = ClosingIterator(app_iter, close_func) 561 return (app_iter, response[0], response[1]) 562 563 564 from werkzeug.wsgi import ClosingIterator