diff --git a/docs/applications.md b/docs/applications.md index 6fb74f19f..439183cbb 100644 --- a/docs/applications.md +++ b/docs/applications.md @@ -3,6 +3,8 @@ Starlette includes an application class `Starlette` that nicely ties together al its other functionality. ```python +from contextlib import asynccontextmanager + from starlette.applications import Starlette from starlette.responses import PlainTextResponse from starlette.routing import Route, Mount, WebSocketRoute @@ -25,8 +27,11 @@ async def websocket_endpoint(websocket): await websocket.send_text('Hello, websocket!') await websocket.close() -def startup(): +@asynccontextmanager +async def lifespan(app: Starlette): print('Ready to go') + yield + print('Shutting down') routes = [ @@ -37,7 +42,7 @@ routes = [ Mount('/static', StaticFiles(directory="static")), ] -app = Starlette(debug=True, routes=routes, on_startup=[startup]) +app = Starlette(debug=True, routes=routes, lifespan=lifespan) ``` ### Instantiating the application diff --git a/docs/release-notes.md b/docs/release-notes.md index 7d78e824d..4798895d6 100644 --- a/docs/release-notes.md +++ b/docs/release-notes.md @@ -1,3 +1,19 @@ +## 1.0.0 + +The project versioning policy is now explicitly governed by SEMVER. See https://semver.org/. + +### Removed + +All deprecated features have been removed. + +* Removed `WSGIMiddleware`. Please use [`a2wsgi`](https://github.com/abersheeran/a2wsgi) instead. +* Removed `run_until_first_complete`. +* Removed `on_startup` and `on_shutdown` events. Please use `lifespan` instead. +* Removed `iscoroutinefunction_or_partial`, which we have replaced by `_utils.is_async_callable`. +* Removed `WS_1004_NO_STATUS_RCVD` and `WS_1005_ABNORMAL_CLOSURE` from the `status` module. +* Removed `ExceptionMiddleware` from the `exceptions` module, it can now be found in the `middleware.exceptions` module. +* Removed multiple possible argument sequences from `TemplateResponse`. + ## 0.38.1 July 23, 2024 diff --git a/docs/templates.md b/docs/templates.md index 01f343238..ea026d846 100644 --- a/docs/templates.md +++ b/docs/templates.md @@ -3,11 +3,11 @@ Jinja2 provides an excellent choice. ### Jinja2Templates -Signature: `Jinja2Templates(directory, context_processors=None, **env_options)` +Signature: `Jinja2Templates(directory, context_processors=None, env=None)` * `directory` - A string, [os.Pathlike][pathlike] or a list of strings or [os.Pathlike][pathlike] denoting a directory path. * `context_processors` - A list of functions that return a dictionary to add to the template context. -* `**env_options` - Additional keyword arguments to pass to the Jinja2 environment. +* `env` - A preconfigured [`jinja2.Environment`](https://jinja.palletsprojects.com/en/3.0.x/api/#api) instance. Starlette provides a simple way to get `jinja2` configured. This is probably what you want to use by default. @@ -58,11 +58,9 @@ templates = Jinja2Templates(directory='templates') templates.env.filters['marked'] = marked_filter ``` +## Using custom `jinja2.Environment` instance -## Using custom jinja2.Environment instance - -Starlette also accepts a preconfigured [`jinja2.Environment`](https://jinja.palletsprojects.com/en/3.0.x/api/#api) instance. - +Starlette accepts a preconfigured [`jinja2.Environment`](https://jinja.palletsprojects.com/en/3.0.x/api/#api) instance. ```python import jinja2 @@ -72,7 +70,6 @@ env = jinja2.Environment(...) templates = Jinja2Templates(env=env) ``` - ## Context processors A context processor is a function that returns a dictionary to be merged into a template context. @@ -126,20 +123,6 @@ def test_homepage(): assert "request" in response.context ``` -## Customizing Jinja2 Environment - -`Jinja2Templates` accepts all options supported by Jinja2 `Environment`. -This will allow more control over the `Environment` instance created by Starlette. - -For the list of options available to `Environment` you can check Jinja2 documentation [here](https://jinja.palletsprojects.com/en/3.0.x/api/#jinja2.Environment) - -```python -from starlette.templating import Jinja2Templates - - -templates = Jinja2Templates(directory='templates', autoescape=False, auto_reload=True) -``` - ## Asynchronous template rendering Jinja2 supports async template rendering, however as a general rule diff --git a/pyproject.toml b/pyproject.toml index 2fd7fb482..6cfe92426 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -70,12 +70,9 @@ xfail_strict = true filterwarnings = [ # Turn warnings that aren't filtered into exceptions "error", - "ignore: run_until_first_complete is deprecated and will be removed in a future version.:DeprecationWarning", - "ignore: starlette.middleware.wsgi is deprecated and will be removed in a future release.*:DeprecationWarning", "ignore: Async generator 'starlette.requests.Request.stream' was garbage collected before it had been exhausted.*:ResourceWarning", "ignore: path is deprecated.*:DeprecationWarning:certifi", "ignore: Use 'content=<...>' to upload raw bytes/text content.:DeprecationWarning", - "ignore: The `allow_redirects` argument is deprecated. Use `follow_redirects` instead.:DeprecationWarning", "ignore: 'cgi' is deprecated and slated for removal in Python 3.13:DeprecationWarning", "ignore: You seem to already have a custom sys.excepthook handler installed. I'll skip installing Trio's custom handler, but this means MultiErrors will not show full tracebacks.:RuntimeWarning", ] diff --git a/starlette/__init__.py b/starlette/__init__.py index 53b39ad36..5becc17c0 100644 --- a/starlette/__init__.py +++ b/starlette/__init__.py @@ -1 +1 @@ -__version__ = "0.38.1" +__version__ = "1.0.0" diff --git a/starlette/applications.py b/starlette/applications.py index 913fd4c9d..e2d0178b6 100644 --- a/starlette/applications.py +++ b/starlette/applications.py @@ -2,7 +2,6 @@ import sys import typing -import warnings if sys.version_info >= (3, 10): # pragma: no cover from typing import ParamSpec @@ -11,7 +10,6 @@ from starlette.datastructures import State, URLPath from starlette.middleware import Middleware, _MiddlewareClass -from starlette.middleware.base import BaseHTTPMiddleware from starlette.middleware.errors import ServerErrorMiddleware from starlette.middleware.exceptions import ExceptionMiddleware from starlette.requests import Request @@ -43,15 +41,8 @@ class Starlette: Exception handler callables should be of the form `handler(request, exc) -> response` and may be either standard functions, or async functions. - * **on_startup** - A list of callables to run on application startup. - Startup handler callables do not take any arguments, and may be either - standard functions, or async functions. - * **on_shutdown** - A list of callables to run on application shutdown. - Shutdown handler callables do not take any arguments, and may be either - standard functions, or async functions. * **lifespan** - A lifespan context function, which can be used to perform - startup and shutdown tasks. This is a newer style that replaces the - `on_startup` and `on_shutdown` handlers. Use one or the other, not both. + startup and shutdown tasks. """ def __init__( @@ -60,21 +51,11 @@ def __init__( routes: typing.Sequence[BaseRoute] | None = None, middleware: typing.Sequence[Middleware] | None = None, exception_handlers: typing.Mapping[typing.Any, ExceptionHandler] | None = None, - on_startup: typing.Sequence[typing.Callable[[], typing.Any]] | None = None, - on_shutdown: typing.Sequence[typing.Callable[[], typing.Any]] | None = None, lifespan: Lifespan[AppType] | None = None, ) -> None: - # The lifespan context function is a newer style that replaces - # on_startup / on_shutdown handlers. Use one or the other, not both. - assert lifespan is None or ( - on_startup is None and on_shutdown is None - ), "Use either 'lifespan' or 'on_startup'/'on_shutdown', not both." - self.debug = debug self.state = State() - self.router = Router( - routes, on_startup=on_startup, on_shutdown=on_shutdown, lifespan=lifespan - ) + self.router = Router(routes, lifespan=lifespan) self.exception_handlers = ( {} if exception_handlers is None else dict(exception_handlers) ) @@ -122,9 +103,6 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: self.middleware_stack = self.build_middleware_stack() await self.middleware_stack(scope, receive, send) - def on_event(self, event_type: str) -> typing.Callable: # type: ignore[type-arg] - return self.router.on_event(event_type) # pragma: nocover - def mount(self, path: str, app: ASGIApp, name: str | None = None) -> None: self.router.mount(path, app=app, name=name) # pragma: no cover @@ -148,13 +126,6 @@ def add_exception_handler( ) -> None: # pragma: no cover self.exception_handlers[exc_class_or_status_code] = handler - def add_event_handler( - self, - event_type: str, - func: typing.Callable, # type: ignore[type-arg] - ) -> None: # pragma: no cover - self.router.add_event_handler(event_type, func) - def add_route( self, path: str, @@ -174,93 +145,3 @@ def add_websocket_route( name: str | None = None, ) -> None: # pragma: no cover self.router.add_websocket_route(path, route, name=name) - - def exception_handler( - self, exc_class_or_status_code: int | type[Exception] - ) -> typing.Callable: # type: ignore[type-arg] - warnings.warn( - "The `exception_handler` decorator is deprecated, and will be removed in version 1.0.0. " # noqa: E501 - "Refer to https://www.starlette.io/exceptions/ for the recommended approach.", # noqa: E501 - DeprecationWarning, - ) - - def decorator(func: typing.Callable) -> typing.Callable: # type: ignore[type-arg] # noqa: E501 - self.add_exception_handler(exc_class_or_status_code, func) - return func - - return decorator - - def route( - self, - path: str, - methods: list[str] | None = None, - name: str | None = None, - include_in_schema: bool = True, - ) -> typing.Callable: # type: ignore[type-arg] - """ - We no longer document this decorator style API, and its usage is discouraged. - Instead you should use the following approach: - - >>> routes = [Route(path, endpoint=...), ...] - >>> app = Starlette(routes=routes) - """ - warnings.warn( - "The `route` decorator is deprecated, and will be removed in version 1.0.0. " # noqa: E501 - "Refer to https://www.starlette.io/routing/ for the recommended approach.", # noqa: E501 - DeprecationWarning, - ) - - def decorator(func: typing.Callable) -> typing.Callable: # type: ignore[type-arg] # noqa: E501 - self.router.add_route( - path, - func, - methods=methods, - name=name, - include_in_schema=include_in_schema, - ) - return func - - return decorator - - def websocket_route(self, path: str, name: str | None = None) -> typing.Callable: # type: ignore[type-arg] - """ - We no longer document this decorator style API, and its usage is discouraged. - Instead you should use the following approach: - - >>> routes = [WebSocketRoute(path, endpoint=...), ...] - >>> app = Starlette(routes=routes) - """ - warnings.warn( - "The `websocket_route` decorator is deprecated, and will be removed in version 1.0.0. " # noqa: E501 - "Refer to https://www.starlette.io/routing/#websocket-routing for the recommended approach.", # noqa: E501 - DeprecationWarning, - ) - - def decorator(func: typing.Callable) -> typing.Callable: # type: ignore[type-arg] # noqa: E501 - self.router.add_websocket_route(path, func, name=name) - return func - - return decorator - - def middleware(self, middleware_type: str) -> typing.Callable: # type: ignore[type-arg] # noqa: E501 - """ - We no longer document this decorator style API, and its usage is discouraged. - Instead you should use the following approach: - - >>> middleware = [Middleware(...), ...] - >>> app = Starlette(middleware=middleware) - """ - warnings.warn( - "The `middleware` decorator is deprecated, and will be removed in version 1.0.0. " # noqa: E501 - "Refer to https://www.starlette.io/middleware/#using-middleware for recommended approach.", # noqa: E501 - DeprecationWarning, - ) - assert ( - middleware_type == "http" - ), 'Currently only middleware("http") is supported.' - - def decorator(func: typing.Callable) -> typing.Callable: # type: ignore[type-arg] # noqa: E501 - self.add_middleware(BaseHTTPMiddleware, dispatch=func) - return func - - return decorator diff --git a/starlette/concurrency.py b/starlette/concurrency.py index 215e3a63b..5c929d1ef 100644 --- a/starlette/concurrency.py +++ b/starlette/concurrency.py @@ -3,7 +3,6 @@ import functools import sys import typing -import warnings import anyio.to_thread @@ -16,23 +15,6 @@ T = typing.TypeVar("T") -async def run_until_first_complete(*args: tuple[typing.Callable, dict]) -> None: # type: ignore[type-arg] # noqa: E501 - warnings.warn( - "run_until_first_complete is deprecated " - "and will be removed in a future version.", - DeprecationWarning, - ) - - async with anyio.create_task_group() as task_group: - - async def run(func: typing.Callable[[], typing.Coroutine]) -> None: # type: ignore[type-arg] # noqa: E501 - await func() - task_group.cancel_scope.cancel() - - for func, kwargs in args: - task_group.start_soon(run, functools.partial(func, **kwargs)) - - async def run_in_threadpool( func: typing.Callable[P, T], *args: P.args, **kwargs: P.kwargs ) -> T: diff --git a/starlette/exceptions.py b/starlette/exceptions.py index bd3352eb0..70a4f436f 100644 --- a/starlette/exceptions.py +++ b/starlette/exceptions.py @@ -1,8 +1,6 @@ from __future__ import annotations import http -import typing -import warnings __all__ = ("HTTPException", "WebSocketException") @@ -39,24 +37,3 @@ def __str__(self) -> str: def __repr__(self) -> str: class_name = self.__class__.__name__ return f"{class_name}(code={self.code!r}, reason={self.reason!r})" - - -__deprecated__ = "ExceptionMiddleware" - - -def __getattr__(name: str) -> typing.Any: # pragma: no cover - if name == __deprecated__: - from starlette.middleware.exceptions import ExceptionMiddleware - - warnings.warn( - f"{__deprecated__} is deprecated on `starlette.exceptions`. " - f"Import it from `starlette.middleware.exceptions` instead.", - category=DeprecationWarning, - stacklevel=3, - ) - return ExceptionMiddleware - raise AttributeError(f"module '{__name__}' has no attribute '{name}'") - - -def __dir__() -> list[str]: - return sorted(list(__all__) + [__deprecated__]) # pragma: no cover diff --git a/starlette/middleware/wsgi.py b/starlette/middleware/wsgi.py deleted file mode 100644 index c9a7e1328..000000000 --- a/starlette/middleware/wsgi.py +++ /dev/null @@ -1,156 +0,0 @@ -from __future__ import annotations - -import io -import math -import sys -import typing -import warnings - -import anyio -from anyio.abc import ObjectReceiveStream, ObjectSendStream - -from starlette.types import Receive, Scope, Send - -warnings.warn( - "starlette.middleware.wsgi is deprecated and will be removed in a future release. " - "Please refer to https://github.com/abersheeran/a2wsgi as a replacement.", - DeprecationWarning, -) - - -def build_environ(scope: Scope, body: bytes) -> dict[str, typing.Any]: - """ - Builds a scope and request body into a WSGI environ object. - """ - - script_name = scope.get("root_path", "").encode("utf8").decode("latin1") - path_info = scope["path"].encode("utf8").decode("latin1") - if path_info.startswith(script_name): - path_info = path_info[len(script_name) :] - - environ = { - "REQUEST_METHOD": scope["method"], - "SCRIPT_NAME": script_name, - "PATH_INFO": path_info, - "QUERY_STRING": scope["query_string"].decode("ascii"), - "SERVER_PROTOCOL": f"HTTP/{scope['http_version']}", - "wsgi.version": (1, 0), - "wsgi.url_scheme": scope.get("scheme", "http"), - "wsgi.input": io.BytesIO(body), - "wsgi.errors": sys.stdout, - "wsgi.multithread": True, - "wsgi.multiprocess": True, - "wsgi.run_once": False, - } - - # Get server name and port - required in WSGI, not in ASGI - server = scope.get("server") or ("localhost", 80) - environ["SERVER_NAME"] = server[0] - environ["SERVER_PORT"] = server[1] - - # Get client IP address - if scope.get("client"): - environ["REMOTE_ADDR"] = scope["client"][0] - - # Go through headers and make them into environ entries - for name, value in scope.get("headers", []): - name = name.decode("latin1") - if name == "content-length": - corrected_name = "CONTENT_LENGTH" - elif name == "content-type": - corrected_name = "CONTENT_TYPE" - else: - corrected_name = f"HTTP_{name}".upper().replace("-", "_") - # HTTPbis say only ASCII chars are allowed in headers, but we latin1 just in - # case - value = value.decode("latin1") - if corrected_name in environ: - value = environ[corrected_name] + "," + value - environ[corrected_name] = value - return environ - - -class WSGIMiddleware: - def __init__(self, app: typing.Callable[..., typing.Any]) -> None: - self.app = app - - async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: - assert scope["type"] == "http" - responder = WSGIResponder(self.app, scope) - await responder(receive, send) - - -class WSGIResponder: - stream_send: ObjectSendStream[typing.MutableMapping[str, typing.Any]] - stream_receive: ObjectReceiveStream[typing.MutableMapping[str, typing.Any]] - - def __init__(self, app: typing.Callable[..., typing.Any], scope: Scope) -> None: - self.app = app - self.scope = scope - self.status = None - self.response_headers = None - self.stream_send, self.stream_receive = anyio.create_memory_object_stream( - math.inf - ) - self.response_started = False - self.exc_info: typing.Any = None - - async def __call__(self, receive: Receive, send: Send) -> None: - body = b"" - more_body = True - while more_body: - message = await receive() - body += message.get("body", b"") - more_body = message.get("more_body", False) - environ = build_environ(self.scope, body) - - async with anyio.create_task_group() as task_group: - task_group.start_soon(self.sender, send) - async with self.stream_send: - await anyio.to_thread.run_sync(self.wsgi, environ, self.start_response) - if self.exc_info is not None: - raise self.exc_info[0].with_traceback(self.exc_info[1], self.exc_info[2]) - - async def sender(self, send: Send) -> None: - async with self.stream_receive: - async for message in self.stream_receive: - await send(message) - - def start_response( - self, - status: str, - response_headers: list[tuple[str, str]], - exc_info: typing.Any = None, - ) -> None: - self.exc_info = exc_info - if not self.response_started: - self.response_started = True - status_code_string, _ = status.split(" ", 1) - status_code = int(status_code_string) - headers = [ - (name.strip().encode("ascii").lower(), value.strip().encode("ascii")) - for name, value in response_headers - ] - anyio.from_thread.run( - self.stream_send.send, - { - "type": "http.response.start", - "status": status_code, - "headers": headers, - }, - ) - - def wsgi( - self, - environ: dict[str, typing.Any], - start_response: typing.Callable[..., typing.Any], - ) -> None: - for chunk in self.app(environ, start_response): - anyio.from_thread.run( - self.stream_send.send, - {"type": "http.response.body", "body": chunk, "more_body": True}, - ) - - anyio.from_thread.run( - self.stream_send.send, {"type": "http.response.body", "body": b""} - ) diff --git a/starlette/responses.py b/starlette/responses.py index 1f41c23d4..ac2eb47d2 100644 --- a/starlette/responses.py +++ b/starlette/responses.py @@ -5,7 +5,6 @@ import os import stat import typing -import warnings from datetime import datetime from email.utils import format_datetime, formatdate from functools import partial @@ -280,17 +279,11 @@ def __init__( background: BackgroundTask | None = None, filename: str | None = None, stat_result: os.stat_result | None = None, - method: str | None = None, content_disposition_type: str = "attachment", ) -> None: self.path = path self.status_code = status_code self.filename = filename - if method is not None: - warnings.warn( - "The 'method' parameter is not used, and it will be removed.", - DeprecationWarning, - ) if media_type is None: media_type = guess_type(filename or path)[0] or "text/plain" self.media_type = media_type diff --git a/starlette/routing.py b/starlette/routing.py index 75a5ec3f3..d23a53324 100644 --- a/starlette/routing.py +++ b/starlette/routing.py @@ -1,14 +1,10 @@ from __future__ import annotations -import contextlib import functools import inspect import re import traceback -import types import typing -import warnings -from contextlib import asynccontextmanager from enum import Enum from starlette._exception_handler import wrap_app_handling_exceptions @@ -41,21 +37,6 @@ class Match(Enum): FULL = 2 -def iscoroutinefunction_or_partial(obj: typing.Any) -> bool: # pragma: no cover - """ - Correctly determines if an object is a coroutine function, - including those wrapped in functools.partial objects. - """ - warnings.warn( - "iscoroutinefunction_or_partial is deprecated, " - "and will be removed in a future release.", - DeprecationWarning, - ) - while isinstance(obj, functools.partial): - obj = obj.func - return inspect.iscoroutinefunction(obj) - - def request_response( func: typing.Callable[[Request], typing.Awaitable[Response] | Response], ) -> ASGIApp: @@ -570,45 +551,15 @@ def __repr__(self) -> str: _T = typing.TypeVar("_T") -class _AsyncLiftContextManager(typing.AsyncContextManager[_T]): - def __init__(self, cm: typing.ContextManager[_T]): - self._cm = cm - - async def __aenter__(self) -> _T: - return self._cm.__enter__() - - async def __aexit__( - self, - exc_type: type[BaseException] | None, - exc_value: BaseException | None, - traceback: types.TracebackType | None, - ) -> bool | None: - return self._cm.__exit__(exc_type, exc_value, traceback) - - -def _wrap_gen_lifespan_context( - lifespan_context: typing.Callable[ - [typing.Any], typing.Generator[typing.Any, typing.Any, typing.Any] - ], -) -> typing.Callable[[typing.Any], typing.AsyncContextManager[typing.Any]]: - cmgr = contextlib.contextmanager(lifespan_context) - - @functools.wraps(cmgr) - def wrapper(app: typing.Any) -> _AsyncLiftContextManager[typing.Any]: - return _AsyncLiftContextManager(cmgr(app)) - - return wrapper - - class _DefaultLifespan: def __init__(self, router: Router): self._router = router async def __aenter__(self) -> None: - await self._router.startup() + ... async def __aexit__(self, *exc_info: object) -> None: - await self._router.shutdown() + ... def __call__(self: _T, app: object) -> _T: return self @@ -620,8 +571,6 @@ def __init__( routes: typing.Sequence[BaseRoute] | None = None, redirect_slashes: bool = True, default: ASGIApp | None = None, - on_startup: typing.Sequence[typing.Callable[[], typing.Any]] | None = None, - on_shutdown: typing.Sequence[typing.Callable[[], typing.Any]] | None = None, # the generic to Lifespan[AppType] is the type of the top level application # which the router cannot know statically, so we use typing.Any lifespan: Lifespan[typing.Any] | None = None, @@ -631,44 +580,9 @@ def __init__( self.routes = [] if routes is None else list(routes) self.redirect_slashes = redirect_slashes self.default = self.not_found if default is None else default - self.on_startup = [] if on_startup is None else list(on_startup) - self.on_shutdown = [] if on_shutdown is None else list(on_shutdown) - - if on_startup or on_shutdown: - warnings.warn( - "The on_startup and on_shutdown parameters are deprecated, and they " - "will be removed on version 1.0. Use the lifespan parameter instead. " - "See more about it on https://www.starlette.io/lifespan/.", - DeprecationWarning, - ) - if lifespan: - warnings.warn( - "The `lifespan` parameter cannot be used with `on_startup` or " - "`on_shutdown`. Both `on_startup` and `on_shutdown` will be " - "ignored." - ) if lifespan is None: self.lifespan_context: Lifespan[typing.Any] = _DefaultLifespan(self) - - elif inspect.isasyncgenfunction(lifespan): - warnings.warn( - "async generator function lifespans are deprecated, " - "use an @contextlib.asynccontextmanager function instead", - DeprecationWarning, - ) - self.lifespan_context = asynccontextmanager( - lifespan, - ) - elif inspect.isgeneratorfunction(lifespan): - warnings.warn( - "generator function lifespans are deprecated, " - "use an @contextlib.asynccontextmanager function instead", - DeprecationWarning, - ) - self.lifespan_context = _wrap_gen_lifespan_context( - lifespan, - ) else: self.lifespan_context = lifespan @@ -700,26 +614,6 @@ def url_path_for(self, name: str, /, **path_params: typing.Any) -> URLPath: pass raise NoMatchFound(name, path_params) - async def startup(self) -> None: - """ - Run any `.on_startup` event handlers. - """ - for handler in self.on_startup: - if is_async_callable(handler): - await handler() - else: - handler() - - async def shutdown(self) -> None: - """ - Run any `.on_shutdown` event handlers. - """ - for handler in self.on_shutdown: - if is_async_callable(handler): - await handler() - else: - handler() - async def lifespan(self, scope: Scope, receive: Receive, send: Send) -> None: """ Handle ASGI lifespan messages, which allows us to manage application @@ -810,7 +704,7 @@ def __eq__(self, other: typing.Any) -> bool: def mount( self, path: str, app: ASGIApp, name: str | None = None - ) -> None: # pragma: nocover + ) -> None: # pragma: no cover route = Mount(path, app=app, name=name) self.routes.append(route) @@ -827,7 +721,7 @@ def add_route( methods: list[str] | None = None, name: str | None = None, include_in_schema: bool = True, - ) -> None: # pragma: nocover + ) -> None: # pragma: no cover route = Route( path, endpoint=endpoint, @@ -845,78 +739,3 @@ def add_websocket_route( ) -> None: # pragma: no cover route = WebSocketRoute(path, endpoint=endpoint, name=name) self.routes.append(route) - - def route( - self, - path: str, - methods: list[str] | None = None, - name: str | None = None, - include_in_schema: bool = True, - ) -> typing.Callable: # type: ignore[type-arg] - """ - We no longer document this decorator style API, and its usage is discouraged. - Instead you should use the following approach: - - >>> routes = [Route(path, endpoint=...), ...] - >>> app = Starlette(routes=routes) - """ - warnings.warn( - "The `route` decorator is deprecated, and will be removed in version 1.0.0." - "Refer to https://www.starlette.io/routing/#http-routing for the recommended approach.", # noqa: E501 - DeprecationWarning, - ) - - def decorator(func: typing.Callable) -> typing.Callable: # type: ignore[type-arg] # noqa: E501 - self.add_route( - path, - func, - methods=methods, - name=name, - include_in_schema=include_in_schema, - ) - return func - - return decorator - - def websocket_route(self, path: str, name: str | None = None) -> typing.Callable: # type: ignore[type-arg] - """ - We no longer document this decorator style API, and its usage is discouraged. - Instead you should use the following approach: - - >>> routes = [WebSocketRoute(path, endpoint=...), ...] - >>> app = Starlette(routes=routes) - """ - warnings.warn( - "The `websocket_route` decorator is deprecated, and will be removed in version 1.0.0. Refer to " # noqa: E501 - "https://www.starlette.io/routing/#websocket-routing for the recommended approach.", # noqa: E501 - DeprecationWarning, - ) - - def decorator(func: typing.Callable) -> typing.Callable: # type: ignore[type-arg] # noqa: E501 - self.add_websocket_route(path, func, name=name) - return func - - return decorator - - def add_event_handler( - self, event_type: str, func: typing.Callable[[], typing.Any] - ) -> None: # pragma: no cover - assert event_type in ("startup", "shutdown") - - if event_type == "startup": - self.on_startup.append(func) - else: - self.on_shutdown.append(func) - - def on_event(self, event_type: str) -> typing.Callable: # type: ignore[type-arg] - warnings.warn( - "The `on_event` decorator is deprecated, and will be removed in version 1.0.0. " # noqa: E501 - "Refer to https://www.starlette.io/lifespan/ for recommended approach.", - DeprecationWarning, - ) - - def decorator(func: typing.Callable) -> typing.Callable: # type: ignore[type-arg] # noqa: E501 - self.add_event_handler(event_type, func) - return func - - return decorator diff --git a/starlette/status.py b/starlette/status.py index 2cd5db575..8d2800774 100644 --- a/starlette/status.py +++ b/starlette/status.py @@ -5,90 +5,6 @@ And RFC 2324 - https://tools.ietf.org/html/rfc2324 """ -from __future__ import annotations - -import warnings - -__all__ = ( - "HTTP_100_CONTINUE", - "HTTP_101_SWITCHING_PROTOCOLS", - "HTTP_102_PROCESSING", - "HTTP_103_EARLY_HINTS", - "HTTP_200_OK", - "HTTP_201_CREATED", - "HTTP_202_ACCEPTED", - "HTTP_203_NON_AUTHORITATIVE_INFORMATION", - "HTTP_204_NO_CONTENT", - "HTTP_205_RESET_CONTENT", - "HTTP_206_PARTIAL_CONTENT", - "HTTP_207_MULTI_STATUS", - "HTTP_208_ALREADY_REPORTED", - "HTTP_226_IM_USED", - "HTTP_300_MULTIPLE_CHOICES", - "HTTP_301_MOVED_PERMANENTLY", - "HTTP_302_FOUND", - "HTTP_303_SEE_OTHER", - "HTTP_304_NOT_MODIFIED", - "HTTP_305_USE_PROXY", - "HTTP_306_RESERVED", - "HTTP_307_TEMPORARY_REDIRECT", - "HTTP_308_PERMANENT_REDIRECT", - "HTTP_400_BAD_REQUEST", - "HTTP_401_UNAUTHORIZED", - "HTTP_402_PAYMENT_REQUIRED", - "HTTP_403_FORBIDDEN", - "HTTP_404_NOT_FOUND", - "HTTP_405_METHOD_NOT_ALLOWED", - "HTTP_406_NOT_ACCEPTABLE", - "HTTP_407_PROXY_AUTHENTICATION_REQUIRED", - "HTTP_408_REQUEST_TIMEOUT", - "HTTP_409_CONFLICT", - "HTTP_410_GONE", - "HTTP_411_LENGTH_REQUIRED", - "HTTP_412_PRECONDITION_FAILED", - "HTTP_413_REQUEST_ENTITY_TOO_LARGE", - "HTTP_414_REQUEST_URI_TOO_LONG", - "HTTP_415_UNSUPPORTED_MEDIA_TYPE", - "HTTP_416_REQUESTED_RANGE_NOT_SATISFIABLE", - "HTTP_417_EXPECTATION_FAILED", - "HTTP_418_IM_A_TEAPOT", - "HTTP_421_MISDIRECTED_REQUEST", - "HTTP_422_UNPROCESSABLE_ENTITY", - "HTTP_423_LOCKED", - "HTTP_424_FAILED_DEPENDENCY", - "HTTP_425_TOO_EARLY", - "HTTP_426_UPGRADE_REQUIRED", - "HTTP_428_PRECONDITION_REQUIRED", - "HTTP_429_TOO_MANY_REQUESTS", - "HTTP_431_REQUEST_HEADER_FIELDS_TOO_LARGE", - "HTTP_451_UNAVAILABLE_FOR_LEGAL_REASONS", - "HTTP_500_INTERNAL_SERVER_ERROR", - "HTTP_501_NOT_IMPLEMENTED", - "HTTP_502_BAD_GATEWAY", - "HTTP_503_SERVICE_UNAVAILABLE", - "HTTP_504_GATEWAY_TIMEOUT", - "HTTP_505_HTTP_VERSION_NOT_SUPPORTED", - "HTTP_506_VARIANT_ALSO_NEGOTIATES", - "HTTP_507_INSUFFICIENT_STORAGE", - "HTTP_508_LOOP_DETECTED", - "HTTP_510_NOT_EXTENDED", - "HTTP_511_NETWORK_AUTHENTICATION_REQUIRED", - "WS_1000_NORMAL_CLOSURE", - "WS_1001_GOING_AWAY", - "WS_1002_PROTOCOL_ERROR", - "WS_1003_UNSUPPORTED_DATA", - "WS_1005_NO_STATUS_RCVD", - "WS_1006_ABNORMAL_CLOSURE", - "WS_1007_INVALID_FRAME_PAYLOAD_DATA", - "WS_1008_POLICY_VIOLATION", - "WS_1009_MESSAGE_TOO_BIG", - "WS_1010_MANDATORY_EXT", - "WS_1011_INTERNAL_ERROR", - "WS_1012_SERVICE_RESTART", - "WS_1013_TRY_AGAIN_LATER", - "WS_1014_BAD_GATEWAY", - "WS_1015_TLS_HANDSHAKE", -) HTTP_100_CONTINUE = 100 HTTP_101_SWITCHING_PROTOCOLS = 101 @@ -175,26 +91,3 @@ WS_1013_TRY_AGAIN_LATER = 1013 WS_1014_BAD_GATEWAY = 1014 WS_1015_TLS_HANDSHAKE = 1015 - - -__deprecated__ = {"WS_1004_NO_STATUS_RCVD": 1004, "WS_1005_ABNORMAL_CLOSURE": 1005} - - -def __getattr__(name: str) -> int: - deprecation_changes = { - "WS_1004_NO_STATUS_RCVD": "WS_1005_NO_STATUS_RCVD", - "WS_1005_ABNORMAL_CLOSURE": "WS_1006_ABNORMAL_CLOSURE", - } - deprecated = __deprecated__.get(name) - if deprecated: - warnings.warn( - f"'{name}' is deprecated. Use '{deprecation_changes[name]}' instead.", - category=DeprecationWarning, - stacklevel=3, - ) - return deprecated - raise AttributeError(f"module '{__name__}' has no attribute '{name}'") - - -def __dir__() -> list[str]: - return sorted(list(__all__) + list(__deprecated__.keys())) # pragma: no cover diff --git a/starlette/templating.py b/starlette/templating.py index 8e1737f61..2a9a0d9ad 100644 --- a/starlette/templating.py +++ b/starlette/templating.py @@ -1,7 +1,6 @@ from __future__ import annotations import typing -import warnings from os import PathLike from starlette.background import BackgroundTask @@ -22,7 +21,11 @@ else: # pragma: nocover pass_context = jinja2.contextfunction # type: ignore[attr-defined] except ModuleNotFoundError: # pragma: nocover - jinja2 = None # type: ignore[assignment] + raise RuntimeError( + "The starlette.templating module requires the `jinja2` package to be installed." + "\nYou can install this with:\n" + " $ pip install jinja2\n" + ) class _TemplateResponse(HTMLResponse): @@ -63,27 +66,6 @@ class Jinja2Templates: return templates.TemplateResponse("index.html", {"request": request}) """ - @typing.overload - def __init__( - self, - directory: str | PathLike[str] | typing.Sequence[str | PathLike[str]], - *, - context_processors: list[typing.Callable[[Request], dict[str, typing.Any]]] - | None = None, - **env_options: typing.Any, - ) -> None: - ... - - @typing.overload - def __init__( - self, - *, - env: jinja2.Environment, - context_processors: list[typing.Callable[[Request], dict[str, typing.Any]]] - | None = None, - ) -> None: - ... - def __init__( self, directory: str @@ -94,36 +76,18 @@ def __init__( context_processors: list[typing.Callable[[Request], dict[str, typing.Any]]] | None = None, env: jinja2.Environment | None = None, - **env_options: typing.Any, ) -> None: - if env_options: - warnings.warn( - "Extra environment options are deprecated. Use a preconfigured jinja2.Environment instead.", # noqa: E501 - DeprecationWarning, - ) - assert jinja2 is not None, "jinja2 must be installed to use Jinja2Templates" assert bool(directory) ^ bool( env ), "either 'directory' or 'env' arguments must be passed" self.context_processors = context_processors or [] if directory is not None: - self.env = self._create_env(directory, **env_options) + self.env = jinja2.Environment(loader=jinja2.FileSystemLoader(directory)) elif env is not None: self.env = env self._setup_env_defaults(self.env) - def _create_env( - self, - directory: str | PathLike[str] | typing.Sequence[str | PathLike[str]], - **env_options: typing.Any, - ) -> jinja2.Environment: - loader = jinja2.FileSystemLoader(directory) - env_options.setdefault("loader", loader) - env_options.setdefault("autoescape", True) - - return jinja2.Environment(**env_options) - def _setup_env_defaults(self, env: jinja2.Environment) -> None: @pass_context def url_for( @@ -140,86 +104,17 @@ def url_for( def get_template(self, name: str) -> jinja2.Template: return self.env.get_template(name) - @typing.overload def TemplateResponse( self, request: Request, name: str, - context: dict[str, typing.Any] | None = None, + context: dict[typing.Any, typing.Any] | None = None, status_code: int = 200, headers: typing.Mapping[str, str] | None = None, media_type: str | None = None, background: BackgroundTask | None = None, ) -> _TemplateResponse: - ... - - @typing.overload - def TemplateResponse( - self, - name: str, - context: dict[str, typing.Any] | None = None, - status_code: int = 200, - headers: typing.Mapping[str, str] | None = None, - media_type: str | None = None, - background: BackgroundTask | None = None, - ) -> _TemplateResponse: - # Deprecated usage - ... - - def TemplateResponse( - self, *args: typing.Any, **kwargs: typing.Any - ) -> _TemplateResponse: - if args: - if isinstance( - args[0], str - ): # the first argument is template name (old style) - warnings.warn( - "The `name` is not the first parameter anymore. " - "The first parameter should be the `Request` instance.\n" - 'Replace `TemplateResponse(name, {"request": request})` by `TemplateResponse(request, name)`.', # noqa: E501 - DeprecationWarning, - ) - - name = args[0] - context = args[1] if len(args) > 1 else kwargs.get("context", {}) - status_code = ( - args[2] if len(args) > 2 else kwargs.get("status_code", 200) - ) - headers = args[2] if len(args) > 2 else kwargs.get("headers") - media_type = args[3] if len(args) > 3 else kwargs.get("media_type") - background = args[4] if len(args) > 4 else kwargs.get("background") - - if "request" not in context: - raise ValueError('context must include a "request" key') - request = context["request"] - else: # the first argument is a request instance (new style) - request = args[0] - name = args[1] if len(args) > 1 else kwargs["name"] - context = args[2] if len(args) > 2 else kwargs.get("context", {}) - status_code = ( - args[3] if len(args) > 3 else kwargs.get("status_code", 200) - ) - headers = args[4] if len(args) > 4 else kwargs.get("headers") - media_type = args[5] if len(args) > 5 else kwargs.get("media_type") - background = args[6] if len(args) > 6 else kwargs.get("background") - else: # all arguments are kwargs - if "request" not in kwargs: - warnings.warn( - "The `TemplateResponse` now requires the `request` argument.\n" - 'Replace `TemplateResponse(name, {"context": context})` by `TemplateResponse(request, name)`.', # noqa: E501 - DeprecationWarning, - ) - if "request" not in kwargs.get("context", {}): - raise ValueError('context must include a "request" key') - - context = kwargs.get("context", {}) - request = kwargs.get("request", context.get("request")) - name = typing.cast(str, kwargs["name"]) - status_code = kwargs.get("status_code", 200) - headers = kwargs.get("headers") - media_type = kwargs.get("media_type") - background = kwargs.get("background") - + context = context or {} context.setdefault("request", request) for context_processor in self.context_processors: context.update(context_processor(request)) diff --git a/starlette/testclient.py b/starlette/testclient.py index bf928d23f..5f1dafaad 100644 --- a/starlette/testclient.py +++ b/starlette/testclient.py @@ -8,7 +8,6 @@ import queue import sys import typing -import warnings from concurrent.futures import Future from functools import cached_property from types import GeneratorType @@ -473,27 +472,6 @@ def _portal_factory(self) -> typing.Generator[anyio.abc.BlockingPortal, None, No ) as portal: yield portal - def _choose_redirect_arg( - self, follow_redirects: bool | None, allow_redirects: bool | None - ) -> bool | httpx._client.UseClientDefault: - redirect: bool | httpx._client.UseClientDefault = ( - httpx._client.USE_CLIENT_DEFAULT - ) - if allow_redirects is not None: - message = ( - "The `allow_redirects` argument is deprecated. " - "Use `follow_redirects` instead." - ) - warnings.warn(message, DeprecationWarning) - redirect = allow_redirects - if follow_redirects is not None: - redirect = follow_redirects - elif allow_redirects is not None and follow_redirects is not None: - raise RuntimeError( # pragma: no cover - "Cannot use both `allow_redirects` and `follow_redirects`." - ) - return redirect - def request( # type: ignore[override] self, method: str, @@ -508,14 +486,13 @@ def request( # type: ignore[override] cookies: httpx._types.CookieTypes | None = None, auth: httpx._types.AuthTypes | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT, - follow_redirects: bool | None = None, - allow_redirects: bool | None = None, + follow_redirects: bool + | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT, timeout: httpx._types.TimeoutTypes | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT, extensions: dict[str, typing.Any] | None = None, ) -> httpx.Response: url = self._merge_url(url) - redirect = self._choose_redirect_arg(follow_redirects, allow_redirects) return super().request( method, url, @@ -527,7 +504,7 @@ def request( # type: ignore[override] headers=headers, cookies=cookies, auth=auth, - follow_redirects=redirect, + follow_redirects=follow_redirects, timeout=timeout, extensions=extensions, ) @@ -541,20 +518,19 @@ def get( # type: ignore[override] cookies: httpx._types.CookieTypes | None = None, auth: httpx._types.AuthTypes | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT, - follow_redirects: bool | None = None, - allow_redirects: bool | None = None, + follow_redirects: bool + | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT, timeout: httpx._types.TimeoutTypes | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT, extensions: dict[str, typing.Any] | None = None, ) -> httpx.Response: - redirect = self._choose_redirect_arg(follow_redirects, allow_redirects) return super().get( url, params=params, headers=headers, cookies=cookies, auth=auth, - follow_redirects=redirect, + follow_redirects=follow_redirects, timeout=timeout, extensions=extensions, ) @@ -568,20 +544,19 @@ def options( # type: ignore[override] cookies: httpx._types.CookieTypes | None = None, auth: httpx._types.AuthTypes | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT, - follow_redirects: bool | None = None, - allow_redirects: bool | None = None, + follow_redirects: bool + | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT, timeout: httpx._types.TimeoutTypes | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT, extensions: dict[str, typing.Any] | None = None, ) -> httpx.Response: - redirect = self._choose_redirect_arg(follow_redirects, allow_redirects) return super().options( url, params=params, headers=headers, cookies=cookies, auth=auth, - follow_redirects=redirect, + follow_redirects=follow_redirects, timeout=timeout, extensions=extensions, ) @@ -595,20 +570,19 @@ def head( # type: ignore[override] cookies: httpx._types.CookieTypes | None = None, auth: httpx._types.AuthTypes | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT, - follow_redirects: bool | None = None, - allow_redirects: bool | None = None, + follow_redirects: bool + | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT, timeout: httpx._types.TimeoutTypes | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT, extensions: dict[str, typing.Any] | None = None, ) -> httpx.Response: - redirect = self._choose_redirect_arg(follow_redirects, allow_redirects) return super().head( url, params=params, headers=headers, cookies=cookies, auth=auth, - follow_redirects=redirect, + follow_redirects=follow_redirects, timeout=timeout, extensions=extensions, ) @@ -626,13 +600,12 @@ def post( # type: ignore[override] cookies: httpx._types.CookieTypes | None = None, auth: httpx._types.AuthTypes | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT, - follow_redirects: bool | None = None, - allow_redirects: bool | None = None, + follow_redirects: bool + | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT, timeout: httpx._types.TimeoutTypes | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT, extensions: dict[str, typing.Any] | None = None, ) -> httpx.Response: - redirect = self._choose_redirect_arg(follow_redirects, allow_redirects) return super().post( url, content=content, @@ -643,7 +616,7 @@ def post( # type: ignore[override] headers=headers, cookies=cookies, auth=auth, - follow_redirects=redirect, + follow_redirects=follow_redirects, timeout=timeout, extensions=extensions, ) @@ -661,13 +634,12 @@ def put( # type: ignore[override] cookies: httpx._types.CookieTypes | None = None, auth: httpx._types.AuthTypes | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT, - follow_redirects: bool | None = None, - allow_redirects: bool | None = None, + follow_redirects: bool + | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT, timeout: httpx._types.TimeoutTypes | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT, extensions: dict[str, typing.Any] | None = None, ) -> httpx.Response: - redirect = self._choose_redirect_arg(follow_redirects, allow_redirects) return super().put( url, content=content, @@ -678,7 +650,7 @@ def put( # type: ignore[override] headers=headers, cookies=cookies, auth=auth, - follow_redirects=redirect, + follow_redirects=follow_redirects, timeout=timeout, extensions=extensions, ) @@ -696,13 +668,12 @@ def patch( # type: ignore[override] cookies: httpx._types.CookieTypes | None = None, auth: httpx._types.AuthTypes | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT, - follow_redirects: bool | None = None, - allow_redirects: bool | None = None, + follow_redirects: bool + | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT, timeout: httpx._types.TimeoutTypes | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT, extensions: dict[str, typing.Any] | None = None, ) -> httpx.Response: - redirect = self._choose_redirect_arg(follow_redirects, allow_redirects) return super().patch( url, content=content, @@ -713,7 +684,7 @@ def patch( # type: ignore[override] headers=headers, cookies=cookies, auth=auth, - follow_redirects=redirect, + follow_redirects=follow_redirects, timeout=timeout, extensions=extensions, ) @@ -727,20 +698,19 @@ def delete( # type: ignore[override] cookies: httpx._types.CookieTypes | None = None, auth: httpx._types.AuthTypes | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT, - follow_redirects: bool | None = None, - allow_redirects: bool | None = None, + follow_redirects: bool + | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT, timeout: httpx._types.TimeoutTypes | httpx._client.UseClientDefault = httpx._client.USE_CLIENT_DEFAULT, extensions: dict[str, typing.Any] | None = None, ) -> httpx.Response: - redirect = self._choose_redirect_arg(follow_redirects, allow_redirects) return super().delete( url, params=params, headers=headers, cookies=cookies, auth=auth, - follow_redirects=redirect, + follow_redirects=follow_redirects, timeout=timeout, extensions=extensions, ) diff --git a/tests/middleware/test_https_redirect.py b/tests/middleware/test_https_redirect.py index 9195694a3..17d549e0c 100644 --- a/tests/middleware/test_https_redirect.py +++ b/tests/middleware/test_https_redirect.py @@ -25,21 +25,21 @@ def homepage(request: Request) -> PlainTextResponse: assert response.status_code == 200 client = test_client_factory(app) - response = client.get("/", allow_redirects=False) + response = client.get("/", follow_redirects=False) assert response.status_code == 307 assert response.headers["location"] == "https://testserver/" client = test_client_factory(app, base_url="http://testserver:80") - response = client.get("/", allow_redirects=False) + response = client.get("/", follow_redirects=False) assert response.status_code == 307 assert response.headers["location"] == "https://testserver/" client = test_client_factory(app, base_url="http://testserver:443") - response = client.get("/", allow_redirects=False) + response = client.get("/", follow_redirects=False) assert response.status_code == 307 assert response.headers["location"] == "https://testserver/" client = test_client_factory(app, base_url="http://testserver:123") - response = client.get("/", allow_redirects=False) + response = client.get("/", follow_redirects=False) assert response.status_code == 307 assert response.headers["location"] == "https://testserver:123/" diff --git a/tests/middleware/test_wsgi.py b/tests/middleware/test_wsgi.py deleted file mode 100644 index 69842d3ad..000000000 --- a/tests/middleware/test_wsgi.py +++ /dev/null @@ -1,165 +0,0 @@ -import sys -from typing import Any, Callable, Dict, Iterable - -import pytest - -from starlette._utils import collapse_excgroups -from starlette.middleware.wsgi import WSGIMiddleware, build_environ -from starlette.testclient import TestClient - -WSGIResponse = Iterable[bytes] -TestClientFactory = Callable[..., TestClient] -StartResponse = Callable[..., Any] -Environment = Dict[str, Any] - - -def hello_world( - environ: Environment, - start_response: StartResponse, -) -> WSGIResponse: - status = "200 OK" - output = b"Hello World!\n" - headers = [ - ("Content-Type", "text/plain; charset=utf-8"), - ("Content-Length", str(len(output))), - ] - start_response(status, headers) - return [output] - - -def echo_body( - environ: Environment, - start_response: StartResponse, -) -> WSGIResponse: - status = "200 OK" - output = environ["wsgi.input"].read() - headers = [ - ("Content-Type", "text/plain; charset=utf-8"), - ("Content-Length", str(len(output))), - ] - start_response(status, headers) - return [output] - - -def raise_exception( - environ: Environment, - start_response: StartResponse, -) -> WSGIResponse: - raise RuntimeError("Something went wrong") - - -def return_exc_info( - environ: Environment, - start_response: StartResponse, -) -> WSGIResponse: - try: - raise RuntimeError("Something went wrong") - except RuntimeError: - status = "500 Internal Server Error" - output = b"Internal Server Error" - headers = [ - ("Content-Type", "text/plain; charset=utf-8"), - ("Content-Length", str(len(output))), - ] - start_response(status, headers, exc_info=sys.exc_info()) - return [output] - - -def test_wsgi_get(test_client_factory: TestClientFactory) -> None: - app = WSGIMiddleware(hello_world) - client = test_client_factory(app) - response = client.get("/") - assert response.status_code == 200 - assert response.text == "Hello World!\n" - - -def test_wsgi_post(test_client_factory: TestClientFactory) -> None: - app = WSGIMiddleware(echo_body) - client = test_client_factory(app) - response = client.post("/", json={"example": 123}) - assert response.status_code == 200 - assert response.text == '{"example": 123}' - - -def test_wsgi_exception(test_client_factory: TestClientFactory) -> None: - # Note that we're testing the WSGI app directly here. - # The HTTP protocol implementations would catch this error and return 500. - app = WSGIMiddleware(raise_exception) - client = test_client_factory(app) - with pytest.raises(RuntimeError), collapse_excgroups(): - client.get("/") - - -def test_wsgi_exc_info(test_client_factory: TestClientFactory) -> None: - # Note that we're testing the WSGI app directly here. - # The HTTP protocol implementations would catch this error and return 500. - app = WSGIMiddleware(return_exc_info) - client = test_client_factory(app) - with pytest.raises(RuntimeError): - response = client.get("/") - - app = WSGIMiddleware(return_exc_info) - client = test_client_factory(app, raise_server_exceptions=False) - response = client.get("/") - assert response.status_code == 500 - assert response.text == "Internal Server Error" - - -def test_build_environ() -> None: - scope = { - "type": "http", - "http_version": "1.1", - "method": "GET", - "scheme": "https", - "path": "/sub/", - "root_path": "/sub", - "query_string": b"a=123&b=456", - "headers": [ - (b"host", b"www.example.org"), - (b"content-type", b"application/json"), - (b"content-length", b"18"), - (b"accept", b"application/json"), - (b"accept", b"text/plain"), - ], - "client": ("134.56.78.4", 1453), - "server": ("www.example.org", 443), - } - body = b'{"example":"body"}' - environ = build_environ(scope, body) - stream = environ.pop("wsgi.input") - assert stream.read() == b'{"example":"body"}' - assert environ == { - "CONTENT_LENGTH": "18", - "CONTENT_TYPE": "application/json", - "HTTP_ACCEPT": "application/json,text/plain", - "HTTP_HOST": "www.example.org", - "PATH_INFO": "/", - "QUERY_STRING": "a=123&b=456", - "REMOTE_ADDR": "134.56.78.4", - "REQUEST_METHOD": "GET", - "SCRIPT_NAME": "/sub", - "SERVER_NAME": "www.example.org", - "SERVER_PORT": 443, - "SERVER_PROTOCOL": "HTTP/1.1", - "wsgi.errors": sys.stdout, - "wsgi.multiprocess": True, - "wsgi.multithread": True, - "wsgi.run_once": False, - "wsgi.url_scheme": "https", - "wsgi.version": (1, 0), - } - - -def test_build_environ_encoding() -> None: - scope = { - "type": "http", - "http_version": "1.1", - "method": "GET", - "path": "/小星", - "root_path": "/中国", - "query_string": b"a=123&b=456", - "headers": [], - } - environ = build_environ(scope, b"") - assert environ["SCRIPT_NAME"] == "/中国".encode().decode("latin-1") - assert environ["PATH_INFO"] == "/小星".encode().decode("latin-1") diff --git a/tests/test_applications.py b/tests/test_applications.py index 5b6c9d545..879f6e69c 100644 --- a/tests/test_applications.py +++ b/tests/test_applications.py @@ -11,7 +11,6 @@ from starlette.endpoints import HTTPEndpoint from starlette.exceptions import HTTPException, WebSocketException from starlette.middleware import Middleware -from starlette.middleware.base import RequestResponseEndpoint from starlette.middleware.trustedhost import TrustedHostMiddleware from starlette.requests import Request from starlette.responses import JSONResponse, PlainTextResponse @@ -338,35 +337,6 @@ async def websocket_endpoint(session: WebSocket) -> None: assert text == "Hello, world!" -def test_app_add_event_handler(test_client_factory: TestClientFactory) -> None: - startup_complete = False - cleanup_complete = False - - def run_startup() -> None: - nonlocal startup_complete - startup_complete = True - - def run_cleanup() -> None: - nonlocal cleanup_complete - cleanup_complete = True - - with pytest.deprecated_call( - match="The on_startup and on_shutdown parameters are deprecated" - ): - app = Starlette( - on_startup=[run_startup], - on_shutdown=[run_cleanup], - ) - - assert not startup_complete - assert not cleanup_complete - with test_client_factory(app): - assert startup_complete - assert not cleanup_complete - assert startup_complete - assert cleanup_complete - - def test_app_async_cm_lifespan(test_client_factory: TestClientFactory) -> None: startup_complete = False cleanup_complete = False @@ -389,118 +359,6 @@ async def lifespan(app: ASGIApp) -> AsyncGenerator[None, None]: assert cleanup_complete -deprecated_lifespan = pytest.mark.filterwarnings( - r"ignore" - r":(async )?generator function lifespans are deprecated, use an " - r"@contextlib\.asynccontextmanager function instead" - r":DeprecationWarning" - r":starlette.routing" -) - - -@deprecated_lifespan -def test_app_async_gen_lifespan(test_client_factory: TestClientFactory) -> None: - startup_complete = False - cleanup_complete = False - - async def lifespan(app: ASGIApp) -> AsyncGenerator[None, None]: - nonlocal startup_complete, cleanup_complete - startup_complete = True - yield - cleanup_complete = True - - app = Starlette(lifespan=lifespan) # type: ignore - - assert not startup_complete - assert not cleanup_complete - with test_client_factory(app): - assert startup_complete - assert not cleanup_complete - assert startup_complete - assert cleanup_complete - - -@deprecated_lifespan -def test_app_sync_gen_lifespan(test_client_factory: TestClientFactory) -> None: - startup_complete = False - cleanup_complete = False - - def lifespan(app: ASGIApp) -> Generator[None, None, None]: - nonlocal startup_complete, cleanup_complete - startup_complete = True - yield - cleanup_complete = True - - app = Starlette(lifespan=lifespan) # type: ignore - - assert not startup_complete - assert not cleanup_complete - with test_client_factory(app): - assert startup_complete - assert not cleanup_complete - assert startup_complete - assert cleanup_complete - - -def test_decorator_deprecations() -> None: - app = Starlette() - - with pytest.deprecated_call( - match=( - "The `exception_handler` decorator is deprecated, " - "and will be removed in version 1.0.0." - ) - ) as record: - app.exception_handler(500)(http_exception) - assert len(record) == 1 - - with pytest.deprecated_call( - match=( - "The `middleware` decorator is deprecated, " - "and will be removed in version 1.0.0." - ) - ) as record: - - async def middleware( - request: Request, call_next: RequestResponseEndpoint - ) -> None: - ... # pragma: no cover - - app.middleware("http")(middleware) - assert len(record) == 1 - - with pytest.deprecated_call( - match=( - "The `route` decorator is deprecated, " - "and will be removed in version 1.0.0." - ) - ) as record: - app.route("/")(async_homepage) - assert len(record) == 1 - - with pytest.deprecated_call( - match=( - "The `websocket_route` decorator is deprecated, " - "and will be removed in version 1.0.0." - ) - ) as record: - app.websocket_route("/ws")(websocket_endpoint) - assert len(record) == 1 - - with pytest.deprecated_call( - match=( - "The `on_event` decorator is deprecated, " - "and will be removed in version 1.0.0." - ) - ) as record: - - async def startup() -> None: - ... # pragma: no cover - - app.on_event("startup")(startup) - assert len(record) == 1 - - def test_middleware_stack_init(test_client_factory: TestClientFactory) -> None: class NoOpMiddleware: def __init__(self, app: ASGIApp): diff --git a/tests/test_concurrency.py b/tests/test_concurrency.py index aba3ceb1a..7f42cc493 100644 --- a/tests/test_concurrency.py +++ b/tests/test_concurrency.py @@ -1,11 +1,10 @@ from contextvars import ContextVar from typing import Callable, Iterator -import anyio import pytest from starlette.applications import Starlette -from starlette.concurrency import iterate_in_threadpool, run_until_first_complete +from starlette.concurrency import iterate_in_threadpool from starlette.requests import Request from starlette.responses import Response from starlette.routing import Route @@ -14,24 +13,6 @@ TestClientFactory = Callable[..., TestClient] -@pytest.mark.anyio -async def test_run_until_first_complete() -> None: - task1_finished = anyio.Event() - task2_finished = anyio.Event() - - async def task1() -> None: - task1_finished.set() - - async def task2() -> None: - await task1_finished.wait() - await anyio.sleep(0) # pragma: nocover - task2_finished.set() # pragma: nocover - - await run_until_first_complete((task1, {}), (task2, {})) - assert task1_finished.is_set() - assert not task2_finished.is_set() - - def test_accessing_context_from_threaded_sync_endpoint( test_client_factory: TestClientFactory, ) -> None: diff --git a/tests/test_exceptions.py b/tests/test_exceptions.py index 401ad8212..7cc91ca7e 100644 --- a/tests/test_exceptions.py +++ b/tests/test_exceptions.py @@ -1,4 +1,3 @@ -import warnings from typing import Callable, Generator import pytest @@ -193,19 +192,6 @@ class CustomWebSocketException(WebSocketException): ) -def test_exception_middleware_deprecation() -> None: - # this test should be removed once the deprecation shim is removed - with pytest.warns(DeprecationWarning): - from starlette.exceptions import ExceptionMiddleware # noqa: F401 - - with warnings.catch_warnings(): - warnings.simplefilter("error") - import starlette.exceptions - - with pytest.warns(DeprecationWarning): - starlette.exceptions.ExceptionMiddleware - - def test_request_in_app_and_handler_is_the_same_object(client: TestClient) -> None: response = client.post("/consume_body_in_endpoint_and_handler", content=b"Hello!") assert response.status_code == 422 diff --git a/tests/test_requests.py b/tests/test_requests.py index c52ebc141..828644fd4 100644 --- a/tests/test_requests.py +++ b/tests/test_requests.py @@ -464,7 +464,7 @@ def post_body() -> Iterator[bytes]: yield b"foo" yield b"bar" - response = client.post("/", data=post_body()) # type: ignore + response = client.post("/", content=post_body()) assert response.json() == {"body": "foobar"} diff --git a/tests/test_responses.py b/tests/test_responses.py index c134cbda5..c88a125ab 100644 --- a/tests/test_responses.py +++ b/tests/test_responses.py @@ -97,7 +97,7 @@ async def app(scope: Scope, receive: Receive, send: Send) -> None: await response(scope, receive, send) client: TestClient = test_client_factory(app) - response = client.request("GET", "/redirect", allow_redirects=False) + response = client.request("GET", "/redirect", follow_redirects=False) assert response.url == "http://testserver/redirect" assert response.headers["content-length"] == "0" @@ -343,13 +343,6 @@ def test_file_response_with_inline_disposition( assert response.headers["content-disposition"] == expected_disposition -def test_file_response_with_method_warns( - tmpdir: Path, test_client_factory: TestClientFactory -) -> None: - with pytest.warns(DeprecationWarning): - FileResponse(path=tmpdir, filename="example.png", method="GET") - - def test_set_cookie( test_client_factory: TestClientFactory, monkeypatch: pytest.MonkeyPatch ) -> None: diff --git a/tests/test_routing.py b/tests/test_routing.py index b75fc47f0..31036069c 100644 --- a/tests/test_routing.py +++ b/tests/test_routing.py @@ -643,128 +643,9 @@ def test_standalone_ws_route_does_not_match( pass # pragma: nocover -def test_lifespan_async(test_client_factory: TestClientFactory) -> None: - startup_complete = False - shutdown_complete = False - - async def hello_world(request: Request) -> PlainTextResponse: - return PlainTextResponse("hello, world") - - async def run_startup() -> None: - nonlocal startup_complete - startup_complete = True - - async def run_shutdown() -> None: - nonlocal shutdown_complete - shutdown_complete = True - - with pytest.deprecated_call( - match="The on_startup and on_shutdown parameters are deprecated" - ): - app = Router( - on_startup=[run_startup], - on_shutdown=[run_shutdown], - routes=[Route("/", hello_world)], - ) - - assert not startup_complete - assert not shutdown_complete - with test_client_factory(app) as client: - assert startup_complete - assert not shutdown_complete - client.get("/") - assert startup_complete - assert shutdown_complete - - -def test_lifespan_with_on_events(test_client_factory: TestClientFactory) -> None: - lifespan_called = False - startup_called = False - shutdown_called = False - +def test_lifespan_state_unsupported(test_client_factory: TestClientFactory) -> None: @contextlib.asynccontextmanager - async def lifespan(app: Starlette) -> typing.AsyncGenerator[None, None]: - nonlocal lifespan_called - lifespan_called = True - yield - - # We do not expected, neither of run_startup nor run_shutdown to be called - # we thus mark them as #pragma: no cover, to fulfill test coverage - def run_startup() -> None: # pragma: no cover - nonlocal startup_called - startup_called = True - - def run_shutdown() -> None: # pragma: no cover - nonlocal shutdown_called - shutdown_called = True - - with pytest.deprecated_call( - match="The on_startup and on_shutdown parameters are deprecated" - ): - with pytest.warns( - UserWarning, - match=( - "The `lifespan` parameter cannot be used with `on_startup` or `on_shutdown`." # noqa: E501 - ), - ): - app = Router( - on_startup=[run_startup], on_shutdown=[run_shutdown], lifespan=lifespan - ) - - assert not lifespan_called - assert not startup_called - assert not shutdown_called - - # Triggers the lifespan events - with test_client_factory(app): - ... - - assert lifespan_called - assert not startup_called - assert not shutdown_called - - -def test_lifespan_sync(test_client_factory: TestClientFactory) -> None: - startup_complete = False - shutdown_complete = False - - def hello_world(request: Request) -> PlainTextResponse: - return PlainTextResponse("hello, world") - - def run_startup() -> None: - nonlocal startup_complete - startup_complete = True - - def run_shutdown() -> None: - nonlocal shutdown_complete - shutdown_complete = True - - with pytest.deprecated_call( - match="The on_startup and on_shutdown parameters are deprecated" - ): - app = Router( - on_startup=[run_startup], - on_shutdown=[run_shutdown], - routes=[Route("/", hello_world)], - ) - - assert not startup_complete - assert not shutdown_complete - with test_client_factory(app) as client: - assert startup_complete - assert not shutdown_complete - client.get("/") - assert startup_complete - assert shutdown_complete - - -def test_lifespan_state_unsupported( - test_client_factory: TestClientFactory, -) -> None: - @contextlib.asynccontextmanager - async def lifespan( - app: ASGIApp, - ) -> typing.AsyncGenerator[dict[str, str], None]: + async def lifespan(app: Starlette) -> typing.AsyncIterator[dict[str, str]]: yield {"foo": "bar"} app = Router( @@ -831,45 +712,6 @@ async def lifespan(app: Starlette) -> typing.AsyncIterator[State]: assert shutdown_complete -def test_raise_on_startup(test_client_factory: TestClientFactory) -> None: - def run_startup() -> None: - raise RuntimeError() - - with pytest.deprecated_call( - match="The on_startup and on_shutdown parameters are deprecated" - ): - router = Router(on_startup=[run_startup]) - startup_failed = False - - async def app(scope: Scope, receive: Receive, send: Send) -> None: - async def _send(message: Message) -> None: - nonlocal startup_failed - if message["type"] == "lifespan.startup.failed": - startup_failed = True - return await send(message) - - await router(scope, receive, _send) - - with pytest.raises(RuntimeError): - with test_client_factory(app): - pass # pragma: nocover - assert startup_failed - - -def test_raise_on_shutdown(test_client_factory: TestClientFactory) -> None: - def run_shutdown() -> None: - raise RuntimeError() - - with pytest.deprecated_call( - match="The on_startup and on_shutdown parameters are deprecated" - ): - app = Router(on_shutdown=[run_shutdown]) - - with pytest.raises(RuntimeError): - with test_client_factory(app): - pass # pragma: nocover - - def test_partial_async_endpoint(test_client_factory: TestClientFactory) -> None: test_client = test_client_factory(app) response = test_client.get("/partial") @@ -908,6 +750,45 @@ def test_duplicated_param_names() -> None: Route("/{id}/{name}/{id}/{name}", user) +def test_raise_on_startup(test_client_factory: TestClientFactory) -> None: + @contextlib.asynccontextmanager + async def lifespan(app: Starlette) -> typing.AsyncIterator[None]: + raise RuntimeError + yield None # pragma: no cover + + router = Router(lifespan=lifespan) + startup_failed = False + + async def app(scope: Scope, receive: Receive, send: Send) -> None: + async def _send(message: Message) -> None: + nonlocal startup_failed + if message["type"] == "lifespan.startup.failed": + startup_failed = True + return await send(message) + + await router(scope, receive, _send) + + with pytest.raises(RuntimeError): + with test_client_factory(app): + pass # pragma: no cover + assert startup_failed + + +def test_raise_on_shutdown( + test_client_factory: typing.Callable[..., TestClient], +) -> None: + @contextlib.asynccontextmanager + async def lifespan(app: Starlette) -> typing.AsyncIterator[None]: + yield None + raise RuntimeError() + + app = Router(lifespan=lifespan) + + with pytest.raises(RuntimeError): + with test_client_factory(app): + pass # pragma: no cover + + class Endpoint: async def my_method(self, request: Request) -> None: ... # pragma: no cover @@ -1243,23 +1124,6 @@ def test_host_named_repr() -> None: assert repr(route).startswith("Host(host='example.com', name='app', app=") -def test_decorator_deprecations() -> None: - router = Router() - - with pytest.deprecated_call(): - router.route("/")(homepage) - - with pytest.deprecated_call(): - router.websocket_route("/ws")(websocket_endpoint) - - with pytest.deprecated_call(): - - async def startup() -> None: - ... # pragma: nocover - - router.on_event("startup")(startup) - - async def echo_paths(request: Request, name: str) -> JSONResponse: return JSONResponse( { diff --git a/tests/test_status.py b/tests/test_status.py deleted file mode 100644 index 04719e87e..000000000 --- a/tests/test_status.py +++ /dev/null @@ -1,25 +0,0 @@ -import importlib - -import pytest - - -@pytest.mark.parametrize( - "constant,msg", - ( - ( - "WS_1004_NO_STATUS_RCVD", - "'WS_1004_NO_STATUS_RCVD' is deprecated. " - "Use 'WS_1005_NO_STATUS_RCVD' instead.", - ), - ( - "WS_1005_ABNORMAL_CLOSURE", - "'WS_1005_ABNORMAL_CLOSURE' is deprecated. " - "Use 'WS_1006_ABNORMAL_CLOSURE' instead.", - ), - ), -) -def test_deprecated_types(constant: str, msg: str) -> None: - with pytest.warns(DeprecationWarning) as record: - getattr(importlib.import_module("starlette.status"), constant) - assert len(record) == 1 - assert msg in str(record.list[0]) diff --git a/tests/test_templates.py b/tests/test_templates.py index 10a1366bc..1edcd843e 100644 --- a/tests/test_templates.py +++ b/tests/test_templates.py @@ -3,13 +3,11 @@ import os import typing from pathlib import Path -from unittest import mock import jinja2 import pytest from starlette.applications import Starlette -from starlette.background import BackgroundTask from starlette.middleware import Middleware from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint from starlette.requests import Request @@ -140,7 +138,7 @@ def test_templates_require_directory_or_environment() -> None: with pytest.raises( AssertionError, match="either 'directory' or 'env' arguments must be passed" ): - Jinja2Templates() # type: ignore[call-overload] + Jinja2Templates() def test_templates_require_directory_or_enviroment_not_both() -> None: @@ -181,153 +179,3 @@ async def homepage(request: Request) -> Response: assert response.text == "Hello, world" assert response.template.name == "index.html" # type: ignore assert set(response.context.keys()) == {"request"} # type: ignore - - -def test_templates_with_environment_options_emit_warning(tmpdir: Path) -> None: - with pytest.warns(DeprecationWarning): - Jinja2Templates(str(tmpdir), autoescape=True) - - -def test_templates_with_kwargs_only( - tmpdir: Path, test_client_factory: TestClientFactory -) -> None: - # MAINTAINERS: remove after 1.0 - path = os.path.join(tmpdir, "index.html") - with open(path, "w") as file: - file.write("value: {{ a }}") - templates = Jinja2Templates(directory=str(tmpdir)) - - spy = mock.MagicMock() - - def page(request: Request) -> Response: - return templates.TemplateResponse( - request=request, - name="index.html", - context={"a": "b"}, - status_code=201, - headers={"x-key": "value"}, - media_type="text/plain", - background=BackgroundTask(func=spy), - ) - - app = Starlette(routes=[Route("/", page)]) - client = test_client_factory(app) - response = client.get("/") - - assert response.text == "value: b" # context was rendered - assert response.status_code == 201 - assert response.headers["x-key"] == "value" - assert response.headers["content-type"] == "text/plain; charset=utf-8" - spy.assert_called() - - -def test_templates_with_kwargs_only_requires_request_in_context(tmpdir: Path) -> None: - # MAINTAINERS: remove after 1.0 - - templates = Jinja2Templates(directory=str(tmpdir)) - with pytest.warns( - DeprecationWarning, - match="requires the `request` argument", - ): - with pytest.raises(ValueError): - templates.TemplateResponse(name="index.html", context={"a": "b"}) - - -def test_templates_with_kwargs_only_warns_when_no_request_keyword( - tmpdir: Path, test_client_factory: TestClientFactory -) -> None: - # MAINTAINERS: remove after 1.0 - - path = os.path.join(tmpdir, "index.html") - with open(path, "w") as file: - file.write("Hello") - - templates = Jinja2Templates(directory=str(tmpdir)) - - def page(request: Request) -> Response: - return templates.TemplateResponse( - name="index.html", context={"request": request} - ) - - app = Starlette(routes=[Route("/", page)]) - client = test_client_factory(app) - - with pytest.warns( - DeprecationWarning, - match="requires the `request` argument", - ): - client.get("/") - - -def test_templates_with_requires_request_in_context(tmpdir: Path) -> None: - # MAINTAINERS: remove after 1.0 - templates = Jinja2Templates(directory=str(tmpdir)) - with pytest.warns(DeprecationWarning): - with pytest.raises(ValueError): - templates.TemplateResponse("index.html", context={}) - - -def test_templates_warns_when_first_argument_isnot_request( - tmpdir: Path, test_client_factory: TestClientFactory -) -> None: - # MAINTAINERS: remove after 1.0 - path = os.path.join(tmpdir, "index.html") - with open(path, "w") as file: - file.write("value: {{ a }}") - templates = Jinja2Templates(directory=str(tmpdir)) - - spy = mock.MagicMock() - - def page(request: Request) -> Response: - return templates.TemplateResponse( - "index.html", - {"a": "b", "request": request}, - status_code=201, - headers={"x-key": "value"}, - media_type="text/plain", - background=BackgroundTask(func=spy), - ) - - app = Starlette(routes=[Route("/", page)]) - client = test_client_factory(app) - with pytest.warns(DeprecationWarning): - response = client.get("/") - - assert response.text == "value: b" # context was rendered - assert response.status_code == 201 - assert response.headers["x-key"] == "value" - assert response.headers["content-type"] == "text/plain; charset=utf-8" - spy.assert_called() - - -def test_templates_when_first_argument_is_request( - tmpdir: Path, test_client_factory: TestClientFactory -) -> None: - # MAINTAINERS: remove after 1.0 - path = os.path.join(tmpdir, "index.html") - with open(path, "w") as file: - file.write("value: {{ a }}") - templates = Jinja2Templates(directory=str(tmpdir)) - - spy = mock.MagicMock() - - def page(request: Request) -> Response: - return templates.TemplateResponse( - request, - "index.html", - {"a": "b"}, - status_code=201, - headers={"x-key": "value"}, - media_type="text/plain", - background=BackgroundTask(func=spy), - ) - - app = Starlette(routes=[Route("/", page)]) - client = test_client_factory(app) - response = client.get("/") - - assert response.text == "value: b" # context was rendered - assert response.status_code == 201 - assert response.headers["x-key"] == "value" - assert response.headers["content-type"] == "text/plain; charset=utf-8" - spy.assert_called() diff --git a/tests/test_testclient.py b/tests/test_testclient.py index 4ed1ced9a..96d6de126 100644 --- a/tests/test_testclient.py +++ b/tests/test_testclient.py @@ -46,10 +46,6 @@ def current_task() -> Task[Any] | trio.lowlevel.Task: raise RuntimeError(f"unsupported asynclib={asynclib_name}") # pragma: no cover -def startup() -> None: - raise RuntimeError() - - def test_use_testclient_in_endpoint(test_client_factory: TestClientFactory) -> None: """ We should be able to use the test client within applications. @@ -170,10 +166,12 @@ async def loop_id(request: Request) -> JSONResponse: def test_error_on_startup(test_client_factory: TestClientFactory) -> None: - with pytest.deprecated_call( - match="The on_startup and on_shutdown parameters are deprecated" - ): - startup_error_app = Starlette(on_startup=[startup]) + @asynccontextmanager + async def lifespan(app: Starlette) -> AsyncGenerator[None, None]: + raise RuntimeError + yield None # pragma: no cover + + startup_error_app = Starlette(lifespan=lifespan) with pytest.raises(RuntimeError): with test_client_factory(startup_error_app):