def test_iter_raw_with_chunksize():
response = httpx.Response(200, content=streaming_body())
- parts = [part for part in response.iter_raw(chunk_size=5)]
+ parts = list(response.iter_raw(chunk_size=5))
assert parts == [b"Hello", b", wor", b"ld!"]
response = httpx.Response(200, content=streaming_body())
- parts = [part for part in response.iter_raw(chunk_size=7)]
+ parts = list(response.iter_raw(chunk_size=7))
assert parts == [b"Hello, ", b"world!"]
response = httpx.Response(200, content=streaming_body())
- parts = [part for part in response.iter_raw(chunk_size=13)]
+ parts = list(response.iter_raw(chunk_size=13))
assert parts == [b"Hello, world!"]
response = httpx.Response(200, content=streaming_body())
- parts = [part for part in response.iter_raw(chunk_size=20)]
+ parts = list(response.iter_raw(chunk_size=20))
assert parts == [b"Hello, world!"]
response = httpx.Response(200, content=streaming_body_with_empty_chunks())
- parts = [part for part in response.iter_raw()]
+ parts = list(response.iter_raw())
assert parts == [b"Hello, ", b"world!"]
)
with pytest.raises(RuntimeError):
- [part for part in response.iter_raw()]
+ list(response.iter_raw())
def test_close_on_async():
def test_iter_bytes_with_chunk_size():
response = httpx.Response(200, content=streaming_body())
- parts = [part for part in response.iter_bytes(chunk_size=5)]
+ parts = list(response.iter_bytes(chunk_size=5))
assert parts == [b"Hello", b", wor", b"ld!"]
response = httpx.Response(200, content=streaming_body())
- parts = [part for part in response.iter_bytes(chunk_size=13)]
+ parts = list(response.iter_bytes(chunk_size=13))
assert parts == [b"Hello, world!"]
response = httpx.Response(200, content=streaming_body())
- parts = [part for part in response.iter_bytes(chunk_size=20)]
+ parts = list(response.iter_bytes(chunk_size=20))
assert parts == [b"Hello, world!"]
def test_iter_bytes_with_empty_response():
response = httpx.Response(200, content=b"")
- parts = [part for part in response.iter_bytes()]
+ parts = list(response.iter_bytes())
assert parts == []
response = httpx.Response(200, content=streaming_body_with_empty_chunks())
- parts = [part for part in response.iter_bytes()]
+ parts = list(response.iter_bytes())
assert parts == [b"Hello, ", b"world!"]
def test_iter_text_with_chunk_size():
response = httpx.Response(200, content=b"Hello, world!")
- parts = [part for part in response.iter_text(chunk_size=5)]
+ parts = list(response.iter_text(chunk_size=5))
assert parts == ["Hello", ", wor", "ld!"]
response = httpx.Response(200, content=b"Hello, world!!")
- parts = [part for part in response.iter_text(chunk_size=7)]
+ parts = list(response.iter_text(chunk_size=7))
assert parts == ["Hello, ", "world!!"]
response = httpx.Response(200, content=b"Hello, world!")
- parts = [part for part in response.iter_text(chunk_size=7)]
+ parts = list(response.iter_text(chunk_size=7))
assert parts == ["Hello, ", "world!"]
response = httpx.Response(200, content=b"Hello, world!")
- parts = [part for part in response.iter_text(chunk_size=13)]
+ parts = list(response.iter_text(chunk_size=13))
assert parts == ["Hello, world!"]
response = httpx.Response(200, content=b"Hello, world!")
- parts = [part for part in response.iter_text(chunk_size=20)]
+ parts = list(response.iter_text(chunk_size=20))
assert parts == ["Hello, world!"]
200,
content=b"Hello,\nworld!",
)
- content = [line for line in response.iter_lines()]
+ content = list(response.iter_lines())
assert content == ["Hello,", "world!"]
assert isinstance(request.stream, httpx.SyncByteStream)
assert isinstance(request.stream, httpx.AsyncByteStream)
- sync_content = b"".join([part for part in request.stream])
+ sync_content = b"".join(list(request.stream))
async_content = b"".join([part async for part in request.stream])
assert request.headers == {"Host": "www.example.com", "Content-Length": "0"}
assert isinstance(request.stream, typing.Iterable)
assert isinstance(request.stream, typing.AsyncIterable)
- sync_content = b"".join([part for part in request.stream])
+ sync_content = b"".join(list(request.stream))
async_content = b"".join([part async for part in request.stream])
assert request.headers == {"Host": "www.example.com", "Content-Length": "13"}
assert isinstance(request.stream, typing.Iterable)
assert isinstance(request.stream, typing.AsyncIterable)
- sync_content = b"".join([part for part in request.stream])
+ sync_content = b"".join(list(request.stream))
async_content = b"".join([part async for part in request.stream])
assert request.headers == {"Host": "www.example.com", "Content-Length": "13"}
assert isinstance(request.stream, typing.Iterable)
assert not isinstance(request.stream, typing.AsyncIterable)
- content = b"".join([part for part in request.stream])
+ content = b"".join(list(request.stream))
assert request.headers == {"Host": "www.example.com", "Content-Length": "13"}
assert content == b"Hello, world!"
assert isinstance(request.stream, typing.Iterable)
assert not isinstance(request.stream, typing.AsyncIterable)
- content = b"".join([part for part in request.stream])
+ content = b"".join(list(request.stream))
assert request.headers == {
"Host": "www.example.com",
assert content == b"Hello, world!"
with pytest.raises(httpx.StreamConsumed):
- [part for part in request.stream]
+ list(request.stream)
# Support 'data' for compat with requests.
with pytest.warns(DeprecationWarning):
assert isinstance(request.stream, typing.Iterable)
assert not isinstance(request.stream, typing.AsyncIterable)
- content = b"".join([part for part in request.stream])
+ content = b"".join(list(request.stream))
assert request.headers == {
"Host": "www.example.com",
assert isinstance(request.stream, typing.Iterable)
assert isinstance(request.stream, typing.AsyncIterable)
- sync_content = b"".join([part for part in request.stream])
+ sync_content = b"".join(list(request.stream))
async_content = b"".join([part async for part in request.stream])
assert request.headers == {
assert isinstance(request.stream, typing.Iterable)
assert isinstance(request.stream, typing.AsyncIterable)
- sync_content = b"".join([part for part in request.stream])
+ sync_content = b"".join(list(request.stream))
async_content = b"".join([part async for part in request.stream])
assert request.headers == {
assert isinstance(request.stream, typing.Iterable)
assert isinstance(request.stream, typing.AsyncIterable)
- sync_content = b"".join([part for part in request.stream])
+ sync_content = b"".join(list(request.stream))
async_content = b"".join([part async for part in request.stream])
assert request.headers == {
assert isinstance(request.stream, typing.Iterable)
assert isinstance(request.stream, typing.AsyncIterable)
- sync_content = b"".join([part for part in request.stream])
+ sync_content = b"".join(list(request.stream))
async_content = b"".join([part async for part in request.stream])
assert request.headers == {
assert isinstance(request.stream, typing.Iterable)
assert isinstance(request.stream, typing.AsyncIterable)
- sync_content = b"".join([part for part in request.stream])
+ sync_content = b"".join(list(request.stream))
async_content = b"".join([part async for part in request.stream])
assert request.headers == {
assert isinstance(request.stream, typing.Iterable)
assert isinstance(request.stream, typing.AsyncIterable)
- sync_content = b"".join([part for part in request.stream])
+ sync_content = b"".join(list(request.stream))
async_content = b"".join([part async for part in request.stream])
assert request.headers == {
assert isinstance(request.stream, typing.Iterable)
assert isinstance(request.stream, typing.AsyncIterable)
- sync_content = b"".join([part for part in request.stream])
+ sync_content = b"".join(list(request.stream))
async_content = b"".join([part async for part in request.stream])
assert request.headers == {
assert isinstance(request.stream, typing.Iterable)
assert isinstance(request.stream, typing.AsyncIterable)
- sync_content = b"".join([part for part in request.stream])
+ sync_content = b"".join(list(request.stream))
async_content = b"".join([part async for part in request.stream])
assert request.headers == {"Host": "www.example.com", "Content-Length": "0"}
assert isinstance(request.stream, typing.Iterable)
assert isinstance(request.stream, typing.AsyncIterable)
- sync_content = b"".join([part for part in request.stream])
+ sync_content = b"".join(list(request.stream))
async_content = b"".join([part async for part in request.stream])
assert request.headers == {
assert isinstance(response.stream, typing.Iterable)
assert isinstance(response.stream, typing.AsyncIterable)
- sync_content = b"".join([part for part in response.stream])
+ sync_content = b"".join(list(response.stream))
async_content = b"".join([part async for part in response.stream])
assert response.headers == {}
assert isinstance(response.stream, typing.Iterable)
assert isinstance(response.stream, typing.AsyncIterable)
- sync_content = b"".join([part for part in response.stream])
+ sync_content = b"".join(list(response.stream))
async_content = b"".join([part async for part in response.stream])
assert response.headers == {"Content-Length": "13"}
assert isinstance(response.stream, typing.Iterable)
assert not isinstance(response.stream, typing.AsyncIterable)
- content = b"".join([part for part in response.stream])
+ content = b"".join(list(response.stream))
assert response.headers == {"Transfer-Encoding": "chunked"}
assert content == b"Hello, world!"
with pytest.raises(httpx.StreamConsumed):
- [part for part in response.stream]
+ list(response.stream)
@pytest.mark.anyio