summaryrefslogtreecommitdiff
path: root/venv/lib/python3.11/site-packages/litestar/testing/client
diff options
context:
space:
mode:
authorcyfraeviolae <cyfraeviolae>2024-04-03 03:17:55 -0400
committercyfraeviolae <cyfraeviolae>2024-04-03 03:17:55 -0400
commit12cf076118570eebbff08c6b3090e0d4798447a1 (patch)
tree3ba25e17e3c3a5e82316558ba3864b955919ff72 /venv/lib/python3.11/site-packages/litestar/testing/client
parentc45662ff3923b34614ddcc8feb9195541166dcc5 (diff)
no venv
Diffstat (limited to 'venv/lib/python3.11/site-packages/litestar/testing/client')
-rw-r--r--venv/lib/python3.11/site-packages/litestar/testing/client/__init__.py36
-rw-r--r--venv/lib/python3.11/site-packages/litestar/testing/client/__pycache__/__init__.cpython-311.pycbin2091 -> 0 bytes
-rw-r--r--venv/lib/python3.11/site-packages/litestar/testing/client/__pycache__/async_client.cpython-311.pycbin17343 -> 0 bytes
-rw-r--r--venv/lib/python3.11/site-packages/litestar/testing/client/__pycache__/base.cpython-311.pycbin9397 -> 0 bytes
-rw-r--r--venv/lib/python3.11/site-packages/litestar/testing/client/__pycache__/sync_client.cpython-311.pycbin19167 -> 0 bytes
-rw-r--r--venv/lib/python3.11/site-packages/litestar/testing/client/async_client.py534
-rw-r--r--venv/lib/python3.11/site-packages/litestar/testing/client/base.py180
-rw-r--r--venv/lib/python3.11/site-packages/litestar/testing/client/sync_client.py593
8 files changed, 0 insertions, 1343 deletions
diff --git a/venv/lib/python3.11/site-packages/litestar/testing/client/__init__.py b/venv/lib/python3.11/site-packages/litestar/testing/client/__init__.py
deleted file mode 100644
index 5d03a7a..0000000
--- a/venv/lib/python3.11/site-packages/litestar/testing/client/__init__.py
+++ /dev/null
@@ -1,36 +0,0 @@
-"""Some code in this module was adapted from https://github.com/encode/starlette/blob/master/starlette/testclient.py.
-
-Copyright © 2018, [Encode OSS Ltd](https://www.encode.io/).
-All rights reserved.
-
-Redistribution and use in source and binary forms, with or without
-modification, are permitted provided that the following conditions are met:
-
-* Redistributions of source code must retain the above copyright notice, this
- list of conditions and the following disclaimer.
-
-* Redistributions in binary form must reproduce the above copyright notice,
- this list of conditions and the following disclaimer in the documentation
- and/or other materials provided with the distribution.
-
-* Neither the name of the copyright holder nor the names of its
- contributors may be used to endorse or promote products derived from
- this software without specific prior written permission.
-
-THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
-AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
-IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
-DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
-FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
-DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
-SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
-CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
-OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
-OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
-"""
-
-from .async_client import AsyncTestClient
-from .base import BaseTestClient
-from .sync_client import TestClient
-
-__all__ = ("TestClient", "AsyncTestClient", "BaseTestClient")
diff --git a/venv/lib/python3.11/site-packages/litestar/testing/client/__pycache__/__init__.cpython-311.pyc b/venv/lib/python3.11/site-packages/litestar/testing/client/__pycache__/__init__.cpython-311.pyc
deleted file mode 100644
index 18ad148..0000000
--- a/venv/lib/python3.11/site-packages/litestar/testing/client/__pycache__/__init__.cpython-311.pyc
+++ /dev/null
Binary files differ
diff --git a/venv/lib/python3.11/site-packages/litestar/testing/client/__pycache__/async_client.cpython-311.pyc b/venv/lib/python3.11/site-packages/litestar/testing/client/__pycache__/async_client.cpython-311.pyc
deleted file mode 100644
index 1ccc805..0000000
--- a/venv/lib/python3.11/site-packages/litestar/testing/client/__pycache__/async_client.cpython-311.pyc
+++ /dev/null
Binary files differ
diff --git a/venv/lib/python3.11/site-packages/litestar/testing/client/__pycache__/base.cpython-311.pyc b/venv/lib/python3.11/site-packages/litestar/testing/client/__pycache__/base.cpython-311.pyc
deleted file mode 100644
index 87d5de7..0000000
--- a/venv/lib/python3.11/site-packages/litestar/testing/client/__pycache__/base.cpython-311.pyc
+++ /dev/null
Binary files differ
diff --git a/venv/lib/python3.11/site-packages/litestar/testing/client/__pycache__/sync_client.cpython-311.pyc b/venv/lib/python3.11/site-packages/litestar/testing/client/__pycache__/sync_client.cpython-311.pyc
deleted file mode 100644
index 29f0576..0000000
--- a/venv/lib/python3.11/site-packages/litestar/testing/client/__pycache__/sync_client.cpython-311.pyc
+++ /dev/null
Binary files differ
diff --git a/venv/lib/python3.11/site-packages/litestar/testing/client/async_client.py b/venv/lib/python3.11/site-packages/litestar/testing/client/async_client.py
deleted file mode 100644
index cf66f12..0000000
--- a/venv/lib/python3.11/site-packages/litestar/testing/client/async_client.py
+++ /dev/null
@@ -1,534 +0,0 @@
-from __future__ import annotations
-
-from contextlib import AsyncExitStack
-from typing import TYPE_CHECKING, Any, Generic, Mapping, TypeVar
-
-from httpx import USE_CLIENT_DEFAULT, AsyncClient, Response
-
-from litestar import HttpMethod
-from litestar.testing.client.base import BaseTestClient
-from litestar.testing.life_span_handler import LifeSpanHandler
-from litestar.testing.transport import TestClientTransport
-from litestar.types import AnyIOBackend, ASGIApp
-
-if TYPE_CHECKING:
- from httpx._client import UseClientDefault
- from httpx._types import (
- AuthTypes,
- CookieTypes,
- HeaderTypes,
- QueryParamTypes,
- RequestContent,
- RequestData,
- RequestFiles,
- TimeoutTypes,
- URLTypes,
- )
- from typing_extensions import Self
-
- from litestar.middleware.session.base import BaseBackendConfig
-
-
-T = TypeVar("T", bound=ASGIApp)
-
-
-class AsyncTestClient(AsyncClient, BaseTestClient, Generic[T]): # type: ignore[misc]
- lifespan_handler: LifeSpanHandler[Any]
- exit_stack: AsyncExitStack
-
- def __init__(
- self,
- app: T,
- base_url: str = "http://testserver.local",
- raise_server_exceptions: bool = True,
- root_path: str = "",
- backend: AnyIOBackend = "asyncio",
- backend_options: Mapping[str, Any] | None = None,
- session_config: BaseBackendConfig | None = None,
- timeout: float | None = None,
- cookies: CookieTypes | None = None,
- ) -> None:
- """An Async client implementation providing a context manager for testing applications asynchronously.
-
- Args:
- app: The instance of :class:`Litestar <litestar.app.Litestar>` under test.
- base_url: URL scheme and domain for test request paths, e.g. ``http://testserver``.
- raise_server_exceptions: Flag for the underlying test client to raise server exceptions instead of
- wrapping them in an HTTP response.
- root_path: Path prefix for requests.
- backend: The async backend to use, options are "asyncio" or "trio".
- backend_options: 'anyio' options.
- session_config: Configuration for Session Middleware class to create raw session cookies for request to the
- route handlers.
- timeout: Request timeout
- cookies: Cookies to set on the client.
- """
- BaseTestClient.__init__(
- self,
- app=app,
- base_url=base_url,
- backend=backend,
- backend_options=backend_options,
- session_config=session_config,
- cookies=cookies,
- )
- AsyncClient.__init__(
- self,
- base_url=base_url,
- headers={"user-agent": "testclient"},
- follow_redirects=True,
- cookies=cookies,
- transport=TestClientTransport( # type: ignore [arg-type]
- client=self,
- raise_server_exceptions=raise_server_exceptions,
- root_path=root_path,
- ),
- timeout=timeout,
- )
-
- async def __aenter__(self) -> Self:
- async with AsyncExitStack() as stack:
- self.blocking_portal = portal = stack.enter_context(self.portal())
- self.lifespan_handler = LifeSpanHandler(client=self)
-
- @stack.callback
- def reset_portal() -> None:
- delattr(self, "blocking_portal")
-
- @stack.callback
- def wait_shutdown() -> None:
- portal.call(self.lifespan_handler.wait_shutdown)
-
- self.exit_stack = stack.pop_all()
- return self
-
- async def __aexit__(self, *args: Any) -> None:
- await self.exit_stack.aclose()
-
- async def request(
- self,
- method: str,
- url: URLTypes,
- *,
- content: RequestContent | None = None,
- data: RequestData | None = None,
- files: RequestFiles | None = None,
- json: Any | None = None,
- params: QueryParamTypes | None = None,
- headers: HeaderTypes | None = None,
- cookies: CookieTypes | None = None,
- auth: AuthTypes | UseClientDefault | None = USE_CLIENT_DEFAULT,
- follow_redirects: bool | UseClientDefault = USE_CLIENT_DEFAULT,
- timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
- extensions: Mapping[str, Any] | None = None,
- ) -> Response:
- """Sends a request.
-
- Args:
- method: An HTTP method.
- url: URL or path for the request.
- content: Request content.
- data: Form encoded data.
- files: Multipart files to send.
- json: JSON data to send.
- params: Query parameters.
- headers: Request headers.
- cookies: Request cookies.
- auth: Auth headers.
- follow_redirects: Whether to follow redirects.
- timeout: Request timeout.
- extensions: Dictionary of ASGI extensions.
-
- Returns:
- An HTTPX Response.
- """
- return await AsyncClient.request(
- self,
- url=self.base_url.join(url),
- method=method.value if isinstance(method, HttpMethod) else method,
- content=content,
- data=data,
- files=files,
- json=json,
- params=params,
- headers=headers,
- cookies=cookies,
- auth=auth,
- follow_redirects=follow_redirects,
- timeout=timeout,
- extensions=None if extensions is None else dict(extensions),
- )
-
- async def get( # type: ignore [override]
- self,
- url: URLTypes,
- *,
- params: QueryParamTypes | None = None,
- headers: HeaderTypes | None = None,
- cookies: CookieTypes | None = None,
- auth: AuthTypes | UseClientDefault = USE_CLIENT_DEFAULT,
- follow_redirects: bool | UseClientDefault = USE_CLIENT_DEFAULT,
- timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
- extensions: Mapping[str, Any] | None = None,
- ) -> Response:
- """Sends a GET request.
-
- Args:
- url: URL or path for the request.
- params: Query parameters.
- headers: Request headers.
- cookies: Request cookies.
- auth: Auth headers.
- follow_redirects: Whether to follow redirects.
- timeout: Request timeout.
- extensions: Dictionary of ASGI extensions.
-
- Returns:
- An HTTPX Response.
- """
- return await AsyncClient.get(
- self,
- url,
- params=params,
- headers=headers,
- cookies=cookies,
- auth=auth,
- follow_redirects=follow_redirects,
- timeout=timeout,
- extensions=None if extensions is None else dict(extensions),
- )
-
- async def options(
- self,
- url: URLTypes,
- *,
- params: QueryParamTypes | None = None,
- headers: HeaderTypes | None = None,
- cookies: CookieTypes | None = None,
- auth: AuthTypes | UseClientDefault = USE_CLIENT_DEFAULT,
- follow_redirects: bool | UseClientDefault = USE_CLIENT_DEFAULT,
- timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
- extensions: Mapping[str, Any] | None = None,
- ) -> Response:
- """Sends an OPTIONS request.
-
- Args:
- url: URL or path for the request.
- params: Query parameters.
- headers: Request headers.
- cookies: Request cookies.
- auth: Auth headers.
- follow_redirects: Whether to follow redirects.
- timeout: Request timeout.
- extensions: Dictionary of ASGI extensions.
-
- Returns:
- An HTTPX Response.
- """
- return await AsyncClient.options(
- self,
- url,
- params=params,
- headers=headers,
- cookies=cookies,
- auth=auth,
- follow_redirects=follow_redirects,
- timeout=timeout,
- extensions=None if extensions is None else dict(extensions),
- )
-
- async def head(
- self,
- url: URLTypes,
- *,
- params: QueryParamTypes | None = None,
- headers: HeaderTypes | None = None,
- cookies: CookieTypes | None = None,
- auth: AuthTypes | UseClientDefault = USE_CLIENT_DEFAULT,
- follow_redirects: bool | UseClientDefault = USE_CLIENT_DEFAULT,
- timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
- extensions: Mapping[str, Any] | None = None,
- ) -> Response:
- """Sends a HEAD request.
-
- Args:
- url: URL or path for the request.
- params: Query parameters.
- headers: Request headers.
- cookies: Request cookies.
- auth: Auth headers.
- follow_redirects: Whether to follow redirects.
- timeout: Request timeout.
- extensions: Dictionary of ASGI extensions.
-
- Returns:
- An HTTPX Response.
- """
- return await AsyncClient.head(
- self,
- url,
- params=params,
- headers=headers,
- cookies=cookies,
- auth=auth,
- follow_redirects=follow_redirects,
- timeout=timeout,
- extensions=None if extensions is None else dict(extensions),
- )
-
- async def post(
- self,
- url: URLTypes,
- *,
- content: RequestContent | None = None,
- data: RequestData | None = None,
- files: RequestFiles | None = None,
- json: Any | None = None,
- params: QueryParamTypes | None = None,
- headers: HeaderTypes | None = None,
- cookies: CookieTypes | None = None,
- auth: AuthTypes | UseClientDefault = USE_CLIENT_DEFAULT,
- follow_redirects: bool | UseClientDefault = USE_CLIENT_DEFAULT,
- timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
- extensions: Mapping[str, Any] | None = None,
- ) -> Response:
- """Sends a POST request.
-
- Args:
- url: URL or path for the request.
- content: Request content.
- data: Form encoded data.
- files: Multipart files to send.
- json: JSON data to send.
- params: Query parameters.
- headers: Request headers.
- cookies: Request cookies.
- auth: Auth headers.
- follow_redirects: Whether to follow redirects.
- timeout: Request timeout.
- extensions: Dictionary of ASGI extensions.
-
- Returns:
- An HTTPX Response.
- """
- return await AsyncClient.post(
- self,
- url,
- content=content,
- data=data,
- files=files,
- json=json,
- params=params,
- headers=headers,
- cookies=cookies,
- auth=auth,
- follow_redirects=follow_redirects,
- timeout=timeout,
- extensions=None if extensions is None else dict(extensions),
- )
-
- async def put(
- self,
- url: URLTypes,
- *,
- content: RequestContent | None = None,
- data: RequestData | None = None,
- files: RequestFiles | None = None,
- json: Any | None = None,
- params: QueryParamTypes | None = None,
- headers: HeaderTypes | None = None,
- cookies: CookieTypes | None = None,
- auth: AuthTypes | UseClientDefault = USE_CLIENT_DEFAULT,
- follow_redirects: bool | UseClientDefault = USE_CLIENT_DEFAULT,
- timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
- extensions: Mapping[str, Any] | None = None,
- ) -> Response:
- """Sends a PUT request.
-
- Args:
- url: URL or path for the request.
- content: Request content.
- data: Form encoded data.
- files: Multipart files to send.
- json: JSON data to send.
- params: Query parameters.
- headers: Request headers.
- cookies: Request cookies.
- auth: Auth headers.
- follow_redirects: Whether to follow redirects.
- timeout: Request timeout.
- extensions: Dictionary of ASGI extensions.
-
- Returns:
- An HTTPX Response.
- """
- return await AsyncClient.put(
- self,
- url,
- content=content,
- data=data,
- files=files,
- json=json,
- params=params,
- headers=headers,
- cookies=cookies,
- auth=auth,
- follow_redirects=follow_redirects,
- timeout=timeout,
- extensions=None if extensions is None else dict(extensions),
- )
-
- async def patch(
- self,
- url: URLTypes,
- *,
- content: RequestContent | None = None,
- data: RequestData | None = None,
- files: RequestFiles | None = None,
- json: Any | None = None,
- params: QueryParamTypes | None = None,
- headers: HeaderTypes | None = None,
- cookies: CookieTypes | None = None,
- auth: AuthTypes | UseClientDefault = USE_CLIENT_DEFAULT,
- follow_redirects: bool | UseClientDefault = USE_CLIENT_DEFAULT,
- timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
- extensions: Mapping[str, Any] | None = None,
- ) -> Response:
- """Sends a PATCH request.
-
- Args:
- url: URL or path for the request.
- content: Request content.
- data: Form encoded data.
- files: Multipart files to send.
- json: JSON data to send.
- params: Query parameters.
- headers: Request headers.
- cookies: Request cookies.
- auth: Auth headers.
- follow_redirects: Whether to follow redirects.
- timeout: Request timeout.
- extensions: Dictionary of ASGI extensions.
-
- Returns:
- An HTTPX Response.
- """
- return await AsyncClient.patch(
- self,
- url,
- content=content,
- data=data,
- files=files,
- json=json,
- params=params,
- headers=headers,
- cookies=cookies,
- auth=auth,
- follow_redirects=follow_redirects,
- timeout=timeout,
- extensions=None if extensions is None else dict(extensions),
- )
-
- async def delete(
- self,
- url: URLTypes,
- *,
- params: QueryParamTypes | None = None,
- headers: HeaderTypes | None = None,
- cookies: CookieTypes | None = None,
- auth: AuthTypes | UseClientDefault = USE_CLIENT_DEFAULT,
- follow_redirects: bool | UseClientDefault = USE_CLIENT_DEFAULT,
- timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
- extensions: Mapping[str, Any] | None = None,
- ) -> Response:
- """Sends a DELETE request.
-
- Args:
- url: URL or path for the request.
- params: Query parameters.
- headers: Request headers.
- cookies: Request cookies.
- auth: Auth headers.
- follow_redirects: Whether to follow redirects.
- timeout: Request timeout.
- extensions: Dictionary of ASGI extensions.
-
- Returns:
- An HTTPX Response.
- """
- return await AsyncClient.delete(
- self,
- url,
- params=params,
- headers=headers,
- cookies=cookies,
- auth=auth,
- follow_redirects=follow_redirects,
- timeout=timeout,
- extensions=None if extensions is None else dict(extensions),
- )
-
- async def get_session_data(self) -> dict[str, Any]:
- """Get session data.
-
- Returns:
- A dictionary containing session data.
-
- Examples:
- .. code-block:: python
-
- from litestar import Litestar, post
- from litestar.middleware.session.memory_backend import MemoryBackendConfig
-
- session_config = MemoryBackendConfig()
-
-
- @post(path="/test")
- def set_session_data(request: Request) -> None:
- request.session["foo"] == "bar"
-
-
- app = Litestar(
- route_handlers=[set_session_data], middleware=[session_config.middleware]
- )
-
- async with AsyncTestClient(app=app, session_config=session_config) as client:
- await client.post("/test")
- assert await client.get_session_data() == {"foo": "bar"}
-
- """
- return await super()._get_session_data()
-
- async def set_session_data(self, data: dict[str, Any]) -> None:
- """Set session data.
-
- Args:
- data: Session data
-
- Returns:
- None
-
- Examples:
- .. code-block:: python
-
- from litestar import Litestar, get
- from litestar.middleware.session.memory_backend import MemoryBackendConfig
-
- session_config = MemoryBackendConfig()
-
-
- @get(path="/test")
- def get_session_data(request: Request) -> Dict[str, Any]:
- return request.session
-
-
- app = Litestar(
- route_handlers=[get_session_data], middleware=[session_config.middleware]
- )
-
- async with AsyncTestClient(app=app, session_config=session_config) as client:
- await client.set_session_data({"foo": "bar"})
- assert await client.get("/test").json() == {"foo": "bar"}
-
- """
- return await super()._set_session_data(data)
diff --git a/venv/lib/python3.11/site-packages/litestar/testing/client/base.py b/venv/lib/python3.11/site-packages/litestar/testing/client/base.py
deleted file mode 100644
index 3c25be1..0000000
--- a/venv/lib/python3.11/site-packages/litestar/testing/client/base.py
+++ /dev/null
@@ -1,180 +0,0 @@
-from __future__ import annotations
-
-from contextlib import contextmanager
-from http.cookiejar import CookieJar
-from typing import TYPE_CHECKING, Any, Generator, Generic, Mapping, TypeVar, cast
-from warnings import warn
-
-from anyio.from_thread import BlockingPortal, start_blocking_portal
-from httpx import Cookies, Request, Response
-
-from litestar import Litestar
-from litestar.connection import ASGIConnection
-from litestar.datastructures import MutableScopeHeaders
-from litestar.enums import ScopeType
-from litestar.exceptions import (
- ImproperlyConfiguredException,
-)
-from litestar.types import AnyIOBackend, ASGIApp, HTTPResponseStartEvent
-from litestar.utils.scope.state import ScopeState
-
-if TYPE_CHECKING:
- from httpx._types import CookieTypes
-
- from litestar.middleware.session.base import BaseBackendConfig, BaseSessionBackend
- from litestar.types.asgi_types import HTTPScope, Receive, Scope, Send
-
-T = TypeVar("T", bound=ASGIApp)
-
-
-def fake_http_send_message(headers: MutableScopeHeaders) -> HTTPResponseStartEvent:
- headers.setdefault("content-type", "application/text")
- return HTTPResponseStartEvent(type="http.response.start", status=200, headers=headers.headers)
-
-
-def fake_asgi_connection(app: ASGIApp, cookies: dict[str, str]) -> ASGIConnection[Any, Any, Any, Any]:
- scope: HTTPScope = {
- "type": ScopeType.HTTP,
- "path": "/",
- "raw_path": b"/",
- "root_path": "",
- "scheme": "http",
- "query_string": b"",
- "client": ("testclient", 50000),
- "server": ("testserver", 80),
- "headers": [],
- "method": "GET",
- "http_version": "1.1",
- "extensions": {"http.response.template": {}},
- "app": app, # type: ignore[typeddict-item]
- "state": {},
- "path_params": {},
- "route_handler": None, # type: ignore[typeddict-item]
- "asgi": {"version": "3.0", "spec_version": "2.1"},
- "auth": None,
- "session": None,
- "user": None,
- }
- ScopeState.from_scope(scope).cookies = cookies
- return ASGIConnection[Any, Any, Any, Any](scope=scope)
-
-
-def _wrap_app_to_add_state(app: ASGIApp) -> ASGIApp:
- """Wrap an ASGI app to add state to the scope.
-
- Litestar depends on `state` being present in the ASGI connection scope. Scope state is optional in the ASGI spec,
- however, the Litestar app always ensures it is present so that it can be depended on internally.
-
- When the ASGI app that is passed to the test client is _not_ a Litestar app, we need to add
- state to the scope, because httpx does not do this for us.
-
- This assists us in testing Litestar components that rely on state being present in the scope, without having
- to create a Litestar app for every test case.
-
- Args:
- app: The ASGI app to wrap.
-
- Returns:
- The wrapped ASGI app.
- """
-
- async def wrapped(scope: Scope, receive: Receive, send: Send) -> None:
- scope["state"] = {}
- await app(scope, receive, send)
-
- return wrapped
-
-
-class BaseTestClient(Generic[T]):
- __test__ = False
- blocking_portal: BlockingPortal
-
- __slots__ = (
- "app",
- "base_url",
- "backend",
- "backend_options",
- "session_config",
- "_session_backend",
- "cookies",
- )
-
- def __init__(
- self,
- app: T,
- base_url: str = "http://testserver.local",
- backend: AnyIOBackend = "asyncio",
- backend_options: Mapping[str, Any] | None = None,
- session_config: BaseBackendConfig | None = None,
- cookies: CookieTypes | None = None,
- ) -> None:
- if "." not in base_url:
- warn(
- f"The base_url {base_url!r} might cause issues. Try adding a domain name such as .local: "
- f"'{base_url}.local'",
- UserWarning,
- stacklevel=1,
- )
-
- self._session_backend: BaseSessionBackend | None = None
- if session_config:
- self._session_backend = session_config._backend_class(config=session_config)
-
- if not isinstance(app, Litestar):
- app = _wrap_app_to_add_state(app) # type: ignore[assignment]
-
- self.app = cast("T", app) # type: ignore[redundant-cast] # pyright needs this
-
- self.base_url = base_url
- self.backend = backend
- self.backend_options = backend_options
- self.cookies = cookies
-
- @property
- def session_backend(self) -> BaseSessionBackend[Any]:
- if not self._session_backend:
- raise ImproperlyConfiguredException(
- "Session has not been initialized for this TestClient instance. You can"
- "do so by passing a configuration object to TestClient: TestClient(app=app, session_config=...)"
- )
- return self._session_backend
-
- @contextmanager
- def portal(self) -> Generator[BlockingPortal, None, None]:
- """Get a BlockingPortal.
-
- Returns:
- A contextmanager for a BlockingPortal.
- """
- if hasattr(self, "blocking_portal"):
- yield self.blocking_portal
- else:
- with start_blocking_portal(
- backend=self.backend, backend_options=dict(self.backend_options or {})
- ) as portal:
- yield portal
-
- async def _set_session_data(self, data: dict[str, Any]) -> None:
- mutable_headers = MutableScopeHeaders()
- connection = fake_asgi_connection(
- app=self.app,
- cookies=dict(self.cookies), # type: ignore[arg-type]
- )
- session_id = self.session_backend.get_session_id(connection)
- connection._connection_state.session_id = session_id # pyright: ignore [reportGeneralTypeIssues]
- await self.session_backend.store_in_message(
- scope_session=data, message=fake_http_send_message(mutable_headers), connection=connection
- )
- response = Response(200, request=Request("GET", self.base_url), headers=mutable_headers.headers)
-
- cookies = Cookies(CookieJar())
- cookies.extract_cookies(response)
- self.cookies.update(cookies) # type: ignore[union-attr]
-
- async def _get_session_data(self) -> dict[str, Any]:
- return await self.session_backend.load_from_connection(
- connection=fake_asgi_connection(
- app=self.app,
- cookies=dict(self.cookies), # type: ignore[arg-type]
- ),
- )
diff --git a/venv/lib/python3.11/site-packages/litestar/testing/client/sync_client.py b/venv/lib/python3.11/site-packages/litestar/testing/client/sync_client.py
deleted file mode 100644
index d907056..0000000
--- a/venv/lib/python3.11/site-packages/litestar/testing/client/sync_client.py
+++ /dev/null
@@ -1,593 +0,0 @@
-from __future__ import annotations
-
-from contextlib import ExitStack
-from typing import TYPE_CHECKING, Any, Generic, Mapping, Sequence, TypeVar
-from urllib.parse import urljoin
-
-from httpx import USE_CLIENT_DEFAULT, Client, Response
-
-from litestar import HttpMethod
-from litestar.testing.client.base import BaseTestClient
-from litestar.testing.life_span_handler import LifeSpanHandler
-from litestar.testing.transport import ConnectionUpgradeExceptionError, TestClientTransport
-from litestar.types import AnyIOBackend, ASGIApp
-
-if TYPE_CHECKING:
- from httpx._client import UseClientDefault
- from httpx._types import (
- AuthTypes,
- CookieTypes,
- HeaderTypes,
- QueryParamTypes,
- RequestContent,
- RequestData,
- RequestFiles,
- TimeoutTypes,
- URLTypes,
- )
- from typing_extensions import Self
-
- from litestar.middleware.session.base import BaseBackendConfig
- from litestar.testing.websocket_test_session import WebSocketTestSession
-
-
-T = TypeVar("T", bound=ASGIApp)
-
-
-class TestClient(Client, BaseTestClient, Generic[T]): # type: ignore[misc]
- lifespan_handler: LifeSpanHandler[Any]
- exit_stack: ExitStack
-
- def __init__(
- self,
- app: T,
- base_url: str = "http://testserver.local",
- raise_server_exceptions: bool = True,
- root_path: str = "",
- backend: AnyIOBackend = "asyncio",
- backend_options: Mapping[str, Any] | None = None,
- session_config: BaseBackendConfig | None = None,
- timeout: float | None = None,
- cookies: CookieTypes | None = None,
- ) -> None:
- """A client implementation providing a context manager for testing applications.
-
- Args:
- app: The instance of :class:`Litestar <litestar.app.Litestar>` under test.
- base_url: URL scheme and domain for test request paths, e.g. ``http://testserver``.
- raise_server_exceptions: Flag for the underlying test client to raise server exceptions instead of
- wrapping them in an HTTP response.
- root_path: Path prefix for requests.
- backend: The async backend to use, options are "asyncio" or "trio".
- backend_options: ``anyio`` options.
- session_config: Configuration for Session Middleware class to create raw session cookies for request to the
- route handlers.
- timeout: Request timeout
- cookies: Cookies to set on the client.
- """
- BaseTestClient.__init__(
- self,
- app=app,
- base_url=base_url,
- backend=backend,
- backend_options=backend_options,
- session_config=session_config,
- cookies=cookies,
- )
-
- Client.__init__(
- self,
- base_url=base_url,
- headers={"user-agent": "testclient"},
- follow_redirects=True,
- cookies=cookies,
- transport=TestClientTransport( # type: ignore[arg-type]
- client=self,
- raise_server_exceptions=raise_server_exceptions,
- root_path=root_path,
- ),
- timeout=timeout,
- )
-
- def __enter__(self) -> Self:
- with ExitStack() as stack:
- self.blocking_portal = portal = stack.enter_context(self.portal())
- self.lifespan_handler = LifeSpanHandler(client=self)
-
- @stack.callback
- def reset_portal() -> None:
- delattr(self, "blocking_portal")
-
- @stack.callback
- def wait_shutdown() -> None:
- portal.call(self.lifespan_handler.wait_shutdown)
-
- self.exit_stack = stack.pop_all()
-
- return self
-
- def __exit__(self, *args: Any) -> None:
- self.exit_stack.close()
-
- def request(
- self,
- method: str,
- url: URLTypes,
- *,
- content: RequestContent | None = None,
- data: RequestData | None = None,
- files: RequestFiles | None = None,
- json: Any | None = None,
- params: QueryParamTypes | None = None,
- headers: HeaderTypes | None = None,
- cookies: CookieTypes | None = None,
- auth: AuthTypes | UseClientDefault | None = USE_CLIENT_DEFAULT,
- follow_redirects: bool | UseClientDefault = USE_CLIENT_DEFAULT,
- timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
- extensions: Mapping[str, Any] | None = None,
- ) -> Response:
- """Sends a request.
-
- Args:
- method: An HTTP method.
- url: URL or path for the request.
- content: Request content.
- data: Form encoded data.
- files: Multipart files to send.
- json: JSON data to send.
- params: Query parameters.
- headers: Request headers.
- cookies: Request cookies.
- auth: Auth headers.
- follow_redirects: Whether to follow redirects.
- timeout: Request timeout.
- extensions: Dictionary of ASGI extensions.
-
- Returns:
- An HTTPX Response.
- """
- return Client.request(
- self,
- url=self.base_url.join(url),
- method=method.value if isinstance(method, HttpMethod) else method,
- content=content,
- data=data,
- files=files,
- json=json,
- params=params,
- headers=headers,
- cookies=cookies,
- auth=auth,
- follow_redirects=follow_redirects,
- timeout=timeout,
- extensions=None if extensions is None else dict(extensions),
- )
-
- def get(
- self,
- url: URLTypes,
- *,
- params: QueryParamTypes | None = None,
- headers: HeaderTypes | None = None,
- cookies: CookieTypes | None = None,
- auth: AuthTypes | UseClientDefault = USE_CLIENT_DEFAULT,
- follow_redirects: bool | UseClientDefault = USE_CLIENT_DEFAULT,
- timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
- extensions: Mapping[str, Any] | None = None,
- ) -> Response:
- """Sends a GET request.
-
- Args:
- url: URL or path for the request.
- params: Query parameters.
- headers: Request headers.
- cookies: Request cookies.
- auth: Auth headers.
- follow_redirects: Whether to follow redirects.
- timeout: Request timeout.
- extensions: Dictionary of ASGI extensions.
-
- Returns:
- An HTTPX Response.
- """
- return Client.get(
- self,
- url,
- params=params,
- headers=headers,
- cookies=cookies,
- auth=auth,
- follow_redirects=follow_redirects,
- timeout=timeout,
- extensions=None if extensions is None else dict(extensions),
- )
-
- def options(
- self,
- url: URLTypes,
- *,
- params: QueryParamTypes | None = None,
- headers: HeaderTypes | None = None,
- cookies: CookieTypes | None = None,
- auth: AuthTypes | UseClientDefault = USE_CLIENT_DEFAULT,
- follow_redirects: bool | UseClientDefault = USE_CLIENT_DEFAULT,
- timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
- extensions: Mapping[str, Any] | None = None,
- ) -> Response:
- """Sends an OPTIONS request.
-
- Args:
- url: URL or path for the request.
- params: Query parameters.
- headers: Request headers.
- cookies: Request cookies.
- auth: Auth headers.
- follow_redirects: Whether to follow redirects.
- timeout: Request timeout.
- extensions: Dictionary of ASGI extensions.
-
- Returns:
- An HTTPX Response.
- """
- return Client.options(
- self,
- url,
- params=params,
- headers=headers,
- cookies=cookies,
- auth=auth,
- follow_redirects=follow_redirects,
- timeout=timeout,
- extensions=None if extensions is None else dict(extensions),
- )
-
- def head(
- self,
- url: URLTypes,
- *,
- params: QueryParamTypes | None = None,
- headers: HeaderTypes | None = None,
- cookies: CookieTypes | None = None,
- auth: AuthTypes | UseClientDefault = USE_CLIENT_DEFAULT,
- follow_redirects: bool | UseClientDefault = USE_CLIENT_DEFAULT,
- timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
- extensions: Mapping[str, Any] | None = None,
- ) -> Response:
- """Sends a HEAD request.
-
- Args:
- url: URL or path for the request.
- params: Query parameters.
- headers: Request headers.
- cookies: Request cookies.
- auth: Auth headers.
- follow_redirects: Whether to follow redirects.
- timeout: Request timeout.
- extensions: Dictionary of ASGI extensions.
-
- Returns:
- An HTTPX Response.
- """
- return Client.head(
- self,
- url,
- params=params,
- headers=headers,
- cookies=cookies,
- auth=auth,
- follow_redirects=follow_redirects,
- timeout=timeout,
- extensions=None if extensions is None else dict(extensions),
- )
-
- def post(
- self,
- url: URLTypes,
- *,
- content: RequestContent | None = None,
- data: RequestData | None = None,
- files: RequestFiles | None = None,
- json: Any | None = None,
- params: QueryParamTypes | None = None,
- headers: HeaderTypes | None = None,
- cookies: CookieTypes | None = None,
- auth: AuthTypes | UseClientDefault = USE_CLIENT_DEFAULT,
- follow_redirects: bool | UseClientDefault = USE_CLIENT_DEFAULT,
- timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
- extensions: Mapping[str, Any] | None = None,
- ) -> Response:
- """Sends a POST request.
-
- Args:
- url: URL or path for the request.
- content: Request content.
- data: Form encoded data.
- files: Multipart files to send.
- json: JSON data to send.
- params: Query parameters.
- headers: Request headers.
- cookies: Request cookies.
- auth: Auth headers.
- follow_redirects: Whether to follow redirects.
- timeout: Request timeout.
- extensions: Dictionary of ASGI extensions.
-
- Returns:
- An HTTPX Response.
- """
- return Client.post(
- self,
- url,
- content=content,
- data=data,
- files=files,
- json=json,
- params=params,
- headers=headers,
- cookies=cookies,
- auth=auth,
- follow_redirects=follow_redirects,
- timeout=timeout,
- extensions=None if extensions is None else dict(extensions),
- )
-
- def put(
- self,
- url: URLTypes,
- *,
- content: RequestContent | None = None,
- data: RequestData | None = None,
- files: RequestFiles | None = None,
- json: Any | None = None,
- params: QueryParamTypes | None = None,
- headers: HeaderTypes | None = None,
- cookies: CookieTypes | None = None,
- auth: AuthTypes | UseClientDefault = USE_CLIENT_DEFAULT,
- follow_redirects: bool | UseClientDefault = USE_CLIENT_DEFAULT,
- timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
- extensions: Mapping[str, Any] | None = None,
- ) -> Response:
- """Sends a PUT request.
-
- Args:
- url: URL or path for the request.
- content: Request content.
- data: Form encoded data.
- files: Multipart files to send.
- json: JSON data to send.
- params: Query parameters.
- headers: Request headers.
- cookies: Request cookies.
- auth: Auth headers.
- follow_redirects: Whether to follow redirects.
- timeout: Request timeout.
- extensions: Dictionary of ASGI extensions.
-
- Returns:
- An HTTPX Response.
- """
- return Client.put(
- self,
- url,
- content=content,
- data=data,
- files=files,
- json=json,
- params=params,
- headers=headers,
- cookies=cookies,
- auth=auth,
- follow_redirects=follow_redirects,
- timeout=timeout,
- extensions=None if extensions is None else dict(extensions),
- )
-
- def patch(
- self,
- url: URLTypes,
- *,
- content: RequestContent | None = None,
- data: RequestData | None = None,
- files: RequestFiles | None = None,
- json: Any | None = None,
- params: QueryParamTypes | None = None,
- headers: HeaderTypes | None = None,
- cookies: CookieTypes | None = None,
- auth: AuthTypes | UseClientDefault = USE_CLIENT_DEFAULT,
- follow_redirects: bool | UseClientDefault = USE_CLIENT_DEFAULT,
- timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
- extensions: Mapping[str, Any] | None = None,
- ) -> Response:
- """Sends a PATCH request.
-
- Args:
- url: URL or path for the request.
- content: Request content.
- data: Form encoded data.
- files: Multipart files to send.
- json: JSON data to send.
- params: Query parameters.
- headers: Request headers.
- cookies: Request cookies.
- auth: Auth headers.
- follow_redirects: Whether to follow redirects.
- timeout: Request timeout.
- extensions: Dictionary of ASGI extensions.
-
- Returns:
- An HTTPX Response.
- """
- return Client.patch(
- self,
- url,
- content=content,
- data=data,
- files=files,
- json=json,
- params=params,
- headers=headers,
- cookies=cookies,
- auth=auth,
- follow_redirects=follow_redirects,
- timeout=timeout,
- extensions=None if extensions is None else dict(extensions),
- )
-
- def delete(
- self,
- url: URLTypes,
- *,
- params: QueryParamTypes | None = None,
- headers: HeaderTypes | None = None,
- cookies: CookieTypes | None = None,
- auth: AuthTypes | UseClientDefault = USE_CLIENT_DEFAULT,
- follow_redirects: bool | UseClientDefault = USE_CLIENT_DEFAULT,
- timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
- extensions: Mapping[str, Any] | None = None,
- ) -> Response:
- """Sends a DELETE request.
-
- Args:
- url: URL or path for the request.
- params: Query parameters.
- headers: Request headers.
- cookies: Request cookies.
- auth: Auth headers.
- follow_redirects: Whether to follow redirects.
- timeout: Request timeout.
- extensions: Dictionary of ASGI extensions.
-
- Returns:
- An HTTPX Response.
- """
- return Client.delete(
- self,
- url,
- params=params,
- headers=headers,
- cookies=cookies,
- auth=auth,
- follow_redirects=follow_redirects,
- timeout=timeout,
- extensions=None if extensions is None else dict(extensions),
- )
-
- def websocket_connect(
- self,
- url: str,
- subprotocols: Sequence[str] | None = None,
- params: QueryParamTypes | None = None,
- headers: HeaderTypes | None = None,
- cookies: CookieTypes | None = None,
- auth: AuthTypes | UseClientDefault = USE_CLIENT_DEFAULT,
- follow_redirects: bool | UseClientDefault = USE_CLIENT_DEFAULT,
- timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
- extensions: Mapping[str, Any] | None = None,
- ) -> WebSocketTestSession:
- """Sends a GET request to establish a websocket connection.
-
- Args:
- url: Request URL.
- subprotocols: Websocket subprotocols.
- params: Query parameters.
- headers: Request headers.
- cookies: Request cookies.
- auth: Auth headers.
- follow_redirects: Whether to follow redirects.
- timeout: Request timeout.
- extensions: Dictionary of ASGI extensions.
-
- Returns:
- A `WebSocketTestSession <litestar.testing.WebSocketTestSession>` instance.
- """
- url = urljoin("ws://testserver", url)
- default_headers: dict[str, str] = {}
- default_headers.setdefault("connection", "upgrade")
- default_headers.setdefault("sec-websocket-key", "testserver==")
- default_headers.setdefault("sec-websocket-version", "13")
- if subprotocols is not None:
- default_headers.setdefault("sec-websocket-protocol", ", ".join(subprotocols))
- try:
- Client.request(
- self,
- "GET",
- url,
- headers={**dict(headers or {}), **default_headers}, # type: ignore[misc]
- params=params,
- cookies=cookies,
- auth=auth,
- follow_redirects=follow_redirects,
- timeout=timeout,
- extensions=None if extensions is None else dict(extensions),
- )
- except ConnectionUpgradeExceptionError as exc:
- return exc.session
-
- raise RuntimeError("Expected WebSocket upgrade") # pragma: no cover
-
- def set_session_data(self, data: dict[str, Any]) -> None:
- """Set session data.
-
- Args:
- data: Session data
-
- Returns:
- None
-
- Examples:
- .. code-block:: python
-
- from litestar import Litestar, get
- from litestar.middleware.session.memory_backend import MemoryBackendConfig
-
- session_config = MemoryBackendConfig()
-
-
- @get(path="/test")
- def get_session_data(request: Request) -> Dict[str, Any]:
- return request.session
-
-
- app = Litestar(
- route_handlers=[get_session_data], middleware=[session_config.middleware]
- )
-
- with TestClient(app=app, session_config=session_config) as client:
- client.set_session_data({"foo": "bar"})
- assert client.get("/test").json() == {"foo": "bar"}
-
- """
- with self.portal() as portal:
- portal.call(self._set_session_data, data)
-
- def get_session_data(self) -> dict[str, Any]:
- """Get session data.
-
- Returns:
- A dictionary containing session data.
-
- Examples:
- .. code-block:: python
-
- from litestar import Litestar, post
- from litestar.middleware.session.memory_backend import MemoryBackendConfig
-
- session_config = MemoryBackendConfig()
-
-
- @post(path="/test")
- def set_session_data(request: Request) -> None:
- request.session["foo"] == "bar"
-
-
- app = Litestar(
- route_handlers=[set_session_data], middleware=[session_config.middleware]
- )
-
- with TestClient(app=app, session_config=session_config) as client:
- client.post("/test")
- assert client.get_session_data() == {"foo": "bar"}
-
- """
- with self.portal() as portal:
- return portal.call(self._get_session_data)