codekingpro/portable-devtools
114k
1from __future__ import annotations2 3import dataclasses4import mimetypes5import sys6import typing as t7from collections import defaultdict8from datetime import datetime9from io import BytesIO10from itertools import chain11from random import random12from tempfile import TemporaryFile13from time import time14from urllib.parse import unquote15from urllib.parse import urlsplit16from urllib.parse import urlunsplit17 18from ._internal import _get_environ19from ._internal import _wsgi_decoding_dance20from ._internal import _wsgi_encoding_dance21from .datastructures import Authorization22from .datastructures import CallbackDict23from .datastructures import CombinedMultiDict24from .datastructures import EnvironHeaders25from .datastructures import FileMultiDict26from .datastructures import Headers27from .datastructures import MultiDict28from .http import dump_cookie29from .http import dump_options_header30from .http import parse_cookie31from .http import parse_date32from .http import parse_options_header33from .sansio.multipart import Data34from .sansio.multipart import Epilogue35from .sansio.multipart import Field36from .sansio.multipart import File37from .sansio.multipart import MultipartEncoder38from .sansio.multipart import Preamble39from .urls import _urlencode40from .urls import iri_to_uri41from .utils import cached_property42from .utils import get_content_type43from .wrappers.request import Request44from .wrappers.response import Response45from .wsgi import ClosingIterator46from .wsgi import get_current_url47 48if t.TYPE_CHECKING:49 import typing_extensions as te50 from _typeshed.wsgi import WSGIApplication51 from _typeshed.wsgi import WSGIEnvironment52 53 54def stream_encode_multipart(55 data: t.Mapping[str, t.Any],56 use_tempfile: bool = True,57 threshold: int = 1024 * 500,58 boundary: str | None = None,59) -> tuple[t.IO[bytes], int, str]:60 """Encode a dict of values (either strings or file descriptors or61 :class:`FileStorage` objects.) into a multipart encoded string stored62 in a file descriptor.63 64 .. versionchanged:: 3.065 The ``charset`` parameter was removed.66 """67 if boundary is None:68 boundary = f"---------------WerkzeugFormPart_{time()}{random()}"69 70 stream: t.IO[bytes] = BytesIO()71 total_length = 072 on_disk = False73 write_binary: t.Callable[[bytes], int]74 75 if use_tempfile:76 77 def write_binary(s: bytes) -> int:78 nonlocal stream, total_length, on_disk79 80 if on_disk:81 return stream.write(s)82 else:83 length = len(s)84 85 if length + total_length <= threshold:86 stream.write(s)87 else:88 new_stream = t.cast(t.IO[bytes], TemporaryFile("wb+"))89 new_stream.write(stream.getvalue()) # type: ignore90 new_stream.write(s)91 stream = new_stream92 on_disk = True93 94 total_length += length95 return length96 97 else:98 write_binary = stream.write99 100 encoder = MultipartEncoder(boundary.encode())101 write_binary(encoder.send_event(Preamble(data=b"")))102 for key, value in _iter_data(data):103 reader = getattr(value, "read", None)104 if reader is not None:105 filename = getattr(value, "filename", getattr(value, "name", None))106 content_type = getattr(value, "content_type", None)107 if content_type is None:108 content_type = (109 filename110 and mimetypes.guess_type(filename)[0]111 or "application/octet-stream"112 )113 headers = value.headers114 headers.update([("Content-Type", content_type)])115 if filename is None:116 write_binary(encoder.send_event(Field(name=key, headers=headers)))117 else:118 write_binary(119 encoder.send_event(120 File(name=key, filename=filename, headers=headers)121 )122 )123 while True:124 chunk = reader(16384)125 126 if not chunk:127 write_binary(encoder.send_event(Data(data=chunk, more_data=False)))128 break129 130 write_binary(encoder.send_event(Data(data=chunk, more_data=True)))131 else:132 if not isinstance(value, str):133 value = str(value)134 write_binary(encoder.send_event(Field(name=key, headers=Headers())))135 write_binary(encoder.send_event(Data(data=value.encode(), more_data=False)))136 137 write_binary(encoder.send_event(Epilogue(data=b"")))138 139 length = stream.tell()140 stream.seek(0)141 return stream, length, boundary142 143 144def encode_multipart(145 values: t.Mapping[str, t.Any], boundary: str | None = None146) -> tuple[str, bytes]:147 """Like `stream_encode_multipart` but returns a tuple in the form148 (``boundary``, ``data``) where data is bytes.149 150 .. versionchanged:: 3.0151 The ``charset`` parameter was removed.152 """153 stream, length, boundary = stream_encode_multipart(154 values, use_tempfile=False, boundary=boundary155 )156 return boundary, stream.read()157 158 159def _iter_data(data: t.Mapping[str, t.Any]) -> t.Iterator[tuple[str, t.Any]]:160 """Iterate over a mapping that might have a list of values, yielding161 all key, value pairs. Almost like iter_multi_items but only allows162 lists, not tuples, of values so tuples can be used for files.163 """164 if isinstance(data, MultiDict):165 yield from data.items(multi=True)166 else:167 for key, value in data.items():168 if isinstance(value, list):169 for v in value:170 yield key, v171 else:172 yield key, value173 174 175_TAnyMultiDict = t.TypeVar("_TAnyMultiDict", bound="MultiDict[t.Any, t.Any]")176 177 178class EnvironBuilder:179 """This class can be used to conveniently create a WSGI environment180 for testing purposes. It can be used to quickly create WSGI environments181 or request objects from arbitrary data.182 183 The signature of this class is also used in some other places as of184 Werkzeug 0.5 (:func:`create_environ`, :meth:`Response.from_values`,185 :meth:`Client.open`). Because of this most of the functionality is186 available through the constructor alone.187 188 Files and regular form data can be manipulated independently of each189 other with the :attr:`form` and :attr:`files` attributes, but are190 passed with the same argument to the constructor: `data`.191 192 `data` can be any of these values:193 194 - a `str` or `bytes` object: The object is converted into an195 :attr:`input_stream`, the :attr:`content_length` is set and you have to196 provide a :attr:`content_type`.197 - a `dict` or :class:`MultiDict`: The keys have to be strings. The values198 have to be either any of the following objects, or a list of any of the199 following objects:200 201 - a :class:`file`-like object: These are converted into202 :class:`FileStorage` objects automatically.203 - a `tuple`: The :meth:`~FileMultiDict.add_file` method is called204 with the key and the unpacked `tuple` items as positional205 arguments.206 - a `str`: The string is set as form data for the associated key.207 - a file-like object: The object content is loaded in memory and then208 handled like a regular `str` or a `bytes`.209 210 :param path: the path of the request. In the WSGI environment this will211 end up as `PATH_INFO`. If the `query_string` is not defined212 and there is a question mark in the `path` everything after213 it is used as query string.214 :param base_url: the base URL is a URL that is used to extract the WSGI215 URL scheme, host (server name + server port) and the216 script root (`SCRIPT_NAME`).217 :param query_string: an optional string or dict with URL parameters.218 :param method: the HTTP method to use, defaults to `GET`.219 :param input_stream: an optional input stream. Do not specify this and220 `data`. As soon as an input stream is set you can't221 modify :attr:`args` and :attr:`files` unless you222 set the :attr:`input_stream` to `None` again.223 :param content_type: The content type for the request. As of 0.5 you224 don't have to provide this when specifying files225 and form data via `data`.226 :param content_length: The content length for the request. You don't227 have to specify this when providing data via228 `data`.229 :param errors_stream: an optional error stream that is used for230 `wsgi.errors`. Defaults to :data:`stderr`.231 :param multithread: controls `wsgi.multithread`. Defaults to `False`.232 :param multiprocess: controls `wsgi.multiprocess`. Defaults to `False`.233 :param run_once: controls `wsgi.run_once`. Defaults to `False`.234 :param headers: an optional list or :class:`Headers` object of headers.235 :param data: a string or dict of form data or a file-object.236 See explanation above.237 :param json: An object to be serialized and assigned to ``data``.238 Defaults the content type to ``"application/json"``.239 Serialized with the function assigned to :attr:`json_dumps`.240 :param environ_base: an optional dict of environment defaults.241 :param environ_overrides: an optional dict of environment overrides.242 :param auth: An authorization object to use for the243 ``Authorization`` header value. A ``(username, password)`` tuple244 is a shortcut for ``Basic`` authorization.245 246 .. versionchanged:: 3.0247 The ``charset`` parameter was removed.248 249 .. versionchanged:: 2.1250 ``CONTENT_TYPE`` and ``CONTENT_LENGTH`` are not duplicated as251 header keys in the environ.252 253 .. versionchanged:: 2.0254 ``REQUEST_URI`` and ``RAW_URI`` is the full raw URI including255 the query string, not only the path.256 257 .. versionchanged:: 2.0258 The default :attr:`request_class` is ``Request`` instead of259 ``BaseRequest``.260 261 .. versionadded:: 2.0262 Added the ``auth`` parameter.263 264 .. versionadded:: 0.15265 The ``json`` param and :meth:`json_dumps` method.266 267 .. versionadded:: 0.15268 The environ has keys ``REQUEST_URI`` and ``RAW_URI`` containing269 the path before percent-decoding. This is not part of the WSGI270 PEP, but many WSGI servers include it.271 272 .. versionchanged:: 0.6273 ``path`` and ``base_url`` can now be unicode strings that are274 encoded with :func:`iri_to_uri`.275 """276 277 #: the server protocol to use. defaults to HTTP/1.1278 server_protocol = "HTTP/1.1"279 280 #: the wsgi version to use. defaults to (1, 0)281 wsgi_version = (1, 0)282 283 #: The default request class used by :meth:`get_request`.284 request_class = Request285 286 import json287 288 #: The serialization function used when ``json`` is passed.289 json_dumps = staticmethod(json.dumps)290 del json291 292 _args: MultiDict[str, str] | None293 _query_string: str | None294 _input_stream: t.IO[bytes] | None295 _form: MultiDict[str, str] | None296 _files: FileMultiDict | None297 298 def __init__(299 self,300 path: str = "/",301 base_url: str | None = None,302 query_string: t.Mapping[str, str] | str | None = None,303 method: str = "GET",304 input_stream: t.IO[bytes] | None = None,305 content_type: str | None = None,306 content_length: int | None = None,307 errors_stream: t.IO[str] | None = None,308 multithread: bool = False,309 multiprocess: bool = False,310 run_once: bool = False,311 headers: Headers | t.Iterable[tuple[str, str]] | None = None,312 data: None | (t.IO[bytes] | str | bytes | t.Mapping[str, t.Any]) = None,313 environ_base: t.Mapping[str, t.Any] | None = None,314 environ_overrides: t.Mapping[str, t.Any] | None = None,315 mimetype: str | None = None,316 json: t.Mapping[str, t.Any] | None = None,317 auth: Authorization | tuple[str, str] | None = None,318 ) -> None:319 if query_string is not None and "?" in path:320 raise ValueError("Query string is defined in the path and as an argument")321 request_uri = urlsplit(path)322 if query_string is None and "?" in path:323 query_string = request_uri.query324 325 self.path = iri_to_uri(request_uri.path)326 self.request_uri = path327 if base_url is not None:328 base_url = iri_to_uri(base_url)329 self.base_url = base_url # type: ignore330 if isinstance(query_string, str):331 self.query_string = query_string332 else:333 if query_string is None:334 query_string = MultiDict()335 elif not isinstance(query_string, MultiDict):336 query_string = MultiDict(query_string)337 self.args = query_string338 self.method = method339 if headers is None:340 headers = Headers()341 elif not isinstance(headers, Headers):342 headers = Headers(headers)343 self.headers = headers344 if content_type is not None:345 self.content_type = content_type346 if errors_stream is None:347 errors_stream = sys.stderr348 self.errors_stream = errors_stream349 self.multithread = multithread350 self.multiprocess = multiprocess351 self.run_once = run_once352 self.environ_base = environ_base353 self.environ_overrides = environ_overrides354 self.input_stream = input_stream355 self.content_length = content_length356 self.closed = False357 358 if auth is not None:359 if isinstance(auth, tuple):360 auth = Authorization(361 "basic", {"username": auth[0], "password": auth[1]}362 )363 364 self.headers.set("Authorization", auth.to_header())365 366 if json is not None:367 if data is not None:368 raise TypeError("can't provide both json and data")369 370 data = self.json_dumps(json)371 372 if self.content_type is None:373 self.content_type = "application/json"374 375 if data:376 if input_stream is not None:377 raise TypeError("can't provide input stream and data")378 if hasattr(data, "read"):379 data = data.read()380 if isinstance(data, str):381 data = data.encode()382 if isinstance(data, bytes):383 self.input_stream = BytesIO(data)384 if self.content_length is None:385 self.content_length = len(data)386 else:387 for key, value in _iter_data(data):388 if isinstance(value, (tuple, dict)) or hasattr(value, "read"):389 self._add_file_from_data(key, value)390 else:391 self.form.setlistdefault(key).append(value)392 393 if mimetype is not None:394 self.mimetype = mimetype395 396 @classmethod397 def from_environ(cls, environ: WSGIEnvironment, **kwargs: t.Any) -> EnvironBuilder:398 """Turn an environ dict back into a builder. Any extra kwargs399 override the args extracted from the environ.400 401 .. versionchanged:: 2.0402 Path and query values are passed through the WSGI decoding403 dance to avoid double encoding.404 405 .. versionadded:: 0.15406 """407 headers = Headers(EnvironHeaders(environ))408 out = {409 "path": _wsgi_decoding_dance(environ["PATH_INFO"]),410 "base_url": cls._make_base_url(411 environ["wsgi.url_scheme"],412 headers.pop("Host"),413 _wsgi_decoding_dance(environ["SCRIPT_NAME"]),414 ),415 "query_string": _wsgi_decoding_dance(environ["QUERY_STRING"]),416 "method": environ["REQUEST_METHOD"],417 "input_stream": environ["wsgi.input"],418 "content_type": headers.pop("Content-Type", None),419 "content_length": headers.pop("Content-Length", None),420 "errors_stream": environ["wsgi.errors"],421 "multithread": environ["wsgi.multithread"],422 "multiprocess": environ["wsgi.multiprocess"],423 "run_once": environ["wsgi.run_once"],424 "headers": headers,425 }426 out.update(kwargs)427 return cls(**out)428 429 def _add_file_from_data(430 self,431 key: str,432 value: (t.IO[bytes] | tuple[t.IO[bytes], str] | tuple[t.IO[bytes], str, str]),433 ) -> None:434 """Called in the EnvironBuilder to add files from the data dict."""435 if isinstance(value, tuple):436 self.files.add_file(key, *value)437 else:438 self.files.add_file(key, value)439 440 @staticmethod441 def _make_base_url(scheme: str, host: str, script_root: str) -> str:442 return urlunsplit((scheme, host, script_root, "", "")).rstrip("/") + "/"443 444 @property445 def base_url(self) -> str:446 """The base URL is used to extract the URL scheme, host name,447 port, and root path.448 """449 return self._make_base_url(self.url_scheme, self.host, self.script_root)450 451 @base_url.setter452 def base_url(self, value: str | None) -> None:453 if value is None:454 scheme = "http"455 netloc = "localhost"456 script_root = ""457 else:458 scheme, netloc, script_root, qs, anchor = urlsplit(value)459 if qs or anchor:460 raise ValueError("base url must not contain a query string or fragment")461 self.script_root = script_root.rstrip("/")462 self.host = netloc463 self.url_scheme = scheme464 465 @property466 def content_type(self) -> str | None:467 """The content type for the request. Reflected from and to468 the :attr:`headers`. Do not set if you set :attr:`files` or469 :attr:`form` for auto detection.470 """471 ct = self.headers.get("Content-Type")472 if ct is None and not self._input_stream:473 if self._files:474 return "multipart/form-data"475 if self._form:476 return "application/x-www-form-urlencoded"477 return None478 return ct479 480 @content_type.setter481 def content_type(self, value: str | None) -> None:482 if value is None:483 self.headers.pop("Content-Type", None)484 else:485 self.headers["Content-Type"] = value486 487 @property488 def mimetype(self) -> str | None:489 """The mimetype (content type without charset etc.)490 491 .. versionadded:: 0.14492 """493 ct = self.content_type494 return ct.split(";")[0].strip() if ct else None495 496 @mimetype.setter497 def mimetype(self, value: str) -> None:498 self.content_type = get_content_type(value, "utf-8")499 500 @property501 def mimetype_params(self) -> t.Mapping[str, str]:502 """The mimetype parameters as dict. For example if the503 content type is ``text/html; charset=utf-8`` the params would be504 ``{'charset': 'utf-8'}``.505 506 .. versionadded:: 0.14507 """508 509 def on_update(d: CallbackDict[str, str]) -> None:510 self.headers["Content-Type"] = dump_options_header(self.mimetype, d)511 512 d = parse_options_header(self.headers.get("content-type", ""))[1]513 return CallbackDict(d, on_update)514 515 @property516 def content_length(self) -> int | None:517 """The content length as integer. Reflected from and to the518 :attr:`headers`. Do not set if you set :attr:`files` or519 :attr:`form` for auto detection.520 """521 return self.headers.get("Content-Length", type=int)522 523 @content_length.setter524 def content_length(self, value: int | None) -> None:525 if value is None:526 self.headers.pop("Content-Length", None)527 else:528 self.headers["Content-Length"] = str(value)529 530 def _get_form(self, name: str, storage: type[_TAnyMultiDict]) -> _TAnyMultiDict:531 """Common behavior for getting the :attr:`form` and532 :attr:`files` properties.533 534 :param name: Name of the internal cached attribute.535 :param storage: Storage class used for the data.536 """537 if self.input_stream is not None:538 raise AttributeError("an input stream is defined")539 540 rv = getattr(self, name)541 542 if rv is None:543 rv = storage()544 setattr(self, name, rv)545 546 return rv # type: ignore547 548 def _set_form(self, name: str, value: MultiDict[str, t.Any]) -> None:549 """Common behavior for setting the :attr:`form` and550 :attr:`files` properties.551 552 :param name: Name of the internal cached attribute.553 :param value: Value to assign to the attribute.554 """555 self._input_stream = None556 setattr(self, name, value)557 558 @property559 def form(self) -> MultiDict[str, str]:560 """A :class:`MultiDict` of form values."""561 return self._get_form("_form", MultiDict)562 563 @form.setter564 def form(self, value: MultiDict[str, str]) -> None:565 self._set_form("_form", value)566 567 @property568 def files(self) -> FileMultiDict:569 """A :class:`FileMultiDict` of uploaded files. Use570 :meth:`~FileMultiDict.add_file` to add new files.571 """572 return self._get_form("_files", FileMultiDict)573 574 @files.setter575 def files(self, value: FileMultiDict) -> None:576 self._set_form("_files", value)577 578 @property579 def input_stream(self) -> t.IO[bytes] | None:580 """An optional input stream. This is mutually exclusive with581 setting :attr:`form` and :attr:`files`, setting it will clear582 those. Do not provide this if the method is not ``POST`` or583 another method that has a body.584 """585 return self._input_stream586 587 @input_stream.setter588 def input_stream(self, value: t.IO[bytes] | None) -> None:589 self._input_stream = value590 self._form = None591 self._files = None592 593 @property594 def query_string(self) -> str:595 """The query string. If you set this to a string596 :attr:`args` will no longer be available.597 """598 if self._query_string is None:599 if self._args is not None:600 return _urlencode(self._args)601 return ""602 return self._query_string603 604 @query_string.setter605 def query_string(self, value: str | None) -> None:606 self._query_string = value607 self._args = None608 609 @property610 def args(self) -> MultiDict[str, str]:611 """The URL arguments as :class:`MultiDict`."""612 if self._query_string is not None:613 raise AttributeError("a query string is defined")614 if self._args is None:615 self._args = MultiDict()616 return self._args617 618 @args.setter619 def args(self, value: MultiDict[str, str] | None) -> None:620 self._query_string = None621 self._args = value622 623 @property624 def server_name(self) -> str:625 """The server name (read-only, use :attr:`host` to set)"""626 return self.host.split(":", 1)[0]627 628 @property629 def server_port(self) -> int:630 """The server port as integer (read-only, use :attr:`host` to set)"""631 pieces = self.host.split(":", 1)632 633 if len(pieces) == 2:634 try:635 return int(pieces[1])636 except ValueError:637 pass638 639 if self.url_scheme == "https":640 return 443641 return 80642 643 def __del__(self) -> None:644 try:645 self.close()646 except Exception:647 pass648 649 def close(self) -> None:650 """Closes all files. If you put real :class:`file` objects into the651 :attr:`files` dict you can call this method to automatically close652 them all in one go.653 """654 if self.closed:655 return656 try:657 files = self.files.values()658 except AttributeError:659 files = () # type: ignore660 for f in files:661 try:662 f.close()663 except Exception:664 pass665 self.closed = True666 667 def get_environ(self) -> WSGIEnvironment:668 """Return the built environ.669 670 .. versionchanged:: 0.15671 The content type and length headers are set based on672 input stream detection. Previously this only set the WSGI673 keys.674 """675 input_stream = self.input_stream676 content_length = self.content_length677 678 mimetype = self.mimetype679 content_type = self.content_type680 681 if input_stream is not None:682 start_pos = input_stream.tell()683 input_stream.seek(0, 2)684 end_pos = input_stream.tell()685 input_stream.seek(start_pos)686 content_length = end_pos - start_pos687 elif mimetype == "multipart/form-data":688 input_stream, content_length, boundary = stream_encode_multipart(689 CombinedMultiDict([self.form, self.files])690 )691 content_type = f'{mimetype}; boundary="{boundary}"'692 elif mimetype == "application/x-www-form-urlencoded":693 form_encoded = _urlencode(self.form).encode("ascii")694 content_length = len(form_encoded)695 input_stream = BytesIO(form_encoded)696 else:697 input_stream = BytesIO()698 699 result: WSGIEnvironment = {}700 if self.environ_base:701 result.update(self.environ_base)702 703 def _path_encode(x: str) -> str:704 return _wsgi_encoding_dance(unquote(x))705 706 raw_uri = _wsgi_encoding_dance(self.request_uri)707 result.update(708 {709 "REQUEST_METHOD": self.method,710 "SCRIPT_NAME": _path_encode(self.script_root),711 "PATH_INFO": _path_encode(self.path),712 "QUERY_STRING": _wsgi_encoding_dance(self.query_string),713 # Non-standard, added by mod_wsgi, uWSGI714 "REQUEST_URI": raw_uri,715 # Non-standard, added by gunicorn716 "RAW_URI": raw_uri,717 "SERVER_NAME": self.server_name,718 "SERVER_PORT": str(self.server_port),719 "HTTP_HOST": self.host,720 "SERVER_PROTOCOL": self.server_protocol,721 "wsgi.version": self.wsgi_version,722 "wsgi.url_scheme": self.url_scheme,723 "wsgi.input": input_stream,724 "wsgi.errors": self.errors_stream,725 "wsgi.multithread": self.multithread,726 "wsgi.multiprocess": self.multiprocess,727 "wsgi.run_once": self.run_once,728 }729 )730 731 headers = self.headers.copy()732 # Don't send these as headers, they're part of the environ.733 headers.remove("Content-Type")734 headers.remove("Content-Length")735 736 if content_type is not None:737 result["CONTENT_TYPE"] = content_type738 739 if content_length is not None:740 result["CONTENT_LENGTH"] = str(content_length)741 742 combined_headers = defaultdict(list)743 744 for key, value in headers.to_wsgi_list():745 combined_headers[f"HTTP_{key.upper().replace('-', '_')}"].append(value)746 747 for key, values in combined_headers.items():748 result[key] = ", ".join(values)749 750 if self.environ_overrides:751 result.update(self.environ_overrides)752 753 return result754 755 def get_request(self, cls: type[Request] | None = None) -> Request:756 """Returns a request with the data. If the request class is not757 specified :attr:`request_class` is used.758 759 :param cls: The request wrapper to use.760 """761 if cls is None:762 cls = self.request_class763 764 return cls(self.get_environ())765 766 767class ClientRedirectError(Exception):768 """If a redirect loop is detected when using follow_redirects=True with769 the :cls:`Client`, then this exception is raised.770 """771 772 773class Client:774 """Simulate sending requests to a WSGI application without running a WSGI or HTTP775 server.776 777 :param application: The WSGI application to make requests to.778 :param response_wrapper: A :class:`.Response` class to wrap response data with.779 Defaults to :class:`.TestResponse`. If it's not a subclass of ``TestResponse``,780 one will be created.781 :param use_cookies: Persist cookies from ``Set-Cookie`` response headers to the782 ``Cookie`` header in subsequent requests. Domain and path matching is supported,783 but other cookie parameters are ignored.784 :param allow_subdomain_redirects: Allow requests to follow redirects to subdomains.785 Enable this if the application handles subdomains and redirects between them.786 787 .. versionchanged:: 2.3788 Simplify cookie implementation, support domain and path matching.789 790 .. versionchanged:: 2.1791 All data is available as properties on the returned response object. The792 response cannot be returned as a tuple.793 794 .. versionchanged:: 2.0795 ``response_wrapper`` is always a subclass of :class:``TestResponse``.796 797 .. versionchanged:: 0.5798 Added the ``use_cookies`` parameter.799 """800 801 def __init__(802 self,803 application: WSGIApplication,804 response_wrapper: type[Response] | None = None,805 use_cookies: bool = True,806 allow_subdomain_redirects: bool = False,807 ) -> None:808 self.application = application809 810 if response_wrapper in {None, Response}:811 response_wrapper = TestResponse812 elif response_wrapper is not None and not issubclass(813 response_wrapper, TestResponse814 ):815 response_wrapper = type(816 "WrapperTestResponse",817 (TestResponse, response_wrapper),818 {},819 )820 821 self.response_wrapper = t.cast(t.Type["TestResponse"], response_wrapper)822 823 if use_cookies:824 self._cookies: dict[tuple[str, str, str], Cookie] | None = {}825 else:826 self._cookies = None827 828 self.allow_subdomain_redirects = allow_subdomain_redirects829 830 def get_cookie(831 self, key: str, domain: str = "localhost", path: str = "/"832 ) -> Cookie | None:833 """Return a :class:`.Cookie` if it exists. Cookies are uniquely identified by834 ``(domain, path, key)``.835 836 :param key: The decoded form of the key for the cookie.837 :param domain: The domain the cookie was set for.838 :param path: The path the cookie was set for.839 840 .. versionadded:: 2.3841 """842 if self._cookies is None:843 raise TypeError(844 "Cookies are disabled. Create a client with 'use_cookies=True'."845 )846 847 return self._cookies.get((domain, path, key))848 849 def set_cookie(850 self,851 key: str,852 value: str = "",853 *,854 domain: str = "localhost",855 origin_only: bool = True,856 path: str = "/",857 **kwargs: t.Any,858 ) -> None:859 """Set a cookie to be sent in subsequent requests.860 861 This is a convenience to skip making a test request to a route that would set862 the cookie. To test the cookie, make a test request to a route that uses the863 cookie value.864 865 The client uses ``domain``, ``origin_only``, and ``path`` to determine which866 cookies to send with a request. It does not use other cookie parameters that867 browsers use, since they're not applicable in tests.868 869 :param key: The key part of the cookie.870 :param value: The value part of the cookie.871 :param domain: Send this cookie with requests that match this domain. If872 ``origin_only`` is true, it must be an exact match, otherwise it may be a873 suffix match.874 :param origin_only: Whether the domain must be an exact match to the request.875 :param path: Send this cookie with requests that match this path either exactly876 or as a prefix.877 :param kwargs: Passed to :func:`.dump_cookie`.878 879 .. versionchanged:: 3.0880 The parameter ``server_name`` is removed. The first parameter is881 ``key``. Use the ``domain`` and ``origin_only`` parameters instead.882 883 .. versionchanged:: 2.3884 The ``origin_only`` parameter was added.885 886 .. versionchanged:: 2.3887 The ``domain`` parameter defaults to ``localhost``.888 """889 if self._cookies is None:890 raise TypeError(891 "Cookies are disabled. Create a client with 'use_cookies=True'."892 )893 894 cookie = Cookie._from_response_header(895 domain, "/", dump_cookie(key, value, domain=domain, path=path, **kwargs)896 )897 cookie.origin_only = origin_only898 899 if cookie._should_delete:900 self._cookies.pop(cookie._storage_key, None)901 else:902 self._cookies[cookie._storage_key] = cookie903 904 def delete_cookie(905 self,906 key: str,907 *,908 domain: str = "localhost",909 path: str = "/",910 ) -> None:911 """Delete a cookie if it exists. Cookies are uniquely identified by912 ``(domain, path, key)``.913 914 :param key: The decoded form of the key for the cookie.915 :param domain: The domain the cookie was set for.916 :param path: The path the cookie was set for.917 918 .. versionchanged:: 3.0919 The ``server_name`` parameter is removed. The first parameter is920 ``key``. Use the ``domain`` parameter instead.921 922 .. versionchanged:: 3.0923 The ``secure``, ``httponly`` and ``samesite`` parameters are removed.924 925 .. versionchanged:: 2.3926 The ``domain`` parameter defaults to ``localhost``.927 """928 if self._cookies is None:929 raise TypeError(930 "Cookies are disabled. Create a client with 'use_cookies=True'."931 )932 933 self._cookies.pop((domain, path, key), None)934 935 def _add_cookies_to_wsgi(self, environ: WSGIEnvironment) -> None:936 """If cookies are enabled, set the ``Cookie`` header in the environ to the937 cookies that are applicable to the request host and path.938 939 :meta private:940 941 .. versionadded:: 2.3942 """943 if self._cookies is None:944 return945 946 url = urlsplit(get_current_url(environ))947 server_name = url.hostname or "localhost"948 value = "; ".join(949 c._to_request_header()950 for c in self._cookies.values()951 if c._matches_request(server_name, url.path)952 )953 954 if value:955 environ["HTTP_COOKIE"] = value956 else:957 environ.pop("HTTP_COOKIE", None)958 959 def _update_cookies_from_response(960 self, server_name: str, path: str, headers: list[str]961 ) -> None:962 """If cookies are enabled, update the stored cookies from any ``Set-Cookie``963 headers in the response.964 965 :meta private:966 967 .. versionadded:: 2.3968 """969 if self._cookies is None:970 return971 972 for header in headers:973 cookie = Cookie._from_response_header(server_name, path, header)974 975 if cookie._should_delete:976 self._cookies.pop(cookie._storage_key, None)977 else:978 self._cookies[cookie._storage_key] = cookie979 980 def run_wsgi_app(981 self, environ: WSGIEnvironment, buffered: bool = False982 ) -> tuple[t.Iterable[bytes], str, Headers]:983 """Runs the wrapped WSGI app with the given environment.984 985 :meta private:986 """987 self._add_cookies_to_wsgi(environ)988 rv = run_wsgi_app(self.application, environ, buffered=buffered)989 url = urlsplit(get_current_url(environ))990 self._update_cookies_from_response(991 url.hostname or "localhost", url.path, rv[2].getlist("Set-Cookie")992 )993 return rv994 995 def resolve_redirect(996 self, response: TestResponse, buffered: bool = False997 ) -> TestResponse:998 """Perform a new request to the location given by the redirect999 response to the previous request.1000 1001 :meta private:1002 """1003 scheme, netloc, path, qs, anchor = urlsplit(response.location)1004 builder = EnvironBuilder.from_environ(1005 response.request.environ, path=path, query_string=qs1006 )1007 1008 to_name_parts = netloc.split(":", 1)[0].split(".")1009 from_name_parts = builder.server_name.split(".")1010 1011 if to_name_parts != [""]:1012 # The new location has a host, use it for the base URL.1013 builder.url_scheme = scheme1014 builder.host = netloc1015 else:1016 # A local redirect with autocorrect_location_header=False1017 # doesn't have a host, so use the request's host.1018 to_name_parts = from_name_parts1019 1020 # Explain why a redirect to a different server name won't be followed.1021 if to_name_parts != from_name_parts:1022 if to_name_parts[-len(from_name_parts) :] == from_name_parts:1023 if not self.allow_subdomain_redirects:1024 raise RuntimeError("Following subdomain redirects is not enabled.")1025 else:1026 raise RuntimeError("Following external redirects is not supported.")1027 1028 path_parts = path.split("/")1029 root_parts = builder.script_root.split("/")1030 1031 if path_parts[: len(root_parts)] == root_parts:1032 # Strip the script root from the path.1033 builder.path = path[len(builder.script_root) :]1034 else:1035 # The new location is not under the script root, so use the1036 # whole path and clear the previous root.1037 builder.path = path1038 builder.script_root = ""1039 1040 # Only 307 and 308 preserve all of the original request.1041 if response.status_code not in {307, 308}:1042 # HEAD is preserved, everything else becomes GET.1043 if builder.method != "HEAD":1044 builder.method = "GET"1045 1046 # Clear the body and the headers that describe it.1047 1048 if builder.input_stream is not None:1049 builder.input_stream.close()1050 builder.input_stream = None1051 1052 builder.content_type = None1053 builder.content_length = None1054 builder.headers.pop("Transfer-Encoding", None)1055 1056 return self.open(builder, buffered=buffered)1057 1058 def open(1059 self,1060 *args: t.Any,1061 buffered: bool = False,1062 follow_redirects: bool = False,1063 **kwargs: t.Any,1064 ) -> TestResponse:1065 """Generate an environ dict from the given arguments, make a1066 request to the application using it, and return the response.1067 1068 :param args: Passed to :class:`EnvironBuilder` to create the1069 environ for the request. If a single arg is passed, it can1070 be an existing :class:`EnvironBuilder` or an environ dict.1071 :param buffered: Convert the iterator returned by the app into1072 a list. If the iterator has a ``close()`` method, it is1073 called automatically.1074 :param follow_redirects: Make additional requests to follow HTTP1075 redirects until a non-redirect status is returned.1076 :attr:`TestResponse.history` lists the intermediate1077 responses.1078 1079 .. versionchanged:: 2.11080 Removed the ``as_tuple`` parameter.1081 1082 .. versionchanged:: 2.01083 The request input stream is closed when calling1084 ``response.close()``. Input streams for redirects are1085 automatically closed.1086 1087 .. versionchanged:: 0.51088 If a dict is provided as file in the dict for the ``data``1089 parameter the content type has to be called ``content_type``1090 instead of ``mimetype``. This change was made for1091 consistency with :class:`werkzeug.FileWrapper`.1092 1093 .. versionchanged:: 0.51094 Added the ``follow_redirects`` parameter.1095 """1096 request: Request | None = None1097 1098 if not kwargs and len(args) == 1:1099 arg = args[0]1100 1101 if isinstance(arg, EnvironBuilder):1102 request = arg.get_request()1103 elif isinstance(arg, dict):1104 request = EnvironBuilder.from_environ(arg).get_request()1105 elif isinstance(arg, Request):1106 request = arg1107 1108 if request is None:1109 builder = EnvironBuilder(*args, **kwargs)1110 1111 try:1112 request = builder.get_request()1113 finally:1114 builder.close()1115 1116 response_parts = self.run_wsgi_app(request.environ, buffered=buffered)1117 response = self.response_wrapper(*response_parts, request=request)1118 1119 redirects = set()1120 history: list[TestResponse] = []1121 1122 if not follow_redirects:1123 return response1124 1125 while response.status_code in {1126 301,1127 302,1128 303,1129 305,1130 307,1131 308,1132 }:1133 # Exhaust intermediate response bodies to ensure middleware1134 # that returns an iterator runs any cleanup code.1135 if not buffered:1136 response.make_sequence()1137 response.close()1138 1139 new_redirect_entry = (response.location, response.status_code)1140 1141 if new_redirect_entry in redirects:1142 raise ClientRedirectError(1143 f"Loop detected: A {response.status_code} redirect"1144 f" to {response.location} was already made."1145 )1146 1147 redirects.add(new_redirect_entry)1148 response.history = tuple(history)1149 history.append(response)1150 response = self.resolve_redirect(response, buffered=buffered)1151 else:1152 # This is the final request after redirects.1153 response.history = tuple(history)1154 # Close the input stream when closing the response, in case1155 # the input is an open temporary file.1156 response.call_on_close(request.input_stream.close)1157 return response1158 1159 def get(self, *args: t.Any, **kw: t.Any) -> TestResponse:1160 """Call :meth:`open` with ``method`` set to ``GET``."""1161 kw["method"] = "GET"1162 return self.open(*args, **kw)1163 1164 def post(self, *args: t.Any, **kw: t.Any) -> TestResponse:1165 """Call :meth:`open` with ``method`` set to ``POST``."""1166 kw["method"] = "POST"1167 return self.open(*args, **kw)1168 1169 def put(self, *args: t.Any, **kw: t.Any) -> TestResponse:1170 """Call :meth:`open` with ``method`` set to ``PUT``."""1171 kw["method"] = "PUT"1172 return self.open(*args, **kw)1173 1174 def delete(self, *args: t.Any, **kw: t.Any) -> TestResponse:1175 """Call :meth:`open` with ``method`` set to ``DELETE``."""1176 kw["method"] = "DELETE"1177 return self.open(*args, **kw)1178 1179 def patch(self, *args: t.Any, **kw: t.Any) -> TestResponse:1180 """Call :meth:`open` with ``method`` set to ``PATCH``."""1181 kw["method"] = "PATCH"1182 return self.open(*args, **kw)1183 1184 def options(self, *args: t.Any, **kw: t.Any) -> TestResponse:1185 """Call :meth:`open` with ``method`` set to ``OPTIONS``."""1186 kw["method"] = "OPTIONS"1187 return self.open(*args, **kw)1188 1189 def head(self, *args: t.Any, **kw: t.Any) -> TestResponse:1190 """Call :meth:`open` with ``method`` set to ``HEAD``."""1191 kw["method"] = "HEAD"1192 return self.open(*args, **kw)1193 1194 def trace(self, *args: t.Any, **kw: t.Any) -> TestResponse:1195 """Call :meth:`open` with ``method`` set to ``TRACE``."""1196 kw["method"] = "TRACE"1197 return self.open(*args, **kw)1198 1199 def __repr__(self) -> str:1200 return f"<{type(self).__name__} {self.application!r}>"