1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
|
from __future__ import annotations
from pathlib import Path
import pytest
from starlette.applications import Starlette
from starlette.middleware import Middleware
from starlette.middleware.gzip import GZipMiddleware
from starlette.requests import Request
from starlette.responses import ContentStream, FileResponse, PlainTextResponse, StreamingResponse
from starlette.routing import Route
from starlette.types import Message
from tests.types import TestClientFactory
def test_gzip_responses(test_client_factory: TestClientFactory) -> None:
def homepage(request: Request) -> PlainTextResponse:
return PlainTextResponse("x" * 4000, status_code=200)
app = Starlette(
routes=[Route("/", endpoint=homepage)],
middleware=[Middleware(GZipMiddleware)],
)
client = test_client_factory(app)
response = client.get("/", headers={"accept-encoding": "gzip"})
assert response.status_code == 200
assert response.text == "x" * 4000
assert response.headers["Content-Encoding"] == "gzip"
assert response.headers["Vary"] == "Accept-Encoding"
assert int(response.headers["Content-Length"]) < 4000
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(
routes=[Route("/", endpoint=homepage)],
middleware=[Middleware(GZipMiddleware)],
)
client = test_client_factory(app)
response = client.get("/", headers={"accept-encoding": "identity"})
assert response.status_code == 200
assert response.text == "x" * 4000
assert "Content-Encoding" not in response.headers
assert response.headers["Vary"] == "Accept-Encoding"
assert int(response.headers["Content-Length"]) == 4000
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(
routes=[Route("/", endpoint=homepage)],
middleware=[Middleware(GZipMiddleware)],
)
client = test_client_factory(app)
response = client.get("/", headers={"accept-encoding": "gzip"})
assert response.status_code == 200
assert response.text == "OK"
assert "Content-Encoding" not in response.headers
assert "Vary" not in response.headers
assert int(response.headers["Content-Length"]) == 2
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
streaming = generator(bytes=b"x" * 400, count=10)
return StreamingResponse(streaming, status_code=200)
app = Starlette(
routes=[Route("/", endpoint=homepage)],
middleware=[Middleware(GZipMiddleware)],
)
client = test_client_factory(app)
response = client.get("/", headers={"accept-encoding": "gzip"})
assert response.status_code == 200
assert response.text == "x" * 4000
assert response.headers["Content-Encoding"] == "gzip"
assert response.headers["Vary"] == "Accept-Encoding"
assert "Content-Length" not in response.headers
def test_gzip_streaming_response_identity(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
streaming = generator(bytes=b"x" * 400, count=10)
return StreamingResponse(streaming, status_code=200)
app = Starlette(
routes=[Route("/", endpoint=homepage)],
middleware=[Middleware(GZipMiddleware)],
)
client = test_client_factory(app)
response = client.get("/", headers={"accept-encoding": "identity"})
assert response.status_code == 200
assert response.text == "x" * 4000
assert "Content-Encoding" not in response.headers
assert response.headers["Vary"] == "Accept-Encoding"
assert "Content-Length" not in response.headers
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
streaming = generator(bytes=b"x" * 400, count=10)
return StreamingResponse(streaming, status_code=200, headers={"Content-Encoding": "text"})
app = Starlette(
routes=[Route("/", endpoint=homepage)],
middleware=[Middleware(GZipMiddleware)],
)
client = test_client_factory(app)
response = client.get("/", headers={"accept-encoding": "gzip, text"})
assert response.status_code == 200
assert response.text == "x" * 4000
assert response.headers["Content-Encoding"] == "text"
assert "Vary" not in response.headers
assert "Content-Length" not in response.headers
def test_gzip_ignored_on_server_sent_events(test_client_factory: TestClientFactory) -> None:
def homepage(request: Request) -> StreamingResponse:
async def generator(bytes: bytes, count: int) -> ContentStream:
for _ in range(count):
yield bytes
streaming = generator(bytes=b"x" * 400, count=10)
return StreamingResponse(streaming, status_code=200, media_type="text/event-stream")
app = Starlette(
routes=[Route("/", endpoint=homepage)],
middleware=[Middleware(GZipMiddleware)],
)
client = test_client_factory(app)
response = client.get("/", headers={"accept-encoding": "gzip"})
assert response.status_code == 200
assert response.text == "x" * 4000
assert "Content-Encoding" not in response.headers
assert "Content-Length" not in response.headers
@pytest.mark.anyio
async def test_gzip_ignored_for_pathsend_responses(tmpdir: Path) -> None:
path = tmpdir / "example.txt"
with path.open("w") as file:
file.write("<file content>")
events: list[Message] = []
async def endpoint_with_pathsend(request: Request) -> FileResponse:
_ = await request.body()
return FileResponse(path)
app = Starlette(
routes=[Route("/", endpoint=endpoint_with_pathsend)],
middleware=[Middleware(GZipMiddleware)],
)
scope = {
"type": "http",
"version": "3",
"method": "GET",
"path": "/",
"headers": [(b"accept-encoding", b"gzip, text")],
"extensions": {"http.response.pathsend": {}},
}
async def receive() -> Message:
return {"type": "http.request", "body": b"", "more_body": False}
async def send(message: Message) -> None:
events.append(message)
await app(scope, receive, send)
assert len(events) == 2
assert events[0]["type"] == "http.response.start"
assert events[1]["type"] == "http.response.pathsend"
|
