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
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
|
import os
import pytest
import falcon
from falcon import asgi
from falcon import testing
@pytest.mark.parametrize(
'body',
[
b'',
b'\x00',
b'\x00\xff',
b'catsup',
b'\xde\xad\xbe\xef' * 512,
testing.rand_string(1, 2048),
os.urandom(100 * 2**20),
],
ids=['empty', 'null', 'null-ff', 'normal', 'long', 'random', 'random-large'],
)
@pytest.mark.parametrize('extra_body', [True, False])
@pytest.mark.parametrize('set_content_length', [True, False])
@pytest.mark.slow
def test_read_all(body, extra_body, set_content_length):
if extra_body and not set_content_length:
pytest.skip(
'extra_body ignores set_content_length so we only need to test '
'one of the parameter permutations'
)
expected_body = body if isinstance(body, bytes) else body.encode()
def stream():
stream_body = body
content_length = None
if extra_body:
# NOTE(kgriffs): Test emitting more data than expected to the app
content_length = len(expected_body)
stream_body += b'\x00' if isinstance(stream_body, bytes) else '~'
elif set_content_length:
content_length = len(expected_body)
return _stream(stream_body, content_length=content_length)
async def test_iteration():
s = stream()
chunks = [chunk async for chunk in s]
if not (expected_body or extra_body):
assert not chunks
assert b''.join(chunks) == expected_body
assert await s.read() == b''
assert await s.readall() == b''
assert not [chunk async for chunk in s]
assert s.tell() == len(expected_body)
assert s.eof
async def test_readall_a():
s = stream()
assert await s.readall() == expected_body
assert await s.read() == b''
assert await s.readall() == b''
assert not [chunk async for chunk in s]
assert s.tell() == len(expected_body)
assert s.eof
async def test_readall_b():
s = stream()
assert await s.read() == expected_body
assert await s.readall() == b''
assert await s.read() == b''
assert not [chunk async for chunk in s]
assert s.tell() == len(expected_body)
assert s.eof
async def test_readall_c():
s = stream()
body = await s.read(1)
body += await s.read(None)
assert body == expected_body
assert s.tell() == len(expected_body)
assert s.eof
async def test_readall_d():
s = stream()
assert not s.closed
if expected_body:
assert not s.eof
elif set_content_length:
assert s.eof
else:
# NOTE(kgriffs): Stream doesn't know if there is more data
# coming or not until the first read.
assert not s.eof
assert s.tell() == 0
assert await s.read(-2) == b''
assert await s.read(-3) == b''
assert await s.read(-100) == b''
assert await s.read(-1) == expected_body
assert await s.read(-1) == b''
assert await s.readall() == b''
assert await s.read() == b''
assert not [chunk async for chunk in s]
assert await s.read(-2) == b''
assert s.tell() == len(expected_body)
assert s.eof
assert not s.closed
s.close()
assert s.closed
for t in (
test_iteration,
test_readall_a,
test_readall_b,
test_readall_c,
test_readall_d,
):
falcon.async_to_sync(t)
def test_filelike():
s = asgi.BoundedStream(testing.ASGIRequestEventEmitter())
for __ in range(2):
with pytest.raises(OSError):
s.fileno()
assert not s.isatty()
assert s.readable()
assert not s.seekable()
assert not s.writable()
s.close()
assert s.closed
# NOTE(kgriffs): Closing an already-closed stream is a noop.
s.close()
assert s.closed
async def test_iteration():
with pytest.raises(ValueError):
await s.read()
with pytest.raises(ValueError):
await s.readall()
with pytest.raises(ValueError):
await s.exhaust()
with pytest.raises(ValueError):
async for chunk in s:
pass
falcon.async_to_sync(test_iteration)
async def test_iterate_streaming_request():
events = iter(
(
{'type': 'http.request', 'body': b'Hello, ', 'more_body': True},
{'type': 'http.request', 'body': b'World', 'more_body': True},
{'type': 'http.request', 'body': b'!\n', 'more_body': True},
{'type': 'http.request', 'body': b'', 'more_body': False},
{'type': 'http.disconnect'},
)
)
async def receive():
event = next(events)
assert (
event['type'] != 'http.disconnect'
), 'would hang until the client times out'
return event
s = asgi.BoundedStream(receive)
assert b''.join([chunk async for chunk in s]) == b'Hello, World!\n'
@pytest.mark.parametrize(
'body',
[
b'',
b'\x00',
b'\x00\xff',
b'catsup',
b'\xde\xad\xbe\xef' * 512,
testing.rand_string(1, 2048).encode(),
],
ids=['empty', 'null', 'null-ff', 'normal', 'long', 'random'],
)
@pytest.mark.parametrize('chunk_size', [1, 2, 10, 64, 100, 1000, 10000])
def test_read_chunks(body, chunk_size):
def stream():
return _stream(body)
async def test_nonmixed():
s = stream()
assert await s.read(0) == b''
chunks = []
while not s.eof:
chunks.append(await s.read(chunk_size))
assert b''.join(chunks) == body
async def test_mixed_a():
s = stream()
chunks = []
chunks.append(await s.read(chunk_size))
chunks.append(await s.read(chunk_size))
chunks.append(await s.readall())
chunks.append(await s.read(chunk_size))
assert b''.join(chunks) == body
async def test_mixed_b():
s = stream()
chunks = []
chunks.append(await s.read(chunk_size))
chunks.append(await s.read(-1))
assert b''.join(chunks) == body
async def test_mixed_iter():
s = stream()
chunks = [await s.read(chunk_size)]
chunks += [data async for data in s]
assert b''.join(chunks) == body
for t in (test_nonmixed, test_mixed_a, test_mixed_b, test_mixed_iter):
falcon.async_to_sync(t)
falcon.async_to_sync(t)
def test_exhaust_with_disconnect():
async def t():
emitter = testing.ASGIRequestEventEmitter(
b'123456789' * 2,
# NOTE(kgriffs): This must be small enough to create several events
chunk_size=3,
)
s = asgi.BoundedStream(emitter)
assert await s.read(1) == b'1'
assert await s.read(2) == b'23'
emitter.disconnect(exhaust_body=False)
await s.exhaust()
assert await s.read(1) == b''
assert await s.read(100) == b''
assert s.eof
falcon.async_to_sync(t)
async def test_exhaust():
emitter = testing.ASGIRequestEventEmitter(b'123456798' * 1024)
stream = asgi.BoundedStream(emitter)
assert await stream.read(1) == b'1'
assert await stream.read(6) == b'234567'
assert await stream.read(101) == b'98' + b'123456798' * 11
await stream.exhaust()
assert await stream.read(1) == b''
assert await stream.read(6) == b''
assert await stream.read(101) == b''
assert stream.eof
def test_iteration_already_started():
body = testing.rand_string(1, 2048).encode()
s = _stream(body)
async def t():
stream_iter = s.__aiter__()
chunks = [await stream_iter.__anext__()]
with pytest.raises(ValueError):
stream_iter2 = s.__aiter__()
await stream_iter2.__anext__()
while True:
try:
chunks.append(await stream_iter.__anext__())
except StopAsyncIteration:
break
assert b''.join(chunks) == body
falcon.async_to_sync(t)
def _stream(body, content_length=None):
emitter = testing.ASGIRequestEventEmitter(body)
return asgi.BoundedStream(emitter, content_length=content_length)
|
