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
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
|
import itertools
import subprocess
import sys
import textwrap
import pytest
import requests
from pytest_localserver import http
from pytest_localserver import plugin
# define test fixture here again in order to run tests without having to
# install the plugin anew every single time
httpserver = plugin.httpserver
transfer_encoded = pytest.mark.parametrize(
"transfer_encoding_header", ["Transfer-encoding", "Transfer-Encoding", "transfer-encoding", "TRANSFER-ENCODING"]
)
def test_httpserver_funcarg(httpserver):
assert isinstance(httpserver, http.ContentServer)
assert httpserver.is_alive()
assert httpserver.server_address
def test_server_does_not_serve_file_at_startup(httpserver):
assert httpserver.code == 204
assert httpserver.content == ""
def test_some_content_retrieval(httpserver):
httpserver.serve_content("TEST!")
resp = requests.get(httpserver.url)
assert resp.text == "TEST!"
assert resp.status_code == 200
def test_request_is_stored(httpserver):
httpserver.serve_content("TEST!")
assert len(httpserver.requests) == 0
requests.get(httpserver.url)
assert len(httpserver.requests) == 1
def test_GET_request(httpserver):
httpserver.serve_content("TEST!", headers={"Content-type": "text/plain"})
resp = requests.get(httpserver.url, headers={"User-Agent": "Test method"})
assert resp.text == "TEST!"
assert resp.status_code == 200
assert "text/plain" in resp.headers["Content-type"]
# FIXME get compression working!
# def test_gzipped_GET_request(httpserver):
# httpserver.serve_content('TEST!', headers={'Content-type': 'text/plain'})
# httpserver.compress = 'gzip'
# resp = requests.get(httpserver.url, headers={
# 'User-Agent': 'Test method',
# 'Accept-encoding': 'gzip'
# })
# assert resp.text == 'TEST!'
# assert resp.status_code == 200
# assert resp.content_encoding == 'gzip'
# assert resp.headers['Content-type'] == 'text/plain'
# assert resp.headers['content-encoding'] == 'gzip'
def test_HEAD_request(httpserver):
httpserver.serve_content("TEST!", headers={"Content-type": "text/plain"})
resp = requests.head(httpserver.url)
assert resp.status_code == 200
assert resp.headers["Content-type"] == "text/plain"
# def test_POST_request(httpserver):
# headers = {'Content-type': 'application/x-www-form-urlencoded',
# 'set-cookie': 'some _cookie_content'}
#
# httpserver.serve_content('TEST!', headers=headers)
# resp = requests.post(httpserver.url, data={'data': 'value'}, headers=headers)
# assert resp.text == 'TEST!'
# assert resp.status_code == 200
#
# httpserver.serve_content('TEST!', headers=headers, show_post_vars=True)
# resp = requests.post(httpserver.url, data={'data': 'value'}, headers=headers)
# assert resp.json() == {'data': 'value'}
# assert resp.status_code == 200
@pytest.mark.parametrize("chunked_flag", [http.Chunked.YES, http.Chunked.AUTO, http.Chunked.NO])
def test_chunked_attribute_without_header(httpserver, chunked_flag):
"""
Test that passing the chunked attribute to serve_content() properly sets
the chunked property of the server.
"""
httpserver.serve_content(("TEST!", "test"), headers={"Content-type": "text/plain"}, chunked=chunked_flag)
assert httpserver.chunked == chunked_flag
@pytest.mark.parametrize("chunked_flag", [http.Chunked.YES, http.Chunked.AUTO, http.Chunked.NO])
def test_chunked_attribute_with_header(httpserver, chunked_flag):
"""
Test that passing the chunked attribute to serve_content() properly sets
the chunked property of the server even when the transfer-encoding header is
also set.
"""
httpserver.serve_content(
("TEST!", "test"), headers={"Content-type": "text/plain", "Transfer-encoding": "chunked"}, chunked=chunked_flag
)
assert httpserver.chunked == chunked_flag
@transfer_encoded
@pytest.mark.parametrize("chunked_flag", [http.Chunked.YES, http.Chunked.AUTO])
def test_GET_request_chunked_parameter(httpserver, transfer_encoding_header, chunked_flag):
"""
Test that passing YES or AUTO as the chunked parameter to serve_content()
causes the response to be sent using chunking when the Transfer-encoding
header is also set.
"""
httpserver.serve_content(
("TEST!", "test"),
headers={"Content-type": "text/plain", transfer_encoding_header: "chunked"},
chunked=chunked_flag,
)
resp = requests.get(httpserver.url, headers={"User-Agent": "Test method"})
assert resp.text == "TEST!test"
assert resp.status_code == 200
assert "text/plain" in resp.headers["Content-type"]
assert "chunked" in resp.headers["Transfer-encoding"]
@transfer_encoded
@pytest.mark.parametrize("chunked_flag", [http.Chunked.YES, http.Chunked.AUTO])
def test_GET_request_chunked_attribute(httpserver, transfer_encoding_header, chunked_flag):
"""
Test that setting the chunked attribute of httpserver to YES or AUTO
causes the response to be sent using chunking when the Transfer-encoding
header is also set.
"""
httpserver.serve_content(
("TEST!", "test"), headers={"Content-type": "text/plain", transfer_encoding_header: "chunked"}
)
httpserver.chunked = chunked_flag
resp = requests.get(httpserver.url, headers={"User-Agent": "Test method"})
assert resp.text == "TEST!test"
assert resp.status_code == 200
assert "text/plain" in resp.headers["Content-type"]
assert "chunked" in resp.headers["Transfer-encoding"]
@transfer_encoded
def test_GET_request_not_chunked(httpserver, transfer_encoding_header):
"""
Test that setting the chunked attribute of httpserver to NO causes
the response not to be sent using chunking even if the Transfer-encoding
header is set.
"""
httpserver.serve_content(
("TEST!", "test"),
headers={"Content-type": "text/plain", transfer_encoding_header: "chunked"},
chunked=http.Chunked.NO,
)
with pytest.raises(requests.exceptions.ChunkedEncodingError):
requests.get(httpserver.url, headers={"User-Agent": "Test method"})
@pytest.mark.parametrize("chunked_flag", [http.Chunked.NO, http.Chunked.AUTO])
def test_GET_request_chunked_parameter_no_header(httpserver, chunked_flag):
"""
Test that passing NO or AUTO as the chunked parameter to serve_content()
causes the response not to be sent using chunking when the Transfer-encoding
header is not set.
"""
httpserver.serve_content(("TEST!", "test"), headers={"Content-type": "text/plain"}, chunked=chunked_flag)
resp = requests.get(httpserver.url, headers={"User-Agent": "Test method"})
assert resp.text == "TEST!test"
assert resp.status_code == 200
assert "text/plain" in resp.headers["Content-type"]
assert "Transfer-encoding" not in resp.headers
@pytest.mark.parametrize("chunked_flag", [http.Chunked.NO, http.Chunked.AUTO])
def test_GET_request_chunked_attribute_no_header(httpserver, chunked_flag):
"""
Test that setting the chunked attribute of httpserver to NO or AUTO
causes the response not to be sent using chunking when the Transfer-encoding
header is not set.
"""
httpserver.serve_content(("TEST!", "test"), headers={"Content-type": "text/plain"})
httpserver.chunked = chunked_flag
resp = requests.get(httpserver.url, headers={"User-Agent": "Test method"})
assert resp.text == "TEST!test"
assert resp.status_code == 200
assert "text/plain" in resp.headers["Content-type"]
assert "Transfer-encoding" not in resp.headers
def test_GET_request_chunked_no_header(httpserver):
"""
Test that setting the chunked attribute of httpserver to YES causes
the response to be sent using chunking even if the Transfer-encoding
header is not set.
"""
httpserver.serve_content(("TEST!", "test"), headers={"Content-type": "text/plain"}, chunked=http.Chunked.YES)
resp = requests.get(httpserver.url, headers={"User-Agent": "Test method"})
# Without the Transfer-encoding header set, requests does not undo the chunk
# encoding so it comes through as "raw" chunks
assert resp.text == "5\r\nTEST!\r\n4\r\ntest\r\n0\r\n\r\n"
def _format_chunk(chunk):
r = repr(chunk)
if len(r) <= 40:
return r
else:
return r[:13] + "..." + r[-14:] + " (length {})".format(len(chunk))
def _compare_chunks(expected, actual):
__tracebackhide__ = True
if expected != actual:
message = [_format_chunk(expected) + " != " + _format_chunk(actual)]
if type(expected) == type(actual):
for i, (e, a) in enumerate(itertools.zip_longest(expected, actual, fillvalue="<end>")):
if e != a:
message += [
" Chunks differ at index {}:".format(i),
" Expected: " + (repr(expected[i : i + 5]) + "..." if e != "<end>" else "<end>"),
" Found: " + (repr(actual[i : i + 5]) + "..." if a != "<end>" else "<end>"),
]
break
pytest.fail("\n".join(message))
@pytest.mark.parametrize("chunk_size", [400, 499, 500, 512, 750, 1024, 4096, 8192])
def test_GET_request_large_chunks(httpserver, chunk_size):
"""
Test that a response with large chunks comes through correctly
"""
body = b"0123456789abcdef" * 1024 # 16 kb total
# Split body into fixed-size chunks, from https://stackoverflow.com/a/18854817/56541
chunks = [body[0 + i : chunk_size + i] for i in range(0, len(body), chunk_size)]
httpserver.serve_content(
chunks, headers={"Content-type": "text/plain", "Transfer-encoding": "chunked"}, chunked=http.Chunked.YES
)
resp = requests.get(httpserver.url, headers={"User-Agent": "Test method"}, stream=True)
assert resp.status_code == 200
text = b""
for original_chunk, received_chunk in itertools.zip_longest(chunks, resp.iter_content(chunk_size=None)):
_compare_chunks(original_chunk, received_chunk)
text += received_chunk
assert text == body
assert "chunked" in resp.headers["Transfer-encoding"]
@pytest.mark.parametrize("chunked_flag", [http.Chunked.YES, http.Chunked.AUTO])
def test_GET_request_chunked_no_content_length(httpserver, chunked_flag):
"""
Test that a chunked response does not include a Content-length header
"""
httpserver.serve_content(
("TEST!", "test"), headers={"Content-type": "text/plain", "Transfer-encoding": "chunked"}, chunked=chunked_flag
)
resp = requests.get(httpserver.url, headers={"User-Agent": "Test method"})
assert resp.status_code == 200
assert "Transfer-encoding" in resp.headers
assert "Content-length" not in resp.headers
def test_httpserver_init_failure_no_stderr_during_cleanup(tmp_path):
"""
Test that, when the server encounters an error during __init__, its cleanup
does not raise an AttributeError in its __del__ method, which would emit a
warning onto stderr.
"""
script_path = tmp_path.joinpath("script.py")
script_path.write_text(textwrap.dedent("""
from pytest_localserver import http
from unittest.mock import patch
with patch("pytest_localserver.http.make_server", side_effect=RuntimeError("init failure")):
server = http.ContentServer()
"""))
result = subprocess.run([sys.executable, str(script_path)], stderr=subprocess.PIPE)
# We ensure that no warning about an AttributeError is printed on stderr
# due to an error in the server's __del__ method. This AttributeError is
# raised during cleanup and doesn't affect the exit code of the script, so
# we can't use the exit code to tell whether the script did its job.
assert b"AttributeError" not in result.stderr
|
