Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/werkzeug/test.py: 30%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1from __future__ import annotations
3import dataclasses
4import json
5import mimetypes
6import sys
7import typing as t
8import wsgiref.types as wt
9from collections import defaultdict
10from datetime import datetime
11from io import BytesIO
12from itertools import chain
13from random import random
14from tempfile import TemporaryFile
15from time import time
16from types import TracebackType
17from urllib.parse import unquote
18from urllib.parse import urlsplit
19from urllib.parse import urlunsplit
21from ._internal import _get_environ
22from ._internal import _wsgi_decoding_dance
23from ._internal import _wsgi_encoding_dance
24from .datastructures.auth import Authorization
25from .datastructures.file_storage import FileMultiDict
26from .datastructures.headers import EnvironHeaders
27from .datastructures.headers import Headers
28from .datastructures.structures import CallbackDict
29from .datastructures.structures import CombinedMultiDict
30from .datastructures.structures import MultiDict
31from .http import dump_cookie
32from .http import dump_options_header
33from .http import parse_cookie
34from .http import parse_date
35from .http import parse_options_header
36from .sansio.multipart import Data
37from .sansio.multipart import Epilogue
38from .sansio.multipart import Field
39from .sansio.multipart import File
40from .sansio.multipart import MultipartEncoder
41from .sansio.multipart import Preamble
42from .urls import _urlencode
43from .urls import iri_to_uri
44from .utils import _get_content_type
45from .utils import cached_property
46from .wrappers.request import Request
47from .wrappers.response import Response
48from .wsgi import ClosingIterator
51def stream_encode_multipart(
52 data: t.Mapping[str, t.Any],
53 use_tempfile: bool = True,
54 threshold: int = 1024 * 500,
55 boundary: str | None = None,
56) -> tuple[t.IO[bytes], int, str]:
57 """Encode a dict of values (either strings or file descriptors or
58 :class:`FileStorage` objects.) into a multipart encoded string stored
59 in a file descriptor.
61 This is not secure and must only be used during testing, never in applications.
63 .. versionchanged:: 3.0
64 The ``charset`` parameter was removed.
65 """
66 if boundary is None:
67 boundary = f"---------------WerkzeugFormPart_{time()}{random()}"
69 stream: t.IO[bytes] = BytesIO()
70 total_length = 0
71 on_disk = False
72 write_binary: t.Callable[[bytes], int]
74 if use_tempfile:
76 def write_binary(s: bytes) -> int:
77 nonlocal stream, total_length, on_disk
79 if on_disk:
80 return stream.write(s)
81 else:
82 length = len(s)
84 if length + total_length <= threshold:
85 stream.write(s)
86 else:
87 new_stream = t.cast(t.IO[bytes], TemporaryFile("wb+"))
88 new_stream.write(stream.getvalue()) # type: ignore[attr-defined]
89 new_stream.write(s)
90 stream = new_stream
91 on_disk = True
93 total_length += length
94 return length
96 else:
97 write_binary = stream.write
99 encoder = MultipartEncoder(boundary.encode())
100 write_binary(encoder.send_event(Preamble(data=b"")))
101 for key, value in _iter_data(data):
102 reader = getattr(value, "read", None)
103 if reader is not None:
104 filename = getattr(value, "filename", getattr(value, "name", None))
105 content_type = getattr(value, "content_type", None)
106 if content_type is None:
107 content_type = (
108 filename
109 and mimetypes.guess_type(filename)[0]
110 or "application/octet-stream"
111 )
112 headers = value.headers
113 headers.update([("Content-Type", content_type)])
114 if filename is None:
115 write_binary(encoder.send_event(Field(name=key, headers=headers)))
116 else:
117 write_binary(
118 encoder.send_event(
119 File(name=key, filename=filename, headers=headers)
120 )
121 )
122 while True:
123 chunk = reader(16384)
125 if not chunk:
126 write_binary(encoder.send_event(Data(data=chunk, more_data=False)))
127 break
129 write_binary(encoder.send_event(Data(data=chunk, more_data=True)))
130 else:
131 if not isinstance(value, str):
132 value = str(value)
133 write_binary(encoder.send_event(Field(name=key, headers=Headers())))
134 write_binary(encoder.send_event(Data(data=value.encode(), more_data=False)))
136 write_binary(encoder.send_event(Epilogue(data=b"")))
138 length = stream.tell()
139 stream.seek(0)
140 return stream, length, boundary
143def encode_multipart(
144 values: t.Mapping[str, t.Any], boundary: str | None = None
145) -> tuple[str, bytes]:
146 """Like `stream_encode_multipart` but returns a tuple in the form
147 (``boundary``, ``data``) where data is bytes.
149 This is not secure and must only be used during testing, never in applications.
151 .. versionchanged:: 3.0
152 The ``charset`` parameter was removed.
153 """
154 stream, length, boundary = stream_encode_multipart(
155 values, use_tempfile=False, boundary=boundary
156 )
157 return boundary, stream.read()
160def _iter_data(data: t.Mapping[str, t.Any]) -> t.Iterator[tuple[str, t.Any]]:
161 """Iterate over a mapping that might have a list of values, yielding
162 all key, value pairs. Almost like iter_multi_items but only allows
163 lists, not tuples, of values so tuples can be used for files.
164 """
165 if isinstance(data, MultiDict):
166 yield from data.items(multi=True)
167 else:
168 for key, value in data.items():
169 if isinstance(value, list):
170 for v in value:
171 yield key, v
172 else:
173 yield key, value
176class EnvironBuilder:
177 """This class can be used to conveniently create a WSGI environment
178 for testing purposes. It can be used to quickly create WSGI environments
179 or request objects from arbitrary data.
181 The signature of this class is also used in some other places as of
182 Werkzeug 0.5 (:func:`create_environ`, :meth:`Response.from_values`,
183 :meth:`Client.open`). Because of this most of the functionality is
184 available through the constructor alone.
186 :param path: the path of the request. In the WSGI environment this will
187 end up as `PATH_INFO`. If the `query_string` is not defined
188 and there is a question mark in the `path` everything after
189 it is used as query string.
190 :param base_url: the base URL is a URL that is used to extract the WSGI
191 URL scheme, host (server name + server port) and the
192 script root (`SCRIPT_NAME`).
193 :param query_string: A :class:`dict` or :class:`.MultiDict` to encode as the
194 query string of the URL, which sets :attr:`args`. Or a string, which
195 sets :attr:`query_string`, in which case :attr:`args` cannot be used.
196 :param method: the HTTP method to use, defaults to `GET`.
197 :param content_type: The content type for the request. As of 0.5 you
198 don't have to provide this when specifying files
199 and form data via `data`.
200 :param content_length: The content length for the request. You don't
201 have to specify this when providing data via
202 `data`.
203 :param errors_stream: an optional error stream that is used for
204 `wsgi.errors`. Defaults to :data:`stderr`.
205 :param multithread: controls `wsgi.multithread`. Defaults to `False`.
206 :param multiprocess: controls `wsgi.multiprocess`. Defaults to `False`.
207 :param run_once: controls `wsgi.run_once`. Defaults to `False`.
208 :param headers: an optional list or :class:`Headers` object of headers.
209 :param data: A dict of form and file data to encode as the body of the
210 request; file values can be an IO object, ``(stream, filename)``,
211 ``(stream, filename, content type)``, or :class:`.FileStorage`.
212 Alternatively, pass raw bytes to set as :attr:`input_stream`, or an IO
213 object to read and set. ``json`` can be used to set JSON data instead.
214 ``content_length`` is set automatically.
215 :param json: An object to be serialized and assigned to ``data``.
216 Defaults the content type to ``"application/json"``.
217 Serialized with the function assigned to :attr:`json_dumps`.
218 :param environ_base: an optional dict of environment defaults.
219 :param environ_overrides: an optional dict of environment overrides.
220 :param auth: An authorization object to use for the
221 ``Authorization`` header value. A ``(username, password)`` tuple
222 is a shortcut for ``Basic`` authorization.
223 :param input_stream: An IO object to pass through as the body of the
224 request, without reading, which simulates a streaming request. The
225 stream is not closed when calling :meth:`.close`, as it must remain open
226 to be read in the application.
228 .. versionchanged:: 3.2
229 Can be used as a ``with`` context manager to automatically close
231 .. versionchanged:: 3.0
232 The ``charset`` parameter was removed.
234 .. versionchanged:: 2.1
235 ``CONTENT_TYPE`` and ``CONTENT_LENGTH`` are not duplicated as
236 header keys in the environ.
238 .. versionchanged:: 2.0
239 ``REQUEST_URI`` and ``RAW_URI`` is the full raw URI including
240 the query string, not only the path.
242 .. versionchanged:: 2.0
243 The default :attr:`request_class` is ``Request`` instead of
244 ``BaseRequest``.
246 .. versionadded:: 2.0
247 Added the ``auth`` parameter.
249 .. versionadded:: 0.15
250 The ``json`` param and :meth:`json_dumps` method.
252 .. versionadded:: 0.15
253 The environ has keys ``REQUEST_URI`` and ``RAW_URI`` containing
254 the path before percent-decoding. This is not part of the WSGI
255 PEP, but many WSGI servers include it.
257 .. versionchanged:: 0.6
258 ``path`` and ``base_url`` can now be unicode strings that are
259 encoded with :func:`iri_to_uri`.
260 """
262 #: the server protocol to use. defaults to HTTP/1.1
263 server_protocol = "HTTP/1.1"
265 #: the wsgi version to use. defaults to (1, 0)
266 wsgi_version = (1, 0)
268 #: The default request class used by :meth:`get_request`.
269 request_class = Request
271 #: The serialization function used when ``json`` is passed.
272 json_dumps = staticmethod(json.dumps)
274 _args: MultiDict[str, str] = MultiDict()
275 _query_string: str | None = None
276 _form: MultiDict[str, str] = MultiDict()
277 _files: FileMultiDict = FileMultiDict()
278 _input_stream: t.IO[bytes] | None = None
280 def __init__(
281 self,
282 path: str = "/",
283 base_url: str | None = None,
284 query_string: t.Mapping[str, str] | str | None = None,
285 method: str = "GET",
286 input_stream: t.IO[bytes] | None = None,
287 content_type: str | None = None,
288 content_length: int | None = None,
289 errors_stream: t.IO[str] | None = None,
290 multithread: bool = False,
291 multiprocess: bool = False,
292 run_once: bool = False,
293 headers: Headers
294 | t.Mapping[str, str]
295 | t.Iterable[tuple[str, str]]
296 | None = None,
297 data: t.IO[bytes] | str | bytes | t.Mapping[str, t.Any] | None = None,
298 environ_base: t.Mapping[str, t.Any] | None = None,
299 environ_overrides: t.Mapping[str, t.Any] | None = None,
300 mimetype: str | None = None,
301 json: t.Mapping[str, t.Any] | None = None,
302 auth: Authorization | tuple[str, str] | None = None,
303 ) -> None:
304 if query_string is not None and "?" in path:
305 raise ValueError("Query string is defined in the path and as an argument")
306 request_uri = urlsplit(path)
307 if query_string is None and "?" in path:
308 query_string = request_uri.query
310 self.path = iri_to_uri(request_uri.path)
311 self.request_uri = path
312 if base_url is not None:
313 base_url = iri_to_uri(base_url)
314 self.base_url = base_url
315 if isinstance(query_string, str):
316 self.query_string = query_string
317 else:
318 if query_string is None:
319 query_string = MultiDict()
320 elif not isinstance(query_string, MultiDict):
321 query_string = MultiDict(query_string)
322 self.args = query_string
323 self.method = method
324 if headers is None:
325 headers = Headers()
326 elif not isinstance(headers, Headers):
327 headers = Headers(headers)
328 self.headers = headers
329 if content_type is not None:
330 self.content_type = content_type
331 if errors_stream is None:
332 errors_stream = sys.stderr
333 self.errors_stream = errors_stream
334 self.multithread = multithread
335 self.multiprocess = multiprocess
336 self.run_once = run_once
337 self.environ_base = environ_base
338 self.environ_overrides = environ_overrides
339 self.input_stream = input_stream
340 self.content_length = content_length
342 if auth is not None:
343 if isinstance(auth, tuple):
344 auth = Authorization(
345 "basic", {"username": auth[0], "password": auth[1]}
346 )
348 self.headers.set("Authorization", auth.to_header())
350 if json is not None:
351 if data is not None:
352 raise TypeError("can't provide both json and data")
354 data = self.json_dumps(json)
356 if self.content_type is None:
357 self.content_type = "application/json"
359 if data:
360 if input_stream is not None:
361 raise TypeError("can't provide input stream and data")
362 if hasattr(data, "read"):
363 data = data.read()
364 if isinstance(data, str):
365 data = data.encode()
366 if isinstance(data, bytes):
367 self.input_stream = BytesIO(data)
368 if self.content_length is None:
369 self.content_length = len(data)
370 else:
371 for key, value in _iter_data(data):
372 if isinstance(value, (tuple, dict)) or hasattr(value, "read"):
373 self._add_file_from_data(key, value)
374 else:
375 self.form.setlistdefault(key).append(value)
377 if mimetype is not None:
378 self.mimetype = mimetype
380 @classmethod
381 def from_environ(cls, environ: wt.WSGIEnvironment, **kwargs: t.Any) -> t.Self:
382 """Turn an environ dict back into a builder. Any extra kwargs
383 override the args extracted from the environ.
385 .. versionchanged:: 2.0
386 Path and query values are passed through the WSGI decoding
387 dance to avoid double encoding.
389 .. versionadded:: 0.15
390 """
391 headers = Headers(EnvironHeaders(environ))
392 out = {
393 "path": _wsgi_decoding_dance(environ["PATH_INFO"]),
394 "base_url": cls._make_base_url(
395 environ["wsgi.url_scheme"],
396 headers.pop("Host"),
397 _wsgi_decoding_dance(environ["SCRIPT_NAME"]),
398 ),
399 "query_string": _wsgi_decoding_dance(environ["QUERY_STRING"]),
400 "method": environ["REQUEST_METHOD"],
401 "input_stream": environ["wsgi.input"],
402 "content_type": headers.pop("Content-Type", None),
403 "content_length": headers.pop("Content-Length", None),
404 "errors_stream": environ["wsgi.errors"],
405 "multithread": environ["wsgi.multithread"],
406 "multiprocess": environ["wsgi.multiprocess"],
407 "run_once": environ["wsgi.run_once"],
408 "headers": headers,
409 }
410 out.update(kwargs)
411 return cls(**out)
413 def _add_file_from_data(
414 self,
415 key: str,
416 value: (t.IO[bytes] | tuple[t.IO[bytes], str] | tuple[t.IO[bytes], str, str]),
417 ) -> None:
418 """Called in the EnvironBuilder to add files from the data dict."""
419 if isinstance(value, tuple):
420 self.files.add_file(key, *value)
421 else:
422 self.files.add_file(key, value)
424 @staticmethod
425 def _make_base_url(scheme: str, host: str, script_root: str) -> str:
426 return urlunsplit((scheme, host, script_root, "", "")).rstrip("/") + "/"
428 @property
429 def base_url(self) -> str:
430 """The base URL is used to extract the URL scheme, host name,
431 port, and root path.
432 """
433 return self._make_base_url(self.url_scheme, self.host, self.script_root)
435 @base_url.setter
436 def base_url(self, value: str | None) -> None:
437 if value is None:
438 scheme = "http"
439 netloc = "localhost"
440 script_root = ""
441 else:
442 scheme, netloc, script_root, qs, anchor = urlsplit(value)
443 if qs or anchor:
444 raise ValueError("base url must not contain a query string or fragment")
445 self.script_root = script_root.rstrip("/")
446 self.host = netloc
447 self.url_scheme = scheme
449 @property
450 def content_type(self) -> str | None:
451 """The content type for the request. Reflected from and to
452 the :attr:`headers`. Do not set if you set :attr:`files` or
453 :attr:`form` for auto detection.
454 """
455 ct = self.headers.get("Content-Type")
456 if ct is None and not self._input_stream:
457 if self._files:
458 return "multipart/form-data"
459 if self._form:
460 return "application/x-www-form-urlencoded"
461 return None
462 return ct
464 @content_type.setter
465 def content_type(self, value: str | None) -> None:
466 if value is None:
467 self.headers.pop("Content-Type", None)
468 else:
469 self.headers["Content-Type"] = value
471 @property
472 def mimetype(self) -> str | None:
473 """The mimetype (content type without charset etc.)
475 .. versionadded:: 0.14
476 """
477 ct = self.content_type
478 return ct.partition(";")[0].strip() if ct else None
480 @mimetype.setter
481 def mimetype(self, value: str) -> None:
482 self.content_type = _get_content_type(value, "utf-8")
484 @property
485 def mimetype_params(self) -> t.Mapping[str, str]:
486 """The mimetype parameters as dict. For example if the
487 content type is ``text/html; charset=utf-8`` the params would be
488 ``{'charset': 'utf-8'}``.
490 .. versionadded:: 0.14
491 """
493 def on_update(d: CallbackDict[str, str]) -> None:
494 self.headers["Content-Type"] = dump_options_header(self.mimetype, d)
496 d = parse_options_header(self.headers.get("Content-Type", ""))[1]
497 return CallbackDict(d, on_update)
499 @property
500 def content_length(self) -> int | None:
501 """The content length as integer. Reflected from and to the
502 :attr:`headers`. Do not set if you set :attr:`files` or
503 :attr:`form` for auto detection.
504 """
505 return self.headers.get("Content-Length", type=int)
507 @content_length.setter
508 def content_length(self, value: int | None) -> None:
509 if value is None:
510 self.headers.pop("Content-Length", None)
511 else:
512 self.headers["Content-Length"] = str(value)
514 @property
515 def form(self) -> MultiDict[str, str]:
516 """Form data text values. File values are stored in :attr:`files`.
518 If any values are set and no files are set, the request body will be
519 encoded and the content type set to ``application/x-www-form-urlencoded``.
521 Set when the constructor ``data`` parameter is a dict or multidict.
523 Cannot be accessed if :attr:`input_stream` is set. Setting this will
524 unset :attr:`input_stream`.
525 """
526 if self.input_stream is not None:
527 raise AttributeError("Not available when 'input_stream' is set.")
529 return self._form
531 @form.setter
532 def form(self, value: MultiDict[str, str]) -> None:
533 self._input_stream = None
534 self._form = value
536 @property
537 def files(self) -> FileMultiDict:
538 """Form data file values. Text values are stored in :attr:`form`.
540 If any values are set, the request body will be encoded and the content
541 type set to ``multipart/form-data``.
543 Set when the constructor ``data`` parameter is a dict or multidict.
545 The :meth:`.FileMultiDict.add_file` method provides a convenient way to
546 add more files without needing to construct :class:`.FileStorage`
547 objects.
549 The file streams will be read when encoding the data. All streams will
550 be closed when the builder's :meth:`.close` method is called.
552 Cannot be accessed if :attr:`input_stream` is set. Setting this will
553 unset :attr:`input_stream`.
555 Setting this will _not_ close files in the previous dict, call
556 :meth:`.FileMultiDict.close` first if that's needed.
557 """
558 if self.input_stream is not None:
559 raise AttributeError("Not available when 'input_stream' is set.")
561 return self._files
563 @files.setter
564 def files(self, value: FileMultiDict) -> None:
565 self._input_stream = None
566 self._files = value
568 @property
569 def input_stream(self) -> t.IO[bytes] | None:
570 """A binary IO object to pass through as the body of the request,
571 without reading, which simulates a streaming request.
573 If this is set, :attr:`form` and :attr:`files` cannot be accessed.
574 If those are set, this will be ``None``. Setting this will close any
575 files and clear those.
577 The stream is not closed when calling the builder's :meth:`close`
578 method, as it must remain open to be read in the application.
580 .. versionchanged:: 3.2
581 Any values in :attr:`files` are closed first when setting this.
582 """
583 return self._input_stream
585 @input_stream.setter
586 def input_stream(self, value: t.IO[bytes] | None) -> None:
587 self._form.clear()
588 self._files.clear()
589 self._input_stream = value
591 @property
592 def query_string(self) -> str:
593 """The URL query string.
595 If this is set to a string, :attr:`args` cannot be accessed. If
596 :attr:`args` is set, this will be the encoded value of that. If neither
597 is set, this is the empty string.
598 """
599 if self._query_string is None:
600 if self._args is not None:
601 return _urlencode(self._args)
603 return ""
605 return self._query_string
607 @query_string.setter
608 def query_string(self, value: str | None) -> None:
609 self._args.clear()
610 self._query_string = value
612 @property
613 def args(self) -> MultiDict[str, str]:
614 """The URL query string as a :class:`MultiDict`.
616 Set when the constructor ``query_string`` parameter is a dict or
617 multidict.
619 Setting this will unset :attr:`query_string`.
620 """
621 if self._query_string is not None:
622 raise AttributeError("Not available when 'query_string' is set.")
624 return self._args
626 @args.setter
627 def args(self, value: MultiDict[str, str]) -> None:
628 self._query_string = None
629 self._args = value
631 @property
632 def server_name(self) -> str:
633 """The server name (read-only, use :attr:`host` to set)"""
634 if self.host.endswith("]"):
635 return self.host
637 left, sep, right = self.host.rpartition(":")
638 return left if sep else right
640 @property
641 def server_port(self) -> int:
642 """The server port as integer (read-only, use :attr:`host` to set)"""
643 if not self.host.endswith("]"):
644 _, sep, right = self.host.rpartition(":")
646 if sep:
647 return int(right)
649 if self.url_scheme in {"https", "wss"}:
650 return 443
652 return 80
654 def __del__(self) -> None:
655 self.close()
657 def __enter__(self) -> t.Self:
658 return self
660 def __exit__(
661 self,
662 exc_type: type[BaseException] | None,
663 exc_val: BaseException | None,
664 exc_tb: TracebackType | None,
665 ) -> None:
666 self.close()
668 def close(self) -> None:
669 """Close all open files in :attr:`files`. :attr:`input_stream` is not
670 closed, as it is assumed to be managed externally.
671 """
672 self._files.close()
674 def get_environ(self) -> wt.WSGIEnvironment:
675 """Return the built environ.
677 .. versionchanged:: 0.15
678 The content type and length headers are set based on
679 input stream detection. Previously this only set the WSGI
680 keys.
681 """
682 input_stream = self.input_stream
683 content_length = self.content_length
685 mimetype = self.mimetype
686 content_type = self.content_type
688 if input_stream is not None:
689 start_pos = input_stream.tell()
690 input_stream.seek(0, 2)
691 end_pos = input_stream.tell()
692 input_stream.seek(start_pos)
693 content_length = end_pos - start_pos
694 elif mimetype == "multipart/form-data":
695 input_stream, content_length, boundary = stream_encode_multipart(
696 CombinedMultiDict([self.form, self.files])
697 )
698 content_type = f'{mimetype}; boundary="{boundary}"'
699 elif mimetype == "application/x-www-form-urlencoded":
700 form_encoded = _urlencode(self.form).encode("ascii")
701 content_length = len(form_encoded)
702 input_stream = BytesIO(form_encoded)
703 else:
704 input_stream = BytesIO()
706 result: wt.WSGIEnvironment = {}
707 if self.environ_base:
708 result.update(self.environ_base)
710 def _path_encode(x: str) -> str:
711 return _wsgi_encoding_dance(unquote(x))
713 raw_uri = _wsgi_encoding_dance(self.request_uri)
714 result.update(
715 {
716 "REQUEST_METHOD": self.method,
717 "SCRIPT_NAME": _path_encode(self.script_root),
718 "PATH_INFO": _path_encode(self.path),
719 "QUERY_STRING": _wsgi_encoding_dance(self.query_string),
720 # Non-standard, added by mod_wsgi, uWSGI
721 "REQUEST_URI": raw_uri,
722 # Non-standard, added by gunicorn
723 "RAW_URI": raw_uri,
724 "SERVER_NAME": self.server_name,
725 "SERVER_PORT": str(self.server_port),
726 "HTTP_HOST": self.host,
727 "SERVER_PROTOCOL": self.server_protocol,
728 "wsgi.version": self.wsgi_version,
729 "wsgi.url_scheme": self.url_scheme,
730 "wsgi.input": input_stream,
731 "wsgi.errors": self.errors_stream,
732 "wsgi.multithread": self.multithread,
733 "wsgi.multiprocess": self.multiprocess,
734 "wsgi.run_once": self.run_once,
735 }
736 )
738 headers = self.headers.copy()
739 # Don't send these as headers, they're part of the environ.
740 headers.remove("Content-Type")
741 headers.remove("Content-Length")
743 if content_type is not None:
744 result["CONTENT_TYPE"] = content_type
746 if content_length is not None:
747 result["CONTENT_LENGTH"] = str(content_length)
749 combined_headers = defaultdict(list)
751 for key, value in headers.to_wsgi_list():
752 combined_headers[f"HTTP_{key.upper().replace('-', '_')}"].append(value)
754 for key, values in combined_headers.items():
755 result[key] = ", ".join(values)
757 if self.environ_overrides:
758 result.update(self.environ_overrides)
760 return result
762 def get_request(self, cls: type[Request] | None = None) -> Request:
763 """Returns a request with the data. If the request class is not
764 specified :attr:`request_class` is used.
766 :param cls: The request wrapper to use.
767 """
768 if cls is None:
769 cls = self.request_class
771 return cls(self.get_environ())
774class ClientRedirectError(Exception):
775 """If a redirect loop is detected when using follow_redirects=True with
776 the :cls:`Client`, then this exception is raised.
777 """
780class Client:
781 """Simulate sending requests to a WSGI application without running a WSGI or HTTP
782 server.
784 :param application: The WSGI application to make requests to.
785 :param response_wrapper: A :class:`.Response` class to wrap response data with.
786 Defaults to :class:`.TestResponse`. If it's not a subclass of ``TestResponse``,
787 one will be created.
788 :param use_cookies: Persist cookies from ``Set-Cookie`` response headers to the
789 ``Cookie`` header in subsequent requests. Domain and path matching is supported,
790 but other cookie parameters are ignored.
791 :param allow_subdomain_redirects: Allow requests to follow redirects to subdomains.
792 Enable this if the application handles subdomains and redirects between them.
794 .. versionchanged:: 2.3
795 Simplify cookie implementation, support domain and path matching.
797 .. versionchanged:: 2.1
798 All data is available as properties on the returned response object. The
799 response cannot be returned as a tuple.
801 .. versionchanged:: 2.0
802 ``response_wrapper`` is always a subclass of :class:``TestResponse``.
804 .. versionchanged:: 0.5
805 Added the ``use_cookies`` parameter.
806 """
808 def __init__(
809 self,
810 application: wt.WSGIApplication,
811 response_wrapper: type[Response] | None = None,
812 use_cookies: bool = True,
813 allow_subdomain_redirects: bool = False,
814 ) -> None:
815 self.application = application
817 if response_wrapper in {None, Response}:
818 response_wrapper = TestResponse
819 elif response_wrapper is not None and not issubclass(
820 response_wrapper, TestResponse
821 ):
822 response_wrapper = type(
823 "WrapperTestResponse",
824 (TestResponse, response_wrapper),
825 {},
826 )
828 self.response_wrapper = t.cast(type["TestResponse"], response_wrapper)
830 if use_cookies:
831 self._cookies: dict[tuple[str, str, str], Cookie] | None = {}
832 else:
833 self._cookies = None
835 self.allow_subdomain_redirects = allow_subdomain_redirects
837 def get_cookie(
838 self, key: str, domain: str = "localhost", path: str = "/"
839 ) -> Cookie | None:
840 """Return a :class:`.Cookie` if it exists. Cookies are uniquely identified by
841 ``(domain, path, key)``.
843 :param key: The decoded form of the key for the cookie.
844 :param domain: The domain the cookie was set for.
845 :param path: The path the cookie was set for.
847 .. versionadded:: 2.3
848 """
849 if self._cookies is None:
850 raise TypeError(
851 "Cookies are disabled. Create a client with 'use_cookies=True'."
852 )
854 return self._cookies.get((domain, path, key))
856 def set_cookie(
857 self,
858 key: str,
859 value: str = "",
860 *,
861 domain: str = "localhost",
862 origin_only: bool = True,
863 path: str = "/",
864 **kwargs: t.Any,
865 ) -> None:
866 """Set a cookie to be sent in subsequent requests.
868 This is a convenience to skip making a test request to a route that would set
869 the cookie. To test the cookie, make a test request to a route that uses the
870 cookie value.
872 The client uses ``domain``, ``origin_only``, and ``path`` to determine which
873 cookies to send with a request. It does not use other cookie parameters that
874 browsers use, since they're not applicable in tests.
876 :param key: The key part of the cookie.
877 :param value: The value part of the cookie.
878 :param domain: Send this cookie with requests that match this domain. If
879 ``origin_only`` is true, it must be an exact match, otherwise it may be a
880 suffix match.
881 :param origin_only: Whether the domain must be an exact match to the request.
882 :param path: Send this cookie with requests that match this path either exactly
883 or as a prefix.
884 :param kwargs: Passed to :func:`.dump_cookie`.
886 .. versionchanged:: 3.0
887 The parameter ``server_name`` is removed. The first parameter is
888 ``key``. Use the ``domain`` and ``origin_only`` parameters instead.
890 .. versionchanged:: 2.3
891 The ``origin_only`` parameter was added.
893 .. versionchanged:: 2.3
894 The ``domain`` parameter defaults to ``localhost``.
895 """
896 if self._cookies is None:
897 raise TypeError(
898 "Cookies are disabled. Create a client with 'use_cookies=True'."
899 )
901 cookie = Cookie._from_response_header(
902 domain, "/", dump_cookie(key, value, domain=domain, path=path, **kwargs)
903 )
904 cookie.origin_only = origin_only
906 if cookie._should_delete:
907 self._cookies.pop(cookie._storage_key, None)
908 else:
909 self._cookies[cookie._storage_key] = cookie
911 def delete_cookie(
912 self,
913 key: str,
914 *,
915 domain: str = "localhost",
916 path: str = "/",
917 ) -> None:
918 """Delete a cookie if it exists. Cookies are uniquely identified by
919 ``(domain, path, key)``.
921 :param key: The decoded form of the key for the cookie.
922 :param domain: The domain the cookie was set for.
923 :param path: The path the cookie was set for.
925 .. versionchanged:: 3.0
926 The ``server_name`` parameter is removed. The first parameter is
927 ``key``. Use the ``domain`` parameter instead.
929 .. versionchanged:: 3.0
930 The ``secure``, ``httponly`` and ``samesite`` parameters are removed.
932 .. versionchanged:: 2.3
933 The ``domain`` parameter defaults to ``localhost``.
934 """
935 if self._cookies is None:
936 raise TypeError(
937 "Cookies are disabled. Create a client with 'use_cookies=True'."
938 )
940 self._cookies.pop((domain, path, key), None)
942 def _add_cookies_to_wsgi(self, request: Request) -> None:
943 """If cookies are enabled, set the ``Cookie`` header in the environ to the
944 cookies that are applicable to the request host and path.
946 :meta private:
948 .. versionadded:: 2.3
949 """
950 if self._cookies is None:
951 return
953 url = urlsplit(request.url)
954 server_name = url.hostname or "localhost"
955 value = "; ".join(
956 c._to_request_header()
957 for c in self._cookies.values()
958 if c._matches_request(server_name, url.path)
959 )
961 if value:
962 request.environ["HTTP_COOKIE"] = value
963 else:
964 request.environ.pop("HTTP_COOKIE", None)
966 def _update_cookies_from_response(
967 self, request: Request, response: TestResponse
968 ) -> None:
969 """If cookies are enabled, update the stored cookies from any ``Set-Cookie``
970 headers in the response.
972 :meta private:
974 .. versionadded:: 2.3
975 """
976 if self._cookies is None:
977 return
979 url = urlsplit(request.url)
980 server_name = url.hostname or "localhost"
981 path = url.path
983 for header in response.headers.getlist("Set-Cookie"):
984 cookie = Cookie._from_response_header(server_name, path, header)
986 if cookie._should_delete:
987 self._cookies.pop(cookie._storage_key, None)
988 else:
989 self._cookies[cookie._storage_key] = cookie
991 def run_wsgi_app(self, request: Request, buffered: bool = False) -> TestResponse:
992 """Runs the wrapped WSGI app with the given environment.
994 :meta private:
995 """
996 self._add_cookies_to_wsgi(request)
997 app_iter, status, headers = run_wsgi_app(
998 self.application, request.environ, buffered=buffered
999 )
1000 response = self.response_wrapper(app_iter, status, headers, request=request)
1001 self._update_cookies_from_response(request, response)
1002 return response
1004 def resolve_redirect(
1005 self, response: TestResponse, buffered: bool = False
1006 ) -> TestResponse:
1007 """Perform a new request to the location given by the redirect
1008 response to the previous request.
1010 :meta private:
1011 """
1012 assert response.location is not None
1013 scheme, netloc, path, qs, anchor = urlsplit(response.location)
1014 builder = EnvironBuilder.from_environ(
1015 response.request.environ, path=path, query_string=qs
1016 )
1018 to_name_parts = netloc.partition(":")[0].split(".")
1019 from_name_parts = builder.server_name.split(".")
1021 if to_name_parts != [""]:
1022 # The new location has a host, use it for the base URL.
1023 builder.url_scheme = scheme
1024 builder.host = netloc
1025 else:
1026 # A local redirect with autocorrect_location_header=False
1027 # doesn't have a host, so use the request's host.
1028 to_name_parts = from_name_parts
1030 # Explain why a redirect to a different server name won't be followed.
1031 if to_name_parts != from_name_parts:
1032 if to_name_parts[-len(from_name_parts) :] == from_name_parts:
1033 if not self.allow_subdomain_redirects:
1034 raise RuntimeError("Following subdomain redirects is not enabled.")
1035 else:
1036 raise RuntimeError("Following external redirects is not supported.")
1038 path_parts = path.split("/")
1039 root_parts = builder.script_root.split("/")
1041 if path_parts[: len(root_parts)] == root_parts:
1042 # Strip the script root from the path.
1043 builder.path = path[len(builder.script_root) :]
1044 else:
1045 # The new location is not under the script root, so use the
1046 # whole path and clear the previous root.
1047 builder.path = path
1048 builder.script_root = ""
1050 # Certain statuses switch to GET in some cases
1051 # https://fetch.spec.whatwg.org/#http-redirect-fetch
1052 if (response.status_code in {301, 302} and builder.method == "POST") or (
1053 response.status_code == 303 and builder.method not in {"GET", "HEAD"}
1054 ):
1055 builder.method = "GET"
1057 if builder.input_stream is not None:
1058 builder.input_stream.close()
1060 builder.input_stream = None # also closes and clears form and files
1061 builder.content_type = None
1062 builder.content_length = None
1063 builder.headers.pop("Content-Encoding", None)
1064 builder.headers.pop("Content-Language", None)
1065 builder.headers.pop("Content-Location", None)
1066 builder.headers.pop("Transfer-Encoding", None)
1068 return self.open(builder, buffered=buffered)
1070 def open(
1071 self,
1072 *args: t.Any,
1073 buffered: bool = False,
1074 follow_redirects: bool = False,
1075 **kwargs: t.Any,
1076 ) -> TestResponse:
1077 """Generate an environ dict from the given arguments, make a
1078 request to the application using it, and return the response.
1080 :param args: Passed to :class:`EnvironBuilder` to create the
1081 environ for the request. If a single arg is passed, it can
1082 be an existing :class:`EnvironBuilder` or an environ dict.
1083 :param buffered: Convert the iterator returned by the app into
1084 a list. If the iterator has a ``close()`` method, it is
1085 called automatically.
1086 :param follow_redirects: Make additional requests to follow HTTP
1087 redirects until a non-redirect status is returned.
1088 :attr:`TestResponse.history` lists the intermediate
1089 responses.
1091 .. versionchanged:: 2.1
1092 Removed the ``as_tuple`` parameter.
1094 .. versionchanged:: 2.0
1095 The request input stream is closed when calling
1096 ``response.close()``. Input streams for redirects are
1097 automatically closed.
1099 .. versionchanged:: 0.5
1100 If a dict is provided as file in the dict for the ``data``
1101 parameter the content type has to be called ``content_type``
1102 instead of ``mimetype``. This change was made for
1103 consistency with :class:`werkzeug.FileWrapper`.
1105 .. versionchanged:: 0.5
1106 Added the ``follow_redirects`` parameter.
1107 """
1108 request: Request | None = None
1110 if not kwargs and len(args) == 1:
1111 arg = args[0]
1113 if isinstance(arg, EnvironBuilder):
1114 request = arg.get_request()
1115 elif isinstance(arg, dict):
1116 with EnvironBuilder.from_environ(arg) as builder:
1117 request = builder.get_request()
1118 elif isinstance(arg, Request):
1119 request = arg
1121 if request is None:
1122 with EnvironBuilder(*args, **kwargs) as builder:
1123 request = builder.get_request()
1125 response = self.run_wsgi_app(request, buffered=buffered)
1127 redirects = set()
1128 history: list[TestResponse] = []
1130 if not follow_redirects:
1131 return response
1133 while response.status_code in {301, 302, 303, 307, 308}:
1134 # Exhaust intermediate response bodies to ensure middleware
1135 # that returns an iterator runs any cleanup code.
1136 if not buffered:
1137 response.make_sequence()
1138 response.close()
1140 new_redirect_entry = (response.location, response.status_code)
1142 if new_redirect_entry in redirects:
1143 raise ClientRedirectError(
1144 f"Loop detected: A {response.status_code} redirect"
1145 f" to {response.location} was already made."
1146 )
1148 redirects.add(new_redirect_entry)
1149 response.history = tuple(history)
1150 history.append(response)
1151 response = self.resolve_redirect(response, buffered=buffered)
1152 else:
1153 # This is the final request after redirects.
1154 response.history = tuple(history)
1155 # Close the input stream when closing the response, in case
1156 # the input is an open temporary file.
1157 response.call_on_close(request.environ["wsgi.input"].close)
1158 return response
1160 def get(self, *args: t.Any, **kw: t.Any) -> TestResponse:
1161 """Call :meth:`open` with ``method`` set to ``GET``."""
1162 kw["method"] = "GET"
1163 return self.open(*args, **kw)
1165 def query(self, *args: t.Any, **kw: t.Any) -> TestResponse:
1166 """Call :meth:`open` with ``method`` set to ``QUERY``.
1168 .. versionadded:: 3.2
1169 """
1170 kw["method"] = "QUERY"
1171 return self.open(*args, **kw)
1173 def post(self, *args: t.Any, **kw: t.Any) -> TestResponse:
1174 """Call :meth:`open` with ``method`` set to ``POST``."""
1175 kw["method"] = "POST"
1176 return self.open(*args, **kw)
1178 def put(self, *args: t.Any, **kw: t.Any) -> TestResponse:
1179 """Call :meth:`open` with ``method`` set to ``PUT``."""
1180 kw["method"] = "PUT"
1181 return self.open(*args, **kw)
1183 def delete(self, *args: t.Any, **kw: t.Any) -> TestResponse:
1184 """Call :meth:`open` with ``method`` set to ``DELETE``."""
1185 kw["method"] = "DELETE"
1186 return self.open(*args, **kw)
1188 def patch(self, *args: t.Any, **kw: t.Any) -> TestResponse:
1189 """Call :meth:`open` with ``method`` set to ``PATCH``."""
1190 kw["method"] = "PATCH"
1191 return self.open(*args, **kw)
1193 def options(self, *args: t.Any, **kw: t.Any) -> TestResponse:
1194 """Call :meth:`open` with ``method`` set to ``OPTIONS``."""
1195 kw["method"] = "OPTIONS"
1196 return self.open(*args, **kw)
1198 def head(self, *args: t.Any, **kw: t.Any) -> TestResponse:
1199 """Call :meth:`open` with ``method`` set to ``HEAD``."""
1200 kw["method"] = "HEAD"
1201 return self.open(*args, **kw)
1203 def trace(self, *args: t.Any, **kw: t.Any) -> TestResponse:
1204 """Call :meth:`open` with ``method`` set to ``TRACE``."""
1205 kw["method"] = "TRACE"
1206 return self.open(*args, **kw)
1208 def __repr__(self) -> str:
1209 return f"<{type(self).__name__} {self.application!r}>"
1212def create_environ(*args: t.Any, **kwargs: t.Any) -> wt.WSGIEnvironment:
1213 """Create a new WSGI environ dict based on the values passed. The first
1214 parameter should be the path of the request which defaults to '/'. The
1215 second one can either be an absolute path (in that case the host is
1216 localhost:80) or a full path to the request with scheme, netloc port and
1217 the path to the script.
1219 This accepts the same arguments as the :class:`EnvironBuilder`
1220 constructor.
1222 .. versionchanged:: 0.5
1223 This function is now a thin wrapper over :class:`EnvironBuilder` which
1224 was added in 0.5. The `headers`, `environ_base`, `environ_overrides`
1225 and `charset` parameters were added.
1226 """
1227 with EnvironBuilder(*args, **kwargs) as builder:
1228 return builder.get_environ()
1231def run_wsgi_app(
1232 app: wt.WSGIApplication, environ: wt.WSGIEnvironment, buffered: bool = False
1233) -> tuple[t.Iterable[bytes], str, Headers]:
1234 """Return a tuple in the form (app_iter, status, headers) of the
1235 application output. This works best if you pass it an application that
1236 returns an iterator all the time.
1238 Sometimes applications may use the `write()` callable returned
1239 by the `start_response` function. This tries to resolve such edge
1240 cases automatically. But if you don't get the expected output you
1241 should set `buffered` to `True` which enforces buffering.
1243 If passed an invalid WSGI application the behavior of this function is
1244 undefined. Never pass non-conforming WSGI applications to this function.
1246 :param app: the application to execute.
1247 :param buffered: set to `True` to enforce buffering.
1248 :return: tuple in the form ``(app_iter, status, headers)``
1249 """
1250 # Copy environ to ensure any mutations by the app (ProxyFix, for
1251 # example) don't affect subsequent requests (such as redirects).
1252 environ = _get_environ(environ).copy()
1253 status: str
1254 response: tuple[str, list[tuple[str, str]]] | None = None
1255 buffer: list[bytes] = []
1257 def start_response(
1258 status: str,
1259 headers: list[tuple[str, str]],
1260 exc_info: tuple[type[BaseException], BaseException, TracebackType]
1261 | tuple[None, None, None]
1262 | None = None,
1263 ) -> t.Callable[[bytes], object]:
1264 nonlocal response
1266 if exc_info:
1267 try:
1268 raise exc_info[1].with_traceback(exc_info[2]) # type: ignore[union-attr]
1269 finally:
1270 exc_info = None
1272 response = (status, headers)
1273 return buffer.append
1275 app_rv = app(environ, start_response)
1276 close_func = getattr(app_rv, "close", None)
1277 app_iter: t.Iterable[bytes] = iter(app_rv)
1279 # when buffering we emit the close call early and convert the
1280 # application iterator into a regular list
1281 if buffered:
1282 try:
1283 app_iter = list(app_iter)
1284 finally:
1285 if close_func is not None:
1286 close_func()
1288 # Otherwise, iterate until a response, chain any received data with the
1289 # original, and restore a close callable if needed.
1290 else:
1291 for item in app_iter:
1292 buffer.append(item)
1294 if response is not None:
1295 break
1297 if buffer:
1298 app_iter = chain(buffer, app_iter)
1300 if close_func is not None and app_iter is not app_rv:
1301 app_iter = ClosingIterator(app_iter, close_func)
1303 status, headers = response # type: ignore[misc]
1304 return app_iter, status, Headers(headers)
1307class TestResponse(Response):
1308 """:class:`~werkzeug.wrappers.Response` subclass that provides extra
1309 information about requests made with the test :class:`Client`.
1311 Test client requests will always return an instance of this class.
1312 If a custom response class is passed to the client, it is
1313 subclassed along with this to support test information.
1315 If the test request included large files, or if the application is
1316 serving a file, call :meth:`close` to close any open files and
1317 prevent Python showing a ``ResourceWarning``.
1319 .. versionchanged:: 2.2
1320 Set the ``default_mimetype`` to None to prevent a mimetype being
1321 assumed if missing.
1323 .. versionchanged:: 2.1
1324 Response instances cannot be treated as tuples.
1326 .. versionadded:: 2.0
1327 Test client methods always return instances of this class.
1328 """
1330 default_mimetype = None
1331 # Don't assume a mimetype, instead use whatever the response provides
1333 request: Request
1334 """A request object with the environ used to make the request that
1335 resulted in this response.
1336 """
1338 history: tuple[TestResponse, ...]
1339 """A list of intermediate responses. Populated when the test request
1340 is made with ``follow_redirects`` enabled.
1341 """
1343 # Tell Pytest to ignore this, it's not a test class.
1344 __test__ = False
1346 def __init__(
1347 self,
1348 response: t.Iterable[bytes],
1349 status: str,
1350 headers: Headers,
1351 request: Request,
1352 history: tuple[TestResponse, ...] = (),
1353 **kwargs: t.Any,
1354 ) -> None:
1355 super().__init__(response, status, headers, **kwargs)
1356 self.request = request
1357 self.history = history
1358 self._compat_tuple = response, status, headers
1360 @cached_property
1361 def text(self) -> str:
1362 """The response data as text. A shortcut for
1363 ``response.get_data(as_text=True)``.
1365 .. versionadded:: 2.1
1366 """
1367 return self.get_data(as_text=True)
1370@dataclasses.dataclass
1371class Cookie:
1372 """A cookie key, value, and parameters.
1374 The class itself is not a public API. Its attributes are documented for inspection
1375 with :meth:`.Client.get_cookie` only.
1377 .. versionadded:: 2.3
1378 """
1380 key: str
1381 """The cookie key, encoded as a client would see it."""
1383 value: str
1384 """The cookie key, encoded as a client would see it."""
1386 decoded_key: str
1387 """The cookie key, decoded as the application would set and see it."""
1389 decoded_value: str
1390 """The cookie value, decoded as the application would set and see it."""
1392 expires: datetime | None
1393 """The time at which the cookie is no longer valid."""
1395 max_age: int | None
1396 """The number of seconds from when the cookie was set at which it is
1397 no longer valid.
1398 """
1400 domain: str
1401 """The domain that the cookie was set for, or the request domain if not set."""
1403 origin_only: bool
1404 """Whether the cookie will be sent for exact domain matches only. This is ``True``
1405 if the ``Domain`` parameter was not present.
1406 """
1408 path: str
1409 """The path that the cookie was set for."""
1411 secure: bool | None
1412 """The ``Secure`` parameter."""
1414 http_only: bool | None
1415 """The ``HttpOnly`` parameter."""
1417 same_site: str | None
1418 """The ``SameSite`` parameter."""
1420 def _matches_request(self, server_name: str, path: str) -> bool:
1421 return (
1422 server_name == self.domain
1423 or (
1424 not self.origin_only
1425 and server_name.endswith(self.domain)
1426 and server_name[: -len(self.domain)].endswith(".")
1427 )
1428 ) and (
1429 path == self.path
1430 or (
1431 path.startswith(self.path)
1432 and path[len(self.path) - self.path.endswith("/") :].startswith("/")
1433 )
1434 )
1436 def _to_request_header(self) -> str:
1437 return f"{self.key}={self.value}"
1439 @classmethod
1440 def _from_response_header(cls, server_name: str, path: str, header: str) -> t.Self:
1441 header, _, parameters_str = header.partition(";")
1442 key, _, value = header.partition("=")
1443 decoded_key, decoded_value = next(parse_cookie(header).items()) # type: ignore[call-overload]
1444 params = {}
1446 for item in parameters_str.split(";"):
1447 k, sep, v = item.partition("=")
1448 params[k.strip(" \t").lower()] = v.strip(" \t") if sep else None
1450 return cls(
1451 key=key.strip(" \t"),
1452 value=value.strip(" \t"),
1453 decoded_key=decoded_key,
1454 decoded_value=decoded_value,
1455 expires=parse_date(params.get("expires")),
1456 max_age=int(params["max-age"] or 0) if "max-age" in params else None,
1457 domain=params.get("domain") or server_name,
1458 origin_only="domain" not in params,
1459 path=params.get("path") or path.rpartition("/")[0] or "/",
1460 secure="secure" in params,
1461 http_only="httponly" in params,
1462 same_site=params.get("samesite"),
1463 )
1465 @property
1466 def _storage_key(self) -> tuple[str, str, str]:
1467 return self.domain, self.path, self.decoded_key
1469 @property
1470 def _should_delete(self) -> bool:
1471 return self.max_age == 0 or (
1472 self.expires is not None and self.expires.timestamp() == 0
1473 )