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
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)
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)
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""
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."
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
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)
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)
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: