+from typing import Callable
+
from starlette.applications import Starlette
from starlette.middleware import Middleware
from starlette.middleware.gzip import GZipMiddleware
-from starlette.responses import PlainTextResponse, StreamingResponse
+from starlette.requests import Request
+from starlette.responses import ContentStream, PlainTextResponse, StreamingResponse
from starlette.routing import Route
+from starlette.testclient import TestClient
+from starlette.types import ASGIApp
+
+TestClientFactory = Callable[[ASGIApp], TestClient]
-def test_gzip_responses(test_client_factory):
- def homepage(request):
+def test_gzip_responses(test_client_factory: TestClientFactory) -> None:
+ def homepage(request: Request) -> PlainTextResponse:
return PlainTextResponse("x" * 4000, status_code=200)
app = Starlette(
assert int(response.headers["Content-Length"]) < 4000
-def test_gzip_not_in_accept_encoding(test_client_factory):
- def homepage(request):
+def test_gzip_not_in_accept_encoding(test_client_factory: TestClientFactory) -> None:
+ def homepage(request: Request) -> PlainTextResponse:
return PlainTextResponse("x" * 4000, status_code=200)
app = Starlette(
assert int(response.headers["Content-Length"]) == 4000
-def test_gzip_ignored_for_small_responses(test_client_factory):
- def homepage(request):
+def test_gzip_ignored_for_small_responses(
+ test_client_factory: TestClientFactory,
+) -> None:
+ def homepage(request: Request) -> PlainTextResponse:
return PlainTextResponse("OK", status_code=200)
app = Starlette(
assert int(response.headers["Content-Length"]) == 2
-def test_gzip_streaming_response(test_client_factory):
- def homepage(request):
- async def generator(bytes, count):
+def test_gzip_streaming_response(test_client_factory: TestClientFactory) -> None:
+ def homepage(request: Request) -> StreamingResponse:
+ async def generator(bytes: bytes, count: int) -> ContentStream:
for index in range(count):
yield bytes
assert "Content-Length" not in response.headers
-def test_gzip_ignored_for_responses_with_encoding_set(test_client_factory):
- def homepage(request):
- async def generator(bytes, count):
+def test_gzip_ignored_for_responses_with_encoding_set(
+ test_client_factory: TestClientFactory,
+) -> None:
+ def homepage(request: Request) -> StreamingResponse:
+ async def generator(bytes: bytes, count: int) -> ContentStream:
for index in range(count):
yield bytes