tests: branch coverage in _utils.py, middleware/gzip.py, `routing…

….py` and `requests.py` (#2816)

Co-authored-by: Marcelo Trylesinski <[email protected]>

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
 

starlette/middleware/gzip.py

@@ -13,7 +13,7 @@ def __init__(self, app: ASGIApp, minimum_size: int = 500, compresslevel: int = 9
         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 @@ async def send_with_gzip(self, message: Message) -> None:
                 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)

starlette/requests.py

@@ -230,7 +230,7 @@ async def stream(self) -> typing.AsyncGenerator[bytes, None]:
                     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 @@ async def json(self) -> typing.Any:
         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."

starlette/routing.py

@@ -196,7 +196,7 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
             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 @@ def routes(self) -> list[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 @@ def routes(self) -> list[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)

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: