From: Renan Heckert Leal <123672970+lealre@users.noreply.github.com> Date: Sat, 28 Dec 2024 04:55:20 +0000 (+0000) Subject: tests: branch coverage in `_utils.py`, `middleware/gzip.py`, `routing.py` and `reques... X-Git-Tag: 0.44.0~5 X-Git-Url: http://git.ipfire.org/?a=commitdiff_plain;h=d6ace924c4b8732a54c97c198ee48a2b582f84f5;p=thirdparty%2Fstarlette.git tests: branch coverage in `_utils.py`, `middleware/gzip.py`, `routing.py` and `requests.py` (#2816) Co-authored-by: Marcelo Trylesinski --- diff --git a/starlette/_utils.py b/starlette/_utils.py index 74c9f24f..a22a2745 100644 --- a/starlette/_utils.py +++ b/starlette/_utils.py @@ -75,9 +75,9 @@ def collapse_excgroups() -> typing.Generator[None, None, None]: try: yield except BaseException as exc: - if has_exceptiongroups: + if has_exceptiongroups: # pragma: no cover while isinstance(exc, BaseExceptionGroup) and len(exc.exceptions) == 1: - exc = exc.exceptions[0] # pragma: no cover + exc = exc.exceptions[0] raise exc diff --git a/starlette/middleware/gzip.py b/starlette/middleware/gzip.py index 127b91e7..b677063d 100644 --- a/starlette/middleware/gzip.py +++ b/starlette/middleware/gzip.py @@ -13,7 +13,7 @@ class GZipMiddleware: self.compresslevel = compresslevel async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: - if scope["type"] == "http": + if scope["type"] == "http": # pragma: no branch headers = Headers(scope=scope) if "gzip" in headers.get("Accept-Encoding", ""): responder = GZipResponder(self.app, self.minimum_size, compresslevel=self.compresslevel) @@ -88,7 +88,7 @@ class GZipResponder: await self.send(self.initial_message) await self.send(message) - elif message_type == "http.response.body": + elif message_type == "http.response.body": # pragma: no branch # Remaining body in streaming GZip response. body = message.get("body", b"") more_body = message.get("more_body", False) diff --git a/starlette/requests.py b/starlette/requests.py index a29164d0..f85a3b91 100644 --- a/starlette/requests.py +++ b/starlette/requests.py @@ -230,7 +230,7 @@ class Request(HTTPConnection): self._stream_consumed = True if body: yield body - elif message["type"] == "http.disconnect": + elif message["type"] == "http.disconnect": # pragma: no branch self._is_disconnected = True raise ClientDisconnect() yield b"" @@ -250,7 +250,7 @@ class Request(HTTPConnection): return self._json async def _get_form(self, *, max_files: int | float = 1000, max_fields: int | float = 1000) -> FormData: - if self._form is None: + if self._form is None: # pragma: no branch assert ( parse_options_header is not None ), "The `python-multipart` library must be installed to use form parsing." diff --git a/starlette/routing.py b/starlette/routing.py index 3b3c5296..74daeadb 100644 --- a/starlette/routing.py +++ b/starlette/routing.py @@ -196,7 +196,7 @@ class BaseRoute: if scope["type"] == "http": response = PlainTextResponse("Not Found", status_code=404) await response(scope, receive, send) - elif scope["type"] == "websocket": + elif scope["type"] == "websocket": # pragma: no branch websocket_close = WebSocketClose() await websocket_close(scope, receive, send) return @@ -398,7 +398,7 @@ class Mount(BaseRoute): def matches(self, scope: Scope) -> tuple[Match, Scope]: path_params: dict[str, typing.Any] - if scope["type"] in ("http", "websocket"): + if scope["type"] in ("http", "websocket"): # pragma: no branch root_path = scope.get("root_path", "") route_path = get_route_path(scope) match = self.path_regex.match(route_path) @@ -481,7 +481,7 @@ class Host(BaseRoute): return getattr(self.app, "routes", []) def matches(self, scope: Scope) -> tuple[Match, Scope]: - if scope["type"] in ("http", "websocket"): + if scope["type"] in ("http", "websocket"): # pragma:no branch headers = Headers(scope=scope) host = headers.get("host", "").split(":")[0] match = self.host_regex.match(host) diff --git a/tests/test_routing.py b/tests/test_routing.py index 6d7e5d15..933fe7c3 100644 --- a/tests/test_routing.py +++ b/tests/test_routing.py @@ -565,6 +565,16 @@ def test_url_for_with_double_mount() -> None: assert url == "/mount/static/123" +def test_url_for_with_root_path_ending_with_slash(test_client_factory: TestClientFactory) -> None: + def homepage(request: Request) -> JSONResponse: + return JSONResponse({"index": str(request.url_for("homepage"))}) + + app = Starlette(routes=[Route("/", homepage, name="homepage")]) + client = test_client_factory(app, base_url="https://www.example.org/", root_path="/sub_path/") + response = client.get("/sub_path/") + assert response.json() == {"index": "https://www.example.org/sub_path/"} + + def test_standalone_route_matches( test_client_factory: TestClientFactory, ) -> None: