Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/werkzeug/test.py: 29%
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 mimetypes
5import sys
6import typing as t
7from collections import defaultdict
8from datetime import datetime
9from io import BytesIO
10from itertools import chain
11from random import random
12from tempfile import TemporaryFile
13from time import time
14from urllib.parse import unquote
15from urllib.parse import urlsplit
16from urllib.parse import urlunsplit
18from ._internal import _get_environ
19from ._internal import _wsgi_decoding_dance
20from ._internal import _wsgi_encoding_dance
21from .datastructures import Authorization
22from .datastructures import CallbackDict
23from .datastructures import CombinedMultiDict
24from .datastructures import EnvironHeaders
25from .datastructures import FileMultiDict
26from .datastructures import Headers
27from .datastructures import MultiDict
28from .http import dump_cookie
29from .http import dump_options_header
30from .http import parse_cookie
31from .http import parse_date
32from .http import parse_options_header
33from .sansio.multipart import Data
34from .sansio.multipart import Epilogue
35from .sansio.multipart import Field
36from .sansio.multipart import File
37from .sansio.multipart import MultipartEncoder
38from .sansio.multipart import Preamble
39from .urls import _urlencode
40from .urls import iri_to_uri
41from .utils import cached_property
42from .utils import get_content_type
43from .wrappers.request import Request
44from .wrappers.response import Response
45from .wsgi import ClosingIterator
46from .wsgi import get_current_url
48if t.TYPE_CHECKING:
49 import typing_extensions as te
50 from _typeshed.wsgi import WSGIApplication
51 from _typeshed.wsgi import WSGIEnvironment
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 or
61 :class:`FileStorage` objects.) into a multipart encoded string stored
62 in a file descriptor.
64 This is not secure and must only be used during testing, never in applications.
66 .. versionchanged:: 3.0
67 The ``charset`` parameter was removed.
68 """
69 if boundary is None:
70 boundary = f"---------------WerkzeugFormPart_{time()}{random()}"
72 stream: t.IO[bytes] = BytesIO()
73 total_length = 0
74 on_disk = False
75 write_binary: t.Callable[[bytes], int]
77 if use_tempfile:
79 def write_binary(s: bytes) -> int:
80 nonlocal stream, total_length, on_disk
82 if on_disk:
83 return stream.write(s)
84 else:
85 length = len(s)
87 if length + total_length <= threshold:
88 stream.write(s)
89 else:
90 new_stream = t.cast(t.IO[bytes], TemporaryFile("wb+"))
91 new_stream.write(stream.getvalue()) # type: ignore
92 new_stream.write(s)
93 stream = new_stream
94 on_disk = True
96 total_length += length
97 return length
99 else:
100 write_binary = stream.write
102 encoder = MultipartEncoder(boundary.encode())
103 write_binary(encoder.send_event(Preamble(data=b"")))
104 for key, value in _iter_data(data):
105 reader = getattr(value, "read", None)
106 if reader is not None:
107 filename = getattr(value, "filename", getattr(value, "name", None))
108 content_type = getattr(value, "content_type", None)
109 if content_type is None:
110 content_type = (
111 filename
112 and mimetypes.guess_type(filename)[0]
113 or "application/octet-stream"
114 )
115 headers = value.headers
116 headers.update([("Content-Type", content_type)])
117 if filename is None:
118 write_binary(encoder.send_event(Field(name=key, headers=headers)))
119 else:
120 write_binary(
121 encoder.send_event(
122 File(name=key, filename=filename, headers=headers)
123 )
124 )
125 while True:
126 chunk = reader(16384)
128 if not chunk:
129 write_binary(encoder.send_event(Data(data=chunk, more_data=False)))
130 break
132 write_binary(encoder.send_event(Data(data=chunk, more_data=True)))
133 else:
134 if not isinstance(value, str):
135 value = str(value)
136 write_binary(encoder.send_event(Field(name=key, headers=Headers())))
137 write_binary(encoder.send_event(Data(data=value.encode(), more_data=False)))
139 write_binary(encoder.send_event(Epilogue(data=b"")))
141 length = stream.tell()
142 stream.seek(0)
143 return stream, length, boundary
146def encode_multipart(
147 values: t.Mapping[str, t.Any], boundary: str | None = None
148) -> tuple[str, bytes]:
149 """Like `stream_encode_multipart` but returns a tuple in the form
150 (``boundary``, ``data``) where data is bytes.
152 This is not secure and must only be used during testing, never in applications.
154 .. versionchanged:: 3.0
155 The ``charset`` parameter was removed.
156 """
157 stream, length, boundary = stream_encode_multipart(
158 values, use_tempfile=False, boundary=boundary
159 )
160 return boundary, stream.read()
163def _iter_data(data: t.Mapping[str, t.Any]) -> t.Iterator[tuple[str, t.Any]]:
164 """Iterate over a mapping that might have a list of values, yielding
165 all key, value pairs. Almost like iter_multi_items but only allows
166 lists, not tuples, of values so tuples can be used for files.
167 """
168 if isinstance(data, MultiDict):
169 yield from data.items(multi=True)
170 else:
171 for key, value in data.items():
172 if isinstance(value, list):
173 for v in value:
174 yield key, v
175 else:
176 yield key, value
179_TAnyMultiDict = t.TypeVar("_TAnyMultiDict", bound="MultiDict[t.Any, t.Any]")
182class EnvironBuilder:
183 """This class can be used to conveniently create a WSGI environment
184 for testing purposes. It can be used to quickly create WSGI environments
185 or request objects from arbitrary data.
187 The signature of this class is also used in some other places as of
188 Werkzeug 0.5 (:func:`create_environ`, :meth:`Response.from_values`,
189 :meth:`Client.open`). Because of this most of the functionality is
190 available through the constructor alone.
192 Files and regular form data can be manipulated independently of each
193 other with the :attr:`form` and :attr:`files` attributes, but are
194 passed with the same argument to the constructor: `data`.
196 `data` can be any of these values:
198 - a `str` or `bytes` object: The object is converted into an
199 :attr:`input_stream`, the :attr:`content_length` is set and you have to
200 provide a :attr:`content_type`.
201 - a `dict` or :class:`MultiDict`: The keys have to be strings. The values
202 have to be either any of the following objects, or a list of any of the
203 following objects:
205 - a :class:`file`-like object: These are converted into
206 :class:`FileStorage` objects automatically.
207 - a `tuple`: The :meth:`~FileMultiDict.add_file` method is called
208 with the key and the unpacked `tuple` items as positional
209 arguments.
210 - a `str`: The string is set as form data for the associated key.
211 - a file-like object: The object content is loaded in memory and then
212 handled like a regular `str` or a `bytes`.
214 :param path: the path of the request. In the WSGI environment this will
215 end up as `PATH_INFO`. If the `query_string` is not defined
216 and there is a question mark in the `path` everything after
217 it is used as query string.
218 :param base_url: the base URL is a URL that is used to extract the WSGI
219 URL scheme, host (server name + server port) and the
220 script root (`SCRIPT_NAME`).
221 :param query_string: an optional string or dict with URL parameters.
222 :param method: the HTTP method to use, defaults to `GET`.
223 :param input_stream: an optional input stream. Do not specify this and
224 `data`. As soon as an input stream is set you can't
225 modify :attr:`args` and :attr:`files` unless you
226 set the :attr:`input_stream` to `None` again.
227 :param content_type: The content type for the request. As of 0.5 you
228 don't have to provide this when specifying files
229 and form data via `data`.
230 :param content_length: The content length for the request. You don't
231 have to specify this when providing data via
232 `data`.
233 :param errors_stream: an optional error stream that is used for
234 `wsgi.errors`. Defaults to :data:`stderr`.
235 :param multithread: controls `wsgi.multithread`. Defaults to `False`.
236 :param multiprocess: controls `wsgi.multiprocess`. Defaults to `False`.
237 :param run_once: controls `wsgi.run_once`. Defaults to `False`.
238 :param headers: an optional list or :class:`Headers` object of headers.
239 :param data: a string or dict of form data or a file-object.
240 See explanation above.
241 :param json: An object to be serialized and assigned to ``data``.
242 Defaults the content type to ``"application/json"``.
243 Serialized with the function assigned to :attr:`json_dumps`.
244 :param environ_base: an optional dict of environment defaults.
245 :param environ_overrides: an optional dict of environment overrides.
246 :param auth: An authorization object to use for the
247 ``Authorization`` header value. A ``(username, password)`` tuple
248 is a shortcut for ``Basic`` authorization.
250 .. versionchanged:: 3.0
251 The ``charset`` parameter was removed.
253 .. versionchanged:: 2.1
254 ``CONTENT_TYPE`` and ``CONTENT_LENGTH`` are not duplicated as
255 header keys in the environ.
257 .. versionchanged:: 2.0
258 ``REQUEST_URI`` and ``RAW_URI`` is the full raw URI including
259 the query string, not only the path.
261 .. versionchanged:: 2.0
262 The default :attr:`request_class` is ``Request`` instead of
263 ``BaseRequest``.
265 .. versionadded:: 2.0
266 Added the ``auth`` parameter.
268 .. versionadded:: 0.15
269 The ``json`` param and :meth:`json_dumps` method.
271 .. versionadded:: 0.15
272 The environ has keys ``REQUEST_URI`` and ``RAW_URI`` containing
273 the path before percent-decoding. This is not part of the WSGI
274 PEP, but many WSGI servers include it.
276 .. versionchanged:: 0.6
277 ``path`` and ``base_url`` can now be unicode strings that are
278 encoded with :func:`iri_to_uri`.
279 """
281 #: the server protocol to use. defaults to HTTP/1.1
282 server_protocol = "HTTP/1.1"
284 #: the wsgi version to use. defaults to (1, 0)
285 wsgi_version = (1, 0)
287 #: The default request class used by :meth:`get_request`.
288 request_class = Request
290 import json
292 #: The serialization function used when ``json`` is passed.
293 json_dumps = staticmethod(json.dumps)
294 del json
296 _args: MultiDict[str, str] | None
297 _query_string: str | None
298 _input_stream: t.IO[bytes] | None
299 _form: MultiDict[str, str] | None
300 _files: FileMultiDict | None
302 def __init__(
303 self,
304 path: str = "/",
305 base_url: str | None = None,
306 query_string: t.Mapping[str, str] | str | None = None,
307 method: str = "GET",
308 input_stream: t.IO[bytes] | None = None,
309 content_type: str | None = None,
310 content_length: int | None = None,
311 errors_stream: t.IO[str] | None = None,
312 multithread: bool = False,
313 multiprocess: bool = False,
314 run_once: bool = False,
315 headers: Headers | t.Iterable[tuple[str, str]] | None = None,
316 data: None | (t.IO[bytes] | str | bytes | t.Mapping[str, t.Any]) = None,
317 environ_base: t.Mapping[str, t.Any] | None = None,
318 environ_overrides: t.Mapping[str, t.Any] | None = None,
319 mimetype: str | None = None,
320 json: t.Mapping[str, t.Any] | None = None,
321 auth: Authorization | tuple[str, str] | None = None,
322 ) -> None:
323 if query_string is not None and "?" in path:
324 raise ValueError("Query string is defined in the path and as an argument")
325 request_uri = urlsplit(path)
326 if query_string is None and "?" in path:
327 query_string = request_uri.query
329 self.path = iri_to_uri(request_uri.path)
330 self.request_uri = path
331 if base_url is not None:
332 base_url = iri_to_uri(base_url)
333 self.base_url = base_url
334 if isinstance(query_string, str):
335 self.query_string = query_string
336 else:
337 if query_string is None:
338 query_string = MultiDict()
339 elif not isinstance(query_string, MultiDict):
340 query_string = MultiDict(query_string)
341 self.args = query_string
342 self.method = method
343 if headers is None:
344 headers = Headers()
345 elif not isinstance(headers, Headers):
346 headers = Headers(headers)
347 self.headers = headers
348 if content_type is not None:
349 self.content_type = content_type
350 if errors_stream is None:
351 errors_stream = sys.stderr
352 self.errors_stream = errors_stream
353 self.multithread = multithread
354 self.multiprocess = multiprocess
355 self.run_once = run_once
356 self.environ_base = environ_base
357 self.environ_overrides = environ_overrides
358 self.input_stream = input_stream
359 self.content_length = content_length
360 self.closed = False
362 if auth is not None:
363 if isinstance(auth, tuple):
364 auth = Authorization(
365 "basic", {"username": auth[0], "password": auth[1]}
366 )
368 self.headers.set("Authorization", auth.to_header())
370 if json is not None:
371 if data is not None:
372 raise TypeError("can't provide both json and data")
374 data = self.json_dumps(json)
376 if self.content_type is None:
377 self.content_type = "application/json"
379 if data:
380 if input_stream is not None:
381 raise TypeError("can't provide input stream and data")
382 if hasattr(data, "read"):
383 data = data.read()
384 if isinstance(data, str):
385 data = data.encode()
386 if isinstance(data, bytes):
387 self.input_stream = BytesIO(data)
388 if self.content_length is None:
389 self.content_length = len(data)
390 else:
391 for key, value in _iter_data(data):
392 if isinstance(value, (tuple, dict)) or hasattr(value, "read"):
393 self._add_file_from_data(key, value)
394 else:
395 self.form.setlistdefault(key).append(value)
397 if mimetype is not None:
398 self.mimetype = mimetype
400 @classmethod
401 def from_environ(cls, environ: WSGIEnvironment, **kwargs: t.Any) -> EnvironBuilder:
402 """Turn an environ dict back into a builder. Any extra kwargs
403 override the args extracted from the environ.
405 .. versionchanged:: 2.0
406 Path and query values are passed through the WSGI decoding
407 dance to avoid double encoding.
409 .. versionadded:: 0.15
410 """
411 headers = Headers(EnvironHeaders(environ))
412 out = {
413 "path": _wsgi_decoding_dance(environ["PATH_INFO"]),
414 "base_url": cls._make_base_url(
415 environ["wsgi.url_scheme"],
416 headers.pop("Host"),
417 _wsgi_decoding_dance(environ["SCRIPT_NAME"]),
418 ),
419 "query_string": _wsgi_decoding_dance(environ["QUERY_STRING"]),
420 "method": environ["REQUEST_METHOD"],
421 "input_stream": environ["wsgi.input"],
422 "content_type": headers.pop("Content-Type", None),
423 "content_length": headers.pop("Content-Length", None),
424 "errors_stream": environ["wsgi.errors"],
425 "multithread": environ["wsgi.multithread"],
426 "multiprocess": environ["wsgi.multiprocess"],
427 "run_once": environ["wsgi.run_once"],
428 "headers": headers,
429 }
430 out.update(kwargs)
431 return cls(**out)
433 def _add_file_from_data(
434 self,
435 key: str,
436 value: (t.IO[bytes] | tuple[t.IO[bytes], str] | tuple[t.IO[bytes], str, str]),
437 ) -> None:
438 """Called in the EnvironBuilder to add files from the data dict."""
439 if isinstance(value, tuple):
440 self.files.add_file(key, *value)
441 else:
442 self.files.add_file(key, value)
444 @staticmethod
445 def _make_base_url(scheme: str, host: str, script_root: str) -> str:
446 return urlunsplit((scheme, host, script_root, "", "")).rstrip("/") + "/"
448 @property
449 def base_url(self) -> str:
450 """The base URL is used to extract the URL scheme, host name,
451 port, and root path.
452 """
453 return self._make_base_url(self.url_scheme, self.host, self.script_root)
455 @base_url.setter
456 def base_url(self, value: str | None) -> None:
457 if value is None:
458 scheme = "http"
459 netloc = "localhost"
460 script_root = ""
461 else:
462 scheme, netloc, script_root, qs, anchor = urlsplit(value)
463 if qs or anchor:
464 raise ValueError("base url must not contain a query string or fragment")
465 self.script_root = script_root.rstrip("/")
466 self.host = netloc
467 self.url_scheme = scheme
469 @property
470 def content_type(self) -> str | None:
471 """The content type for the request. Reflected from and to
472 the :attr:`headers`. Do not set if you set :attr:`files` or
473 :attr:`form` for auto detection.
474 """
475 ct = self.headers.get("Content-Type")
476 if ct is None and not self._input_stream:
477 if self._files:
478 return "multipart/form-data"
479 if self._form:
480 return "application/x-www-form-urlencoded"
481 return None
482 return ct
484 @content_type.setter
485 def content_type(self, value: str | None) -> None:
486 if value is None:
487 self.headers.pop("Content-Type", None)
488 else:
489 self.headers["Content-Type"] = value
491 @property
492 def mimetype(self) -> str | None:
493 """The mimetype (content type without charset etc.)
495 .. versionadded:: 0.14
496 """
497 ct = self.content_type
498 return ct.partition(";")[0].strip() if ct else None
500 @mimetype.setter
501 def mimetype(self, value: str) -> None:
502 self.content_type = get_content_type(value, "utf-8")
504 @property
505 def mimetype_params(self) -> t.Mapping[str, str]:
506 """The mimetype parameters as dict. For example if the
507 content type is ``text/html; charset=utf-8`` the params would be
508 ``{'charset': 'utf-8'}``.
510 .. versionadded:: 0.14
511 """
513 def on_update(d: CallbackDict[str, str]) -> None:
514 self.headers["Content-Type"] = dump_options_header(self.mimetype, d)
516 d = parse_options_header(self.headers.get("content-type", ""))[1]
517 return CallbackDict(d, on_update)
519 @property
520 def content_length(self) -> int | None:
521 """The content length as integer. Reflected from and to the
522 :attr:`headers`. Do not set if you set :attr:`files` or
523 :attr:`form` for auto detection.
524 """
525 return self.headers.get("Content-Length", type=int)
527 @content_length.setter
528 def content_length(self, value: int | None) -> None:
529 if value is None:
530 self.headers.pop("Content-Length", None)
531 else:
532 self.headers["Content-Length"] = str(value)
534 def _get_form(self, name: str, storage: type[_TAnyMultiDict]) -> _TAnyMultiDict:
535 """Common behavior for getting the :attr:`form` and
536 :attr:`files` properties.
538 :param name: Name of the internal cached attribute.
539 :param storage: Storage class used for the data.
540 """
541 if self.input_stream is not None:
542 raise AttributeError("an input stream is defined")
544 rv = getattr(self, name)
546 if rv is None:
547 rv = storage()
548 setattr(self, name, rv)
550 return rv # type: ignore
552 def _set_form(self, name: str, value: MultiDict[str, t.Any]) -> None:
553 """Common behavior for setting the :attr:`form` and
554 :attr:`files` properties.
556 :param name: Name of the internal cached attribute.
557 :param value: Value to assign to the attribute.
558 """
559 self._input_stream = None
560 setattr(self, name, value)
562 @property
563 def form(self) -> MultiDict[str, str]:
564 """A :class:`MultiDict` of form values."""
565 return self._get_form("_form", MultiDict)
567 @form.setter
568 def form(self, value: MultiDict[str, str]) -> None:
569 self._set_form("_form", value)
571 @property
572 def files(self) -> FileMultiDict:
573 """A :class:`FileMultiDict` of uploaded files. Use
574 :meth:`~FileMultiDict.add_file` to add new files.
575 """
576 return self._get_form("_files", FileMultiDict)
578 @files.setter
579 def files(self, value: FileMultiDict) -> None:
580 self._set_form("_files", value)
582 @property
583 def input_stream(self) -> t.IO[bytes] | None:
584 """An optional input stream. This is mutually exclusive with
585 setting :attr:`form` and :attr:`files`, setting it will clear
586 those. Do not provide this if the method is not ``POST`` or
587 another method that has a body.
588 """
589 return self._input_stream
591 @input_stream.setter
592 def input_stream(self, value: t.IO[bytes] | None) -> None:
593 self._input_stream = value
594 self._form = None
595 self._files = None
597 @property
598 def query_string(self) -> str:
599 """The query string. If you set this to a string
600 :attr:`args` will no longer be available.
601 """
602 if self._query_string is None:
603 if self._args is not None:
604 return _urlencode(self._args)
605 return ""
606 return self._query_string
608 @query_string.setter
609 def query_string(self, value: str | None) -> None:
610 self._query_string = value
611 self._args = None
613 @property
614 def args(self) -> MultiDict[str, str]:
615 """The URL arguments as :class:`MultiDict`."""
616 if self._query_string is not None:
617 raise AttributeError("a query string is defined")
618 if self._args is None:
619 self._args = MultiDict()
620 return self._args
622 @args.setter
623 def args(self, value: MultiDict[str, str] | None) -> None:
624 self._query_string = None
625 self._args = value
627 @property
628 def server_name(self) -> str:
629 """The server name (read-only, use :attr:`host` to set)"""
630 return self.host.partition(":")[0]
632 @property
633 def server_port(self) -> int:
634 """The server port as integer (read-only, use :attr:`host` to set)"""
635 _, sep, port = self.host.partition(":")
637 if sep:
638 try:
639 return int(port)
640 except ValueError:
641 pass
643 if self.url_scheme == "https":
644 return 443
645 return 80
647 def __del__(self) -> None:
648 try:
649 self.close()
650 except Exception:
651 pass
653 def close(self) -> None:
654 """Closes all files. If you put real :class:`file` objects into the
655 :attr:`files` dict you can call this method to automatically close
656 them all in one go.
657 """
658 if self.closed:
659 return
660 try:
661 files = self.files.values()
662 except AttributeError:
663 files = ()
664 for f in files:
665 try:
666 f.close()
667 except Exception:
668 pass
669 self.closed = True
671 def get_environ(self) -> WSGIEnvironment:
672 """Return the built environ.
674 .. versionchanged:: 0.15
675 The content type and length headers are set based on
676 input stream detection. Previously this only set the WSGI
677 keys.
678 """
679 input_stream = self.input_stream
680 content_length = self.content_length
682 mimetype = self.mimetype
683 content_type = self.content_type
685 if input_stream is not None:
686 start_pos = input_stream.tell()
687 input_stream.seek(0, 2)
688 end_pos = input_stream.tell()
689 input_stream.seek(start_pos)
690 content_length = end_pos - start_pos
691 elif mimetype == "multipart/form-data":
692 input_stream, content_length, boundary = stream_encode_multipart(
693 CombinedMultiDict([self.form, self.files])
694 )
695 content_type = f'{mimetype}; boundary="{boundary}"'
696 elif mimetype == "application/x-www-form-urlencoded":
697 form_encoded = _urlencode(self.form).encode("ascii")
698 content_length = len(form_encoded)
699 input_stream = BytesIO(form_encoded)
700 else:
701 input_stream = BytesIO()
703 result: WSGIEnvironment = {}
704 if self.environ_base:
705 result.update(self.environ_base)
707 def _path_encode(x: str) -> str:
708 return _wsgi_encoding_dance(unquote(x))
710 raw_uri = _wsgi_encoding_dance(self.request_uri)
711 result.update(
712 {
713 "REQUEST_METHOD": self.method,
714 "SCRIPT_NAME": _path_encode(self.script_root),
715 "PATH_INFO": _path_encode(self.path),
716 "QUERY_STRING": _wsgi_encoding_dance(self.query_string),
717 # Non-standard, added by mod_wsgi, uWSGI
718 "REQUEST_URI": raw_uri,
719 # Non-standard, added by gunicorn
720 "RAW_URI": raw_uri,
721 "SERVER_NAME": self.server_name,
722 "SERVER_PORT": str(self.server_port),
723 "HTTP_HOST": self.host,
724 "SERVER_PROTOCOL": self.server_protocol,
725 "wsgi.version": self.wsgi_version,
726 "wsgi.url_scheme": self.url_scheme,
727 "wsgi.input": input_stream,
728 "wsgi.errors": self.errors_stream,
729 "wsgi.multithread": self.multithread,
730 "wsgi.multiprocess": self.multiprocess,
731 "wsgi.run_once": self.run_once,
732 }
733 )
735 headers = self.headers.copy()
736 # Don't send these as headers, they're part of the environ.
737 headers.remove("Content-Type")
738 headers.remove("Content-Length")
740 if content_type is not None:
741 result["CONTENT_TYPE"] = content_type
743 if content_length is not None:
744 result["CONTENT_LENGTH"] = str(content_length)
746 combined_headers = defaultdict(list)
748 for key, value in headers.to_wsgi_list():
749 combined_headers[f"HTTP_{key.upper().replace('-', '_')}"].append(value)
751 for key, values in combined_headers.items():
752 result[key] = ", ".join(values)
754 if self.environ_overrides:
755 result.update(self.environ_overrides)
757 return result
759 def get_request(self, cls: type[Request] | None = None) -> Request:
760 """Returns a request with the data. If the request class is not
761 specified :attr:`request_class` is used.
763 :param cls: The request wrapper to use.
764 """
765 if cls is None:
766 cls = self.request_class
768 return cls(self.get_environ())
771class ClientRedirectError(Exception):
772 """If a redirect loop is detected when using follow_redirects=True with
773 the :cls:`Client`, then this exception is raised.
774 """
777class Client:
778 """Simulate sending requests to a WSGI application without running a WSGI or HTTP
779 server.
781 :param application: The WSGI application to make requests to.
782 :param response_wrapper: A :class:`.Response` class to wrap response data with.
783 Defaults to :class:`.TestResponse`. If it's not a subclass of ``TestResponse``,
784 one will be created.
785 :param use_cookies: Persist cookies from ``Set-Cookie`` response headers to the
786 ``Cookie`` header in subsequent requests. Domain and path matching is supported,
787 but other cookie parameters are ignored.
788 :param allow_subdomain_redirects: Allow requests to follow redirects to subdomains.
789 Enable this if the application handles subdomains and redirects between them.
791 .. versionchanged:: 2.3
792 Simplify cookie implementation, support domain and path matching.
794 .. versionchanged:: 2.1
795 All data is available as properties on the returned response object. The
796 response cannot be returned as a tuple.
798 .. versionchanged:: 2.0
799 ``response_wrapper`` is always a subclass of :class:``TestResponse``.
801 .. versionchanged:: 0.5
802 Added the ``use_cookies`` parameter.
803 """
805 def __init__(
806 self,
807 application: WSGIApplication,
808 response_wrapper: type[Response] | None = None,
809 use_cookies: bool = True,
810 allow_subdomain_redirects: bool = False,
811 ) -> None:
812 self.application = application
814 if response_wrapper in {None, Response}:
815 response_wrapper = TestResponse
816 elif response_wrapper is not None and not issubclass(
817 response_wrapper, TestResponse
818 ):
819 response_wrapper = type(
820 "WrapperTestResponse",
821 (TestResponse, response_wrapper),
822 {},
823 )
825 self.response_wrapper = t.cast(type["TestResponse"], response_wrapper)
827 if use_cookies:
828 self._cookies: dict[tuple[str, str, str], Cookie] | None = {}
829 else:
830 self._cookies = None
832 self.allow_subdomain_redirects = allow_subdomain_redirects
834 def get_cookie(
835 self, key: str, domain: str = "localhost", path: str = "/"
836 ) -> Cookie | None:
837 """Return a :class:`.Cookie` if it exists. Cookies are uniquely identified by
838 ``(domain, path, key)``.
840 :param key: The decoded form of the key for the cookie.
841 :param domain: The domain the cookie was set for.
842 :param path: The path the cookie was set for.
844 .. versionadded:: 2.3
845 """
846 if self._cookies is None:
847 raise TypeError(
848 "Cookies are disabled. Create a client with 'use_cookies=True'."
849 )
851 return self._cookies.get((domain, path, key))
853 def set_cookie(
854 self,
855 key: str,
856 value: str = "",
857 *,
858 domain: str = "localhost",
859 origin_only: bool = True,
860 path: str = "/",
861 **kwargs: t.Any,
862 ) -> None:
863 """Set a cookie to be sent in subsequent requests.
865 This is a convenience to skip making a test request to a route that would set
866 the cookie. To test the cookie, make a test request to a route that uses the
867 cookie value.
869 The client uses ``domain``, ``origin_only``, and ``path`` to determine which
870 cookies to send with a request. It does not use other cookie parameters that
871 browsers use, since they're not applicable in tests.
873 :param key: The key part of the cookie.
874 :param value: The value part of the cookie.
875 :param domain: Send this cookie with requests that match this domain. If
876 ``origin_only`` is true, it must be an exact match, otherwise it may be a
877 suffix match.
878 :param origin_only: Whether the domain must be an exact match to the request.
879 :param path: Send this cookie with requests that match this path either exactly
880 or as a prefix.
881 :param kwargs: Passed to :func:`.dump_cookie`.
883 .. versionchanged:: 3.0
884 The parameter ``server_name`` is removed. The first parameter is
885 ``key``. Use the ``domain`` and ``origin_only`` parameters instead.
887 .. versionchanged:: 2.3
888 The ``origin_only`` parameter was added.
890 .. versionchanged:: 2.3
891 The ``domain`` parameter defaults to ``localhost``.
892 """
893 if self._cookies is None:
894 raise TypeError(
895 "Cookies are disabled. Create a client with 'use_cookies=True'."
896 )
898 cookie = Cookie._from_response_header(
899 domain, "/", dump_cookie(key, value, domain=domain, path=path, **kwargs)
900 )
901 cookie.origin_only = origin_only
903 if cookie._should_delete:
904 self._cookies.pop(cookie._storage_key, None)
905 else:
906 self._cookies[cookie._storage_key] = cookie
908 def delete_cookie(
909 self,
910 key: str,
911 *,
912 domain: str = "localhost",
913 path: str = "/",
914 ) -> None:
915 """Delete a cookie if it exists. Cookies are uniquely identified by
916 ``(domain, path, key)``.
918 :param key: The decoded form of the key for the cookie.
919 :param domain: The domain the cookie was set for.
920 :param path: The path the cookie was set for.
922 .. versionchanged:: 3.0
923 The ``server_name`` parameter is removed. The first parameter is
924 ``key``. Use the ``domain`` parameter instead.
926 .. versionchanged:: 3.0
927 The ``secure``, ``httponly`` and ``samesite`` parameters are removed.
929 .. versionchanged:: 2.3
930 The ``domain`` parameter defaults to ``localhost``.
931 """
932 if self._cookies is None:
933 raise TypeError(
934 "Cookies are disabled. Create a client with 'use_cookies=True'."
935 )
937 self._cookies.pop((domain, path, key), None)
939 def _add_cookies_to_wsgi(self, environ: WSGIEnvironment) -> None:
940 """If cookies are enabled, set the ``Cookie`` header in the environ to the
941 cookies that are applicable to the request host and path.
943 :meta private:
945 .. versionadded:: 2.3
946 """
947 if self._cookies is None:
948 return
950 url = urlsplit(get_current_url(environ))
951 server_name = url.hostname or "localhost"
952 value = "; ".join(
953 c._to_request_header()
954 for c in self._cookies.values()
955 if c._matches_request(server_name, url.path)
956 )
958 if value:
959 environ["HTTP_COOKIE"] = value
960 else:
961 environ.pop("HTTP_COOKIE", None)
963 def _update_cookies_from_response(
964 self, server_name: str, path: str, headers: list[str]
965 ) -> None:
966 """If cookies are enabled, update the stored cookies from any ``Set-Cookie``
967 headers in the response.
969 :meta private:
971 .. versionadded:: 2.3
972 """
973 if self._cookies is None:
974 return
976 for header in headers:
977 cookie = Cookie._from_response_header(server_name, path, header)
979 if cookie._should_delete:
980 self._cookies.pop(cookie._storage_key, None)
981 else:
982 self._cookies[cookie._storage_key] = cookie
984 def run_wsgi_app(
985 self, environ: WSGIEnvironment, buffered: bool = False
986 ) -> tuple[t.Iterable[bytes], str, Headers]:
987 """Runs the wrapped WSGI app with the given environment.
989 :meta private:
990 """
991 self._add_cookies_to_wsgi(environ)
992 rv = run_wsgi_app(self.application, environ, buffered=buffered)
993 url = urlsplit(get_current_url(environ))
994 self._update_cookies_from_response(
995 url.hostname or "localhost", url.path, rv[2].getlist("Set-Cookie")
996 )
997 return rv
999 def resolve_redirect(
1000 self, response: TestResponse, buffered: bool = False
1001 ) -> TestResponse:
1002 """Perform a new request to the location given by the redirect
1003 response to the previous request.
1005 :meta private:
1006 """
1007 scheme, netloc, path, qs, anchor = urlsplit(response.location)
1008 builder = EnvironBuilder.from_environ(
1009 response.request.environ, path=path, query_string=qs
1010 )
1012 to_name_parts = netloc.partition(":")[0].split(".")
1013 from_name_parts = builder.server_name.split(".")
1015 if to_name_parts != [""]:
1016 # The new location has a host, use it for the base URL.
1017 builder.url_scheme = scheme
1018 builder.host = netloc
1019 else:
1020 # A local redirect with autocorrect_location_header=False
1021 # doesn't have a host, so use the request's host.
1022 to_name_parts = from_name_parts
1024 # Explain why a redirect to a different server name won't be followed.
1025 if to_name_parts != from_name_parts:
1026 if to_name_parts[-len(from_name_parts) :] == from_name_parts:
1027 if not self.allow_subdomain_redirects:
1028 raise RuntimeError("Following subdomain redirects is not enabled.")
1029 else:
1030 raise RuntimeError("Following external redirects is not supported.")
1032 path_parts = path.split("/")
1033 root_parts = builder.script_root.split("/")
1035 if path_parts[: len(root_parts)] == root_parts:
1036 # Strip the script root from the path.
1037 builder.path = path[len(builder.script_root) :]
1038 else:
1039 # The new location is not under the script root, so use the
1040 # whole path and clear the previous root.
1041 builder.path = path
1042 builder.script_root = ""
1044 # Only 307 and 308 preserve all of the original request.
1045 if response.status_code not in {307, 308}:
1046 # HEAD is preserved, everything else becomes GET.
1047 if builder.method != "HEAD":
1048 builder.method = "GET"
1050 # Clear the body and the headers that describe it.
1052 if builder.input_stream is not None:
1053 builder.input_stream.close()
1054 builder.input_stream = None
1056 builder.content_type = None
1057 builder.content_length = None
1058 builder.headers.pop("Transfer-Encoding", None)
1060 return self.open(builder, buffered=buffered)
1062 def open(
1063 self,
1064 *args: t.Any,
1065 buffered: bool = False,
1066 follow_redirects: bool = False,
1067 **kwargs: t.Any,
1068 ) -> TestResponse:
1069 """Generate an environ dict from the given arguments, make a
1070 request to the application using it, and return the response.
1072 :param args: Passed to :class:`EnvironBuilder` to create the
1073 environ for the request. If a single arg is passed, it can
1074 be an existing :class:`EnvironBuilder` or an environ dict.
1075 :param buffered: Convert the iterator returned by the app into
1076 a list. If the iterator has a ``close()`` method, it is
1077 called automatically.
1078 :param follow_redirects: Make additional requests to follow HTTP
1079 redirects until a non-redirect status is returned.
1080 :attr:`TestResponse.history` lists the intermediate
1081 responses.
1083 .. versionchanged:: 2.1
1084 Removed the ``as_tuple`` parameter.
1086 .. versionchanged:: 2.0
1087 The request input stream is closed when calling
1088 ``response.close()``. Input streams for redirects are
1089 automatically closed.
1091 .. versionchanged:: 0.5
1092 If a dict is provided as file in the dict for the ``data``
1093 parameter the content type has to be called ``content_type``
1094 instead of ``mimetype``. This change was made for
1095 consistency with :class:`werkzeug.FileWrapper`.
1097 .. versionchanged:: 0.5
1098 Added the ``follow_redirects`` parameter.
1099 """
1100 request: Request | None = None
1102 if not kwargs and len(args) == 1:
1103 arg = args[0]
1105 if isinstance(arg, EnvironBuilder):
1106 request = arg.get_request()
1107 elif isinstance(arg, dict):
1108 request = EnvironBuilder.from_environ(arg).get_request()
1109 elif isinstance(arg, Request):
1110 request = arg
1112 if request is None:
1113 builder = EnvironBuilder(*args, **kwargs)
1115 try:
1116 request = builder.get_request()
1117 finally:
1118 builder.close()
1120 response_parts = self.run_wsgi_app(request.environ, buffered=buffered)
1121 response = self.response_wrapper(*response_parts, request=request)
1123 redirects = set()
1124 history: list[TestResponse] = []
1126 if not follow_redirects:
1127 return response
1129 while response.status_code in {
1130 301,
1131 302,
1132 303,
1133 305,
1134 307,
1135 308,
1136 }:
1137 # Exhaust intermediate response bodies to ensure middleware
1138 # that returns an iterator runs any cleanup code.
1139 if not buffered:
1140 response.make_sequence()
1141 response.close()
1143 new_redirect_entry = (response.location, response.status_code)
1145 if new_redirect_entry in redirects:
1146 raise ClientRedirectError(
1147 f"Loop detected: A {response.status_code} redirect"
1148 f" to {response.location} was already made."
1149 )
1151 redirects.add(new_redirect_entry)
1152 response.history = tuple(history)
1153 history.append(response)
1154 response = self.resolve_redirect(response, buffered=buffered)
1155 else:
1156 # This is the final request after redirects.
1157 response.history = tuple(history)
1158 # Close the input stream when closing the response, in case
1159 # the input is an open temporary file.
1160 response.call_on_close(request.input_stream.close)
1161 return response
1163 def get(self, *args: t.Any, **kw: t.Any) -> TestResponse:
1164 """Call :meth:`open` with ``method`` set to ``GET``."""
1165 kw["method"] = "GET"
1166 return self.open(*args, **kw)
1168 def post(self, *args: t.Any, **kw: t.Any) -> TestResponse:
1169 """Call :meth:`open` with ``method`` set to ``POST``."""
1170 kw["method"] = "POST"
1171 return self.open(*args, **kw)
1173 def put(self, *args: t.Any, **kw: t.Any) -> TestResponse:
1174 """Call :meth:`open` with ``method`` set to ``PUT``."""
1175 kw["method"] = "PUT"
1176 return self.open(*args, **kw)
1178 def delete(self, *args: t.Any, **kw: t.Any) -> TestResponse:
1179 """Call :meth:`open` with ``method`` set to ``DELETE``."""
1180 kw["method"] = "DELETE"
1181 return self.open(*args, **kw)
1183 def patch(self, *args: t.Any, **kw: t.Any) -> TestResponse:
1184 """Call :meth:`open` with ``method`` set to ``PATCH``."""
1185 kw["method"] = "PATCH"
1186 return self.open(*args, **kw)
1188 def options(self, *args: t.Any, **kw: t.Any) -> TestResponse:
1189 """Call :meth:`open` with ``method`` set to ``OPTIONS``."""
1190 kw["method"] = "OPTIONS"
1191 return self.open(*args, **kw)
1193 def head(self, *args: t.Any, **kw: t.Any) -> TestResponse:
1194 """Call :meth:`open` with ``method`` set to ``HEAD``."""
1195 kw["method"] = "HEAD"
1196 return self.open(*args, **kw)
1198 def trace(self, *args: t.Any, **kw: t.Any) -> TestResponse:
1199 """Call :meth:`open` with ``method`` set to ``TRACE``."""
1200 kw["method"] = "TRACE"
1201 return self.open(*args, **kw)
1203 def __repr__(self) -> str:
1204 return f"<{type(self).__name__} {self.application!r}>"
1207def create_environ(*args: t.Any, **kwargs: t.Any) -> WSGIEnvironment:
1208 """Create a new WSGI environ dict based on the values passed. The first
1209 parameter should be the path of the request which defaults to '/'. The
1210 second one can either be an absolute path (in that case the host is
1211 localhost:80) or a full path to the request with scheme, netloc port and
1212 the path to the script.
1214 This accepts the same arguments as the :class:`EnvironBuilder`
1215 constructor.
1217 .. versionchanged:: 0.5
1218 This function is now a thin wrapper over :class:`EnvironBuilder` which
1219 was added in 0.5. The `headers`, `environ_base`, `environ_overrides`
1220 and `charset` parameters were added.
1221 """
1222 builder = EnvironBuilder(*args, **kwargs)
1224 try:
1225 return builder.get_environ()
1226 finally:
1227 builder.close()
1230def run_wsgi_app(
1231 app: WSGIApplication, environ: WSGIEnvironment, buffered: bool = False
1232) -> tuple[t.Iterable[bytes], str, Headers]:
1233 """Return a tuple in the form (app_iter, status, headers) of the
1234 application output. This works best if you pass it an application that
1235 returns an iterator all the time.
1237 Sometimes applications may use the `write()` callable returned
1238 by the `start_response` function. This tries to resolve such edge
1239 cases automatically. But if you don't get the expected output you
1240 should set `buffered` to `True` which enforces buffering.
1242 If passed an invalid WSGI application the behavior of this function is
1243 undefined. Never pass non-conforming WSGI applications to this function.
1245 :param app: the application to execute.
1246 :param buffered: set to `True` to enforce buffering.
1247 :return: tuple in the form ``(app_iter, status, headers)``
1248 """
1249 # Copy environ to ensure any mutations by the app (ProxyFix, for
1250 # example) don't affect subsequent requests (such as redirects).
1251 environ = _get_environ(environ).copy()
1252 status: str
1253 response: tuple[str, list[tuple[str, str]]] | None = None
1254 buffer: list[bytes] = []
1256 def start_response(status, headers, exc_info=None): # type: ignore
1257 nonlocal response
1259 if exc_info:
1260 try:
1261 raise exc_info[1].with_traceback(exc_info[2])
1262 finally:
1263 exc_info = None
1265 response = (status, headers)
1266 return buffer.append
1268 app_rv = app(environ, start_response)
1269 close_func = getattr(app_rv, "close", None)
1270 app_iter: t.Iterable[bytes] = iter(app_rv)
1272 # when buffering we emit the close call early and convert the
1273 # application iterator into a regular list
1274 if buffered:
1275 try:
1276 app_iter = list(app_iter)
1277 finally:
1278 if close_func is not None:
1279 close_func()
1281 # otherwise we iterate the application iter until we have a response, chain
1282 # the already received data with the already collected data and wrap it in
1283 # a new `ClosingIterator` if we need to restore a `close` callable from the
1284 # original return value.
1285 else:
1286 for item in app_iter:
1287 buffer.append(item)
1289 if response is not None:
1290 break
1292 if buffer:
1293 app_iter = chain(buffer, app_iter)
1295 if close_func is not None and app_iter is not app_rv:
1296 app_iter = ClosingIterator(app_iter, close_func)
1298 status, headers = response # type: ignore
1299 return app_iter, status, Headers(headers)
1302class TestResponse(Response):
1303 """:class:`~werkzeug.wrappers.Response` subclass that provides extra
1304 information about requests made with the test :class:`Client`.
1306 Test client requests will always return an instance of this class.
1307 If a custom response class is passed to the client, it is
1308 subclassed along with this to support test information.
1310 If the test request included large files, or if the application is
1311 serving a file, call :meth:`close` to close any open files and
1312 prevent Python showing a ``ResourceWarning``.
1314 .. versionchanged:: 2.2
1315 Set the ``default_mimetype`` to None to prevent a mimetype being
1316 assumed if missing.
1318 .. versionchanged:: 2.1
1319 Response instances cannot be treated as tuples.
1321 .. versionadded:: 2.0
1322 Test client methods always return instances of this class.
1323 """
1325 default_mimetype = None
1326 # Don't assume a mimetype, instead use whatever the response provides
1328 request: Request
1329 """A request object with the environ used to make the request that
1330 resulted in this response.
1331 """
1333 history: tuple[TestResponse, ...]
1334 """A list of intermediate responses. Populated when the test request
1335 is made with ``follow_redirects`` enabled.
1336 """
1338 # Tell Pytest to ignore this, it's not a test class.
1339 __test__ = False
1341 def __init__(
1342 self,
1343 response: t.Iterable[bytes],
1344 status: str,
1345 headers: Headers,
1346 request: Request,
1347 history: tuple[TestResponse] = (), # type: ignore
1348 **kwargs: t.Any,
1349 ) -> None:
1350 super().__init__(response, status, headers, **kwargs)
1351 self.request = request
1352 self.history = history
1353 self._compat_tuple = response, status, headers
1355 @cached_property
1356 def text(self) -> str:
1357 """The response data as text. A shortcut for
1358 ``response.get_data(as_text=True)``.
1360 .. versionadded:: 2.1
1361 """
1362 return self.get_data(as_text=True)
1365@dataclasses.dataclass
1366class Cookie:
1367 """A cookie key, value, and parameters.
1369 The class itself is not a public API. Its attributes are documented for inspection
1370 with :meth:`.Client.get_cookie` only.
1372 .. versionadded:: 2.3
1373 """
1375 key: str
1376 """The cookie key, encoded as a client would see it."""
1378 value: str
1379 """The cookie key, encoded as a client would see it."""
1381 decoded_key: str
1382 """The cookie key, decoded as the application would set and see it."""
1384 decoded_value: str
1385 """The cookie value, decoded as the application would set and see it."""
1387 expires: datetime | None
1388 """The time at which the cookie is no longer valid."""
1390 max_age: int | None
1391 """The number of seconds from when the cookie was set at which it is
1392 no longer valid.
1393 """
1395 domain: str
1396 """The domain that the cookie was set for, or the request domain if not set."""
1398 origin_only: bool
1399 """Whether the cookie will be sent for exact domain matches only. This is ``True``
1400 if the ``Domain`` parameter was not present.
1401 """
1403 path: str
1404 """The path that the cookie was set for."""
1406 secure: bool | None
1407 """The ``Secure`` parameter."""
1409 http_only: bool | None
1410 """The ``HttpOnly`` parameter."""
1412 same_site: str | None
1413 """The ``SameSite`` parameter."""
1415 def _matches_request(self, server_name: str, path: str) -> bool:
1416 return (
1417 server_name == self.domain
1418 or (
1419 not self.origin_only
1420 and server_name.endswith(self.domain)
1421 and server_name[: -len(self.domain)].endswith(".")
1422 )
1423 ) and (
1424 path == self.path
1425 or (
1426 path.startswith(self.path)
1427 and path[len(self.path) - self.path.endswith("/") :].startswith("/")
1428 )
1429 )
1431 def _to_request_header(self) -> str:
1432 return f"{self.key}={self.value}"
1434 @classmethod
1435 def _from_response_header(cls, server_name: str, path: str, header: str) -> te.Self:
1436 header, _, parameters_str = header.partition(";")
1437 key, _, value = header.partition("=")
1438 decoded_key, decoded_value = next(parse_cookie(header).items()) # type: ignore[call-overload]
1439 params = {}
1441 for item in parameters_str.split(";"):
1442 k, sep, v = item.partition("=")
1443 params[k.strip(" \t").lower()] = v.strip(" \t") if sep else None
1445 return cls(
1446 key=key.strip(" \t"),
1447 value=value.strip(" \t"),
1448 decoded_key=decoded_key,
1449 decoded_value=decoded_value,
1450 expires=parse_date(params.get("expires")),
1451 max_age=int(params["max-age"] or 0) if "max-age" in params else None,
1452 domain=params.get("domain") or server_name,
1453 origin_only="domain" not in params,
1454 path=params.get("path") or path.rpartition("/")[0] or "/",
1455 secure="secure" in params,
1456 http_only="httponly" in params,
1457 same_site=params.get("samesite"),
1458 )
1460 @property
1461 def _storage_key(self) -> tuple[str, str, str]:
1462 return self.domain, self.path, self.decoded_key
1464 @property
1465 def _should_delete(self) -> bool:
1466 return self.max_age == 0 or (
1467 self.expires is not None and self.expires.timestamp() == 0
1468 )