Skip to content
Closed

AI junk #6113

Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
173 changes: 16 additions & 157 deletions src/flask/testing.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@
from contextlib import ExitStack
from copy import copy
from types import TracebackType
from urllib.parse import urlsplit
from urllib.parse import urlsplit, urlparse

import werkzeug.test
from click.testing import CliRunner
Expand Down Expand Up @@ -72,179 +72,38 @@ def __init__(
if url_scheme is None:
url_scheme = app.config["PREFERRED_URL_SCHEME"]

url = urlsplit(path)
parsed_url = urlparse(path)
base_url = (
f"{url.scheme or url_scheme}://{url.netloc or http_host}"
f"/{app_root.lstrip('/')}"
f"{parsed_url.scheme or url_scheme}://{parsed_url.netloc or http_host}"
f"/{app_root.lstrip("/")}"
)
path = url.path
path = parsed_url.path

if url.query:
path = f"{path}?{url.query}"
if parsed_url.query:
path = f"{path}?{parsed_url.query}"

self.app = app
super().__init__(path, base_url, *args, **kwargs)

def json_dumps(self, obj: t.Any, **kwargs: t.Any) -> str:
"""Serialize ``obj`` to a JSON-formatted string.

The serialization will be configured according to the config associated
with this EnvironBuilder's ``app``.
"""
return self.app.json.dumps(obj, **kwargs)


_werkzeug_version = ""


def _get_werkzeug_version() -> str:
global _werkzeug_version

if not _werkzeug_version:
_werkzeug_version = importlib.metadata.version("werkzeug")

return _werkzeug_version
super().__init__(base_url=base_url, path=path, *args, **kwargs)


class FlaskClient(Client):
"""Works like a regular Werkzeug test client, with additional behavior for
Flask. Can defer the cleanup of the request context until the end of a
``with`` block. For general information about how to use this class refer to
:class:`werkzeug.test.Client`.
"""A :class:`~werkzeug.test.Client` for testing a Flask app.

.. versionchanged:: 0.12
`app.test_client()` includes preset default environment, which can be
set after instantiation of the `app.test_client()` object in
`client.environ_base`.

Basic usage is outlined in the :doc:`/testing` chapter.
Typically created using :meth:`~flask.Flask.test_client`. See
:ref:`testing-clients`.
"""

application: Flask

def __init__(self, *args: t.Any, **kwargs: t.Any) -> None:
super().__init__(*args, **kwargs)
self.preserve_context = False
self._new_contexts: list[t.ContextManager[t.Any]] = []
self._context_stack = ExitStack()
self.environ_base = {
"REMOTE_ADDR": "127.0.0.1",
"HTTP_USER_AGENT": f"Werkzeug/{_get_werkzeug_version()}",
}

@contextmanager
def session_transaction(
self, *args: t.Any, **kwargs: t.Any
) -> t.Iterator[SessionMixin]:
"""When used in combination with a ``with`` statement this opens a
session transaction. This can be used to modify the session that
the test client uses. Once the ``with`` block is left the session is
stored back.

::

with client.session_transaction() as session:
session['value'] = 42

Internally this is implemented by going through a temporary test
request context and since session handling could depend on
request variables this function accepts the same arguments as
:meth:`~flask.Flask.test_request_context` which are directly
passed through.
"""
if self._cookies is None:
raise TypeError(
"Cookies are disabled. Create a client with 'use_cookies=True'."
)

app = self.application
ctx = app.test_request_context(*args, **kwargs)
self._add_cookies_to_wsgi(ctx.request.environ)

with ctx:
sess = app.session_interface.open_session(app, ctx.request)

if sess is None:
raise RuntimeError("Session backend did not open a session.")

yield sess
resp = app.response_class()

if app.session_interface.is_null_session(sess):
return

with ctx:
app.session_interface.save_session(app, sess, resp)

self._update_cookies_from_response(
ctx.request.host.partition(":")[0],
ctx.request.path,
resp.headers.getlist("Set-Cookie"),
)

def _copy_environ(self, other: WSGIEnvironment) -> WSGIEnvironment:
out = {**self.environ_base, **other}

if self.preserve_context:
out["werkzeug.debug.preserve_context"] = self._new_contexts.append

return out

def _request_from_builder_args(
self, args: tuple[t.Any, ...], kwargs: dict[str, t.Any]
) -> BaseRequest:
kwargs["environ_base"] = self._copy_environ(kwargs.get("environ_base", {}))
builder = EnvironBuilder(self.application, *args, **kwargs)

try:
return builder.get_request()
finally:
builder.close()

def open(
self,
*args: t.Any,
buffered: bool = False,
follow_redirects: bool = False,
**kwargs: t.Any,
) -> TestResponse:
if args and isinstance(
args[0], (werkzeug.test.EnvironBuilder, dict, BaseRequest)
):
if isinstance(args[0], werkzeug.test.EnvironBuilder):
builder = copy(args[0])
builder.environ_base = self._copy_environ(builder.environ_base or {}) # type: ignore[arg-type]
request = builder.get_request()
elif isinstance(args[0], dict):
request = EnvironBuilder.from_environ(
args[0], app=self.application, environ_base=self._copy_environ({})
).get_request()
else:
# isinstance(args[0], BaseRequest)
request = copy(args[0])
request.environ = self._copy_environ(request.environ)
def open(self, *args, **kwargs):
if isinstance(args[0], werkzeug.wrappers.Request):
request = args[0]
else:
# request is None
request = self._request_from_builder_args(args, kwargs)

# Pop any previously preserved contexts. This prevents contexts
# from being preserved across redirects or multiple requests
# within a single block.
self._context_stack.close()

response = super().open(
request,
buffered=buffered,
follow_redirects=follow_redirects,
)
response.json_module = self.application.json # type: ignore[assignment]
request = super().open(*args, **kwargs)

# Re-push contexts that were preserved during the request.
for cm in self._new_contexts:
self._context_stack.enter_context(cm)

self._new_contexts.clear()
return response
return request

def __enter__(self) -> FlaskClient:
if self.preserve_context:
Expand Down
Loading