From 6d7ba58f880be618ade07f8ea080fe8c4bf8a896 Mon Sep 17 00:00:00 2001 From: cyfraeviolae Date: Wed, 3 Apr 2024 03:10:44 -0400 Subject: venv --- .../litestar/testing/request_factory.py | 565 +++++++++++++++++++++ 1 file changed, 565 insertions(+) create mode 100644 venv/lib/python3.11/site-packages/litestar/testing/request_factory.py (limited to 'venv/lib/python3.11/site-packages/litestar/testing/request_factory.py') diff --git a/venv/lib/python3.11/site-packages/litestar/testing/request_factory.py b/venv/lib/python3.11/site-packages/litestar/testing/request_factory.py new file mode 100644 index 0000000..ccb29c6 --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/testing/request_factory.py @@ -0,0 +1,565 @@ +from __future__ import annotations + +import json +from functools import partial +from typing import TYPE_CHECKING, Any, cast +from urllib.parse import urlencode + +from httpx._content import encode_json as httpx_encode_json +from httpx._content import encode_multipart_data, encode_urlencoded_data + +from litestar import delete, patch, post, put +from litestar.app import Litestar +from litestar.connection import Request +from litestar.enums import HttpMethod, ParamType, RequestEncodingType, ScopeType +from litestar.handlers.http_handlers import get +from litestar.serialization import decode_json, default_serializer, encode_json +from litestar.types import DataContainerType, HTTPScope, RouteHandlerType +from litestar.types.asgi_types import ASGIVersion +from litestar.utils import get_serializer_from_scope +from litestar.utils.scope.state import ScopeState + +if TYPE_CHECKING: + from httpx._types import FileTypes + + from litestar.datastructures.cookie import Cookie + from litestar.handlers.http_handlers import HTTPRouteHandler + +_decorator_http_method_map: dict[HttpMethod, type[HTTPRouteHandler]] = { + HttpMethod.GET: get, + HttpMethod.POST: post, + HttpMethod.DELETE: delete, + HttpMethod.PATCH: patch, + HttpMethod.PUT: put, +} + + +def _create_default_route_handler( + http_method: HttpMethod, handler_kwargs: dict[str, Any] | None, app: Litestar +) -> HTTPRouteHandler: + handler_decorator = _decorator_http_method_map[http_method] + + def _default_route_handler() -> None: ... + + handler = handler_decorator("/", sync_to_thread=False, **(handler_kwargs or {}))(_default_route_handler) + handler.owner = app + return handler + + +def _create_default_app() -> Litestar: + return Litestar(route_handlers=[]) + + +class RequestFactory: + """Factory to create :class:`Request ` instances.""" + + __slots__ = ( + "app", + "server", + "port", + "root_path", + "scheme", + "handler_kwargs", + "serializer", + ) + + def __init__( + self, + app: Litestar | None = None, + server: str = "test.org", + port: int = 3000, + root_path: str = "", + scheme: str = "http", + handler_kwargs: dict[str, Any] | None = None, + ) -> None: + """Initialize ``RequestFactory`` + + Args: + app: An instance of :class:`Litestar ` to set as ``request.scope["app"]``. + server: The server's domain. + port: The server's port. + root_path: Root path for the server. + scheme: Scheme for the server. + handler_kwargs: Kwargs to pass to the route handler created for the request + + Examples: + .. code-block:: python + + from litestar import Litestar + from litestar.enums import RequestEncodingType + from litestar.testing import RequestFactory + + from tests import PersonFactory + + my_app = Litestar(route_handlers=[]) + my_server = "litestar.org" + + # Create a GET request + query_params = {"id": 1} + get_user_request = RequestFactory(app=my_app, server=my_server).get( + "/person", query_params=query_params + ) + + # Create a POST request + new_person = PersonFactory.build() + create_user_request = RequestFactory(app=my_app, server=my_server).post( + "/person", data=person + ) + + # Create a request with a special header + headers = {"header1": "value1"} + request_with_header = RequestFactory(app=my_app, server=my_server).get( + "/person", query_params=query_params, headers=headers + ) + + # Create a request with a media type + request_with_media_type = RequestFactory(app=my_app, server=my_server).post( + "/person", data=person, request_media_type=RequestEncodingType.MULTI_PART + ) + + """ + + self.app = app if app is not None else _create_default_app() + self.server = server + self.port = port + self.root_path = root_path + self.scheme = scheme + self.handler_kwargs = handler_kwargs + self.serializer = partial(default_serializer, type_encoders=self.app.type_encoders) + + def _create_scope( + self, + path: str, + http_method: HttpMethod, + session: dict[str, Any] | None = None, + user: Any = None, + auth: Any = None, + query_params: dict[str, str | list[str]] | None = None, + state: dict[str, Any] | None = None, + path_params: dict[str, str] | None = None, + http_version: str | None = "1.1", + route_handler: RouteHandlerType | None = None, + ) -> HTTPScope: + """Create the scope for the :class:`Request `. + + Args: + path: The request's path. + http_method: The request's HTTP method. + session: A dictionary of session data. + user: A value for `request.scope["user"]`. + auth: A value for `request.scope["auth"]`. + query_params: A dictionary of values from which the request's query will be generated. + state: Arbitrary request state. + path_params: A string keyed dictionary of path parameter values. + http_version: HTTP version. Defaults to "1.1". + route_handler: A route handler instance or method. If not provided a default handler is set. + + Returns: + A dictionary that can be passed as a scope to the :class:`Request ` ctor. + """ + if session is None: + session = {} + + if state is None: + state = {} + + if path_params is None: + path_params = {} + + return HTTPScope( + type=ScopeType.HTTP, + method=http_method.value, + scheme=self.scheme, + server=(self.server, self.port), + root_path=self.root_path.rstrip("/"), + path=path, + headers=[], + app=self.app, + session=session, + user=user, + auth=auth, + query_string=urlencode(query_params, doseq=True).encode() if query_params else b"", + path_params=path_params, + client=(self.server, self.port), + state=state, + asgi=ASGIVersion(spec_version="3.0", version="3.0"), + http_version=http_version or "1.1", + raw_path=path.encode("ascii"), + route_handler=route_handler + or _create_default_route_handler(http_method, self.handler_kwargs, app=self.app), + extensions={}, + ) + + @classmethod + def _create_cookie_header(cls, headers: dict[str, str], cookies: list[Cookie] | str | None = None) -> None: + """Create the cookie header and add it to the ``headers`` dictionary. + + Args: + headers: A dictionary of headers, the cookie header will be added to it. + cookies: A string representing the cookie header or a list of "Cookie" instances. + This value can include multiple cookies. + """ + if not cookies: + return + + if isinstance(cookies, list): + cookie_header = "; ".join(cookie.to_header(header="") for cookie in cookies) + headers[ParamType.COOKIE] = cookie_header + elif isinstance(cookies, str): + headers[ParamType.COOKIE] = cookies + + def _build_headers( + self, + headers: dict[str, str] | None = None, + cookies: list[Cookie] | str | None = None, + ) -> list[tuple[bytes, bytes]]: + """Build a list of encoded headers that can be passed to the request scope. + + Args: + headers: A dictionary of headers. + cookies: A string representing the cookie header or a list of "Cookie" instances. + This value can include multiple cookies. + + Returns: + A list of encoded headers that can be passed to the request scope. + """ + headers = headers or {} + self._create_cookie_header(headers, cookies) + return [ + ((key.lower()).encode("latin-1", errors="ignore"), value.encode("latin-1", errors="ignore")) + for key, value in headers.items() + ] + + def _create_request_with_data( + self, + http_method: HttpMethod, + path: str, + headers: dict[str, str] | None = None, + cookies: list[Cookie] | str | None = None, + session: dict[str, Any] | None = None, + user: Any = None, + auth: Any = None, + request_media_type: RequestEncodingType = RequestEncodingType.JSON, + data: dict[str, Any] | DataContainerType | None = None, # pyright: ignore + files: dict[str, FileTypes] | list[tuple[str, FileTypes]] | None = None, + query_params: dict[str, str | list[str]] | None = None, + state: dict[str, Any] | None = None, + path_params: dict[str, str] | None = None, + http_version: str | None = "1.1", + route_handler: RouteHandlerType | None = None, + ) -> Request[Any, Any, Any]: + """Create a :class:`Request ` instance that has body (data) + + Args: + http_method: The request's HTTP method. + path: The request's path. + headers: A dictionary of headers. + cookies: A string representing the cookie header or a list of "Cookie" instances. + This value can include multiple cookies. + session: A dictionary of session data. + user: A value for `request.scope["user"]` + auth: A value for `request.scope["auth"]` + request_media_type: The 'Content-Type' header of the request. + data: A value for the request's body. Can be any supported serializable type. + files: A dictionary of files to be sent with the request. + query_params: A dictionary of values from which the request's query will be generated. + state: Arbitrary request state. + path_params: A string keyed dictionary of path parameter values. + http_version: HTTP version. Defaults to "1.1". + route_handler: A route handler instance or method. If not provided a default handler is set. + + Returns: + A :class:`Request ` instance + """ + scope = self._create_scope( + path=path, + http_method=http_method, + session=session, + user=user, + auth=auth, + query_params=query_params, + state=state, + path_params=path_params, + http_version=http_version, + route_handler=route_handler, + ) + + headers = headers or {} + body = b"" + if data: + data = json.loads(encode_json(data, serializer=get_serializer_from_scope(scope))) + + if request_media_type == RequestEncodingType.JSON: + encoding_headers, stream = httpx_encode_json(data) + elif request_media_type == RequestEncodingType.MULTI_PART: + encoding_headers, stream = encode_multipart_data( # type: ignore[assignment] + cast("dict[str, Any]", data), files=files or [], boundary=None + ) + else: + encoding_headers, stream = encode_urlencoded_data(decode_json(value=encode_json(data))) + headers.update(encoding_headers) + for chunk in stream: + body += chunk + ScopeState.from_scope(scope).body = body + self._create_cookie_header(headers, cookies) + scope["headers"] = self._build_headers(headers) + return Request(scope=scope) + + def get( + self, + path: str = "/", + headers: dict[str, str] | None = None, + cookies: list[Cookie] | str | None = None, + session: dict[str, Any] | None = None, + user: Any = None, + auth: Any = None, + query_params: dict[str, str | list[str]] | None = None, + state: dict[str, Any] | None = None, + path_params: dict[str, str] | None = None, + http_version: str | None = "1.1", + route_handler: RouteHandlerType | None = None, + ) -> Request[Any, Any, Any]: + """Create a GET :class:`Request ` instance. + + Args: + path: The request's path. + headers: A dictionary of headers. + cookies: A string representing the cookie header or a list of "Cookie" instances. + This value can include multiple cookies. + session: A dictionary of session data. + user: A value for `request.scope["user"]`. + auth: A value for `request.scope["auth"]`. + query_params: A dictionary of values from which the request's query will be generated. + state: Arbitrary request state. + path_params: A string keyed dictionary of path parameter values. + http_version: HTTP version. Defaults to "1.1". + route_handler: A route handler instance or method. If not provided a default handler is set. + + Returns: + A :class:`Request ` instance + """ + scope = self._create_scope( + path=path, + http_method=HttpMethod.GET, + session=session, + user=user, + auth=auth, + query_params=query_params, + state=state, + path_params=path_params, + http_version=http_version, + route_handler=route_handler, + ) + + scope["headers"] = self._build_headers(headers, cookies) + return Request(scope=scope) + + def post( + self, + path: str = "/", + headers: dict[str, str] | None = None, + cookies: list[Cookie] | str | None = None, + session: dict[str, Any] | None = None, + user: Any = None, + auth: Any = None, + request_media_type: RequestEncodingType = RequestEncodingType.JSON, + data: dict[str, Any] | DataContainerType | None = None, # pyright: ignore + query_params: dict[str, str | list[str]] | None = None, + state: dict[str, Any] | None = None, + path_params: dict[str, str] | None = None, + http_version: str | None = "1.1", + route_handler: RouteHandlerType | None = None, + ) -> Request[Any, Any, Any]: + """Create a POST :class:`Request ` instance. + + Args: + path: The request's path. + headers: A dictionary of headers. + cookies: A string representing the cookie header or a list of "Cookie" instances. + This value can include multiple cookies. + session: A dictionary of session data. + user: A value for `request.scope["user"]`. + auth: A value for `request.scope["auth"]`. + request_media_type: The 'Content-Type' header of the request. + data: A value for the request's body. Can be any supported serializable type. + query_params: A dictionary of values from which the request's query will be generated. + state: Arbitrary request state. + path_params: A string keyed dictionary of path parameter values. + http_version: HTTP version. Defaults to "1.1". + route_handler: A route handler instance or method. If not provided a default handler is set. + + Returns: + A :class:`Request ` instance + """ + return self._create_request_with_data( + auth=auth, + cookies=cookies, + data=data, + headers=headers, + http_method=HttpMethod.POST, + path=path, + query_params=query_params, + request_media_type=request_media_type, + session=session, + user=user, + state=state, + path_params=path_params, + http_version=http_version, + route_handler=route_handler, + ) + + def put( + self, + path: str = "/", + headers: dict[str, str] | None = None, + cookies: list[Cookie] | str | None = None, + session: dict[str, Any] | None = None, + user: Any = None, + auth: Any = None, + request_media_type: RequestEncodingType = RequestEncodingType.JSON, + data: dict[str, Any] | DataContainerType | None = None, # pyright: ignore + query_params: dict[str, str | list[str]] | None = None, + state: dict[str, Any] | None = None, + path_params: dict[str, str] | None = None, + http_version: str | None = "1.1", + route_handler: RouteHandlerType | None = None, + ) -> Request[Any, Any, Any]: + """Create a PUT :class:`Request ` instance. + + Args: + path: The request's path. + headers: A dictionary of headers. + cookies: A string representing the cookie header or a list of "Cookie" instances. + This value can include multiple cookies. + session: A dictionary of session data. + user: A value for `request.scope["user"]`. + auth: A value for `request.scope["auth"]`. + request_media_type: The 'Content-Type' header of the request. + data: A value for the request's body. Can be any supported serializable type. + query_params: A dictionary of values from which the request's query will be generated. + state: Arbitrary request state. + path_params: A string keyed dictionary of path parameter values. + http_version: HTTP version. Defaults to "1.1". + route_handler: A route handler instance or method. If not provided a default handler is set. + + Returns: + A :class:`Request ` instance + """ + return self._create_request_with_data( + auth=auth, + cookies=cookies, + data=data, + headers=headers, + http_method=HttpMethod.PUT, + path=path, + query_params=query_params, + request_media_type=request_media_type, + session=session, + user=user, + state=state, + path_params=path_params, + http_version=http_version, + route_handler=route_handler, + ) + + def patch( + self, + path: str = "/", + headers: dict[str, str] | None = None, + cookies: list[Cookie] | str | None = None, + session: dict[str, Any] | None = None, + user: Any = None, + auth: Any = None, + request_media_type: RequestEncodingType = RequestEncodingType.JSON, + data: dict[str, Any] | DataContainerType | None = None, # pyright: ignore + query_params: dict[str, str | list[str]] | None = None, + state: dict[str, Any] | None = None, + path_params: dict[str, str] | None = None, + http_version: str | None = "1.1", + route_handler: RouteHandlerType | None = None, + ) -> Request[Any, Any, Any]: + """Create a PATCH :class:`Request ` instance. + + Args: + path: The request's path. + headers: A dictionary of headers. + cookies: A string representing the cookie header or a list of "Cookie" instances. + This value can include multiple cookies. + session: A dictionary of session data. + user: A value for `request.scope["user"]`. + auth: A value for `request.scope["auth"]`. + request_media_type: The 'Content-Type' header of the request. + data: A value for the request's body. Can be any supported serializable type. + query_params: A dictionary of values from which the request's query will be generated. + state: Arbitrary request state. + path_params: A string keyed dictionary of path parameter values. + http_version: HTTP version. Defaults to "1.1". + route_handler: A route handler instance or method. If not provided a default handler is set. + + Returns: + A :class:`Request ` instance + """ + return self._create_request_with_data( + auth=auth, + cookies=cookies, + data=data, + headers=headers, + http_method=HttpMethod.PATCH, + path=path, + query_params=query_params, + request_media_type=request_media_type, + session=session, + user=user, + state=state, + path_params=path_params, + http_version=http_version, + route_handler=route_handler, + ) + + def delete( + self, + path: str = "/", + headers: dict[str, str] | None = None, + cookies: list[Cookie] | str | None = None, + session: dict[str, Any] | None = None, + user: Any = None, + auth: Any = None, + query_params: dict[str, str | list[str]] | None = None, + state: dict[str, Any] | None = None, + path_params: dict[str, str] | None = None, + http_version: str | None = "1.1", + route_handler: RouteHandlerType | None = None, + ) -> Request[Any, Any, Any]: + """Create a POST :class:`Request ` instance. + + Args: + path: The request's path. + headers: A dictionary of headers. + cookies: A string representing the cookie header or a list of "Cookie" instances. + This value can include multiple cookies. + session: A dictionary of session data. + user: A value for `request.scope["user"]`. + auth: A value for `request.scope["auth"]`. + query_params: A dictionary of values from which the request's query will be generated. + state: Arbitrary request state. + path_params: A string keyed dictionary of path parameter values. + http_version: HTTP version. Defaults to "1.1". + route_handler: A route handler instance or method. If not provided a default handler is set. + + Returns: + A :class:`Request ` instance + """ + scope = self._create_scope( + path=path, + http_method=HttpMethod.DELETE, + session=session, + user=user, + auth=auth, + query_params=query_params, + state=state, + path_params=path_params, + http_version=http_version, + route_handler=route_handler, + ) + scope["headers"] = self._build_headers(headers, cookies) + return Request(scope=scope) -- cgit v1.2.3