0cbf3c7581
* Support thread-pooled dispatch * Add ConcurrencyBackend.run * Initial work towards support byte-iterators on sync request data * Test case for byte iterator content * byte iterator support for RequestData * Add BaseResponse * Bridge sync/async data in SyncResponse * Add BaseClient * SyncResponse -> Response * Tweaking type annotation * Distinct classes for Request, AsyncRequest * Tweak is_streaming, content in BaseRequest * Stream handling moves to client * Handle mediating to AsyncResponse from a standard sync Dispatcher class * Working on thread-pooled dispatcher * Support threaded dispatch, inc. streaming requests/responses * Increase test coverage * Coverage and tweaks * Include Accept and User-Agent headers by default
89 lines
2.8 KiB
Python
89 lines
2.8 KiB
Python
import zlib
|
|
|
|
import brotli
|
|
import pytest
|
|
|
|
import httpcore
|
|
|
|
|
|
def test_deflate():
|
|
body = b"test 123"
|
|
compressor = zlib.compressobj(9, zlib.DEFLATED, -zlib.MAX_WBITS)
|
|
compressed_body = compressor.compress(body) + compressor.flush()
|
|
|
|
headers = [(b"Content-Encoding", b"deflate")]
|
|
response = httpcore.Response(200, headers=headers, content=compressed_body)
|
|
assert response.content == body
|
|
|
|
|
|
def test_gzip():
|
|
body = b"test 123"
|
|
compressor = zlib.compressobj(9, zlib.DEFLATED, zlib.MAX_WBITS | 16)
|
|
compressed_body = compressor.compress(body) + compressor.flush()
|
|
|
|
headers = [(b"Content-Encoding", b"gzip")]
|
|
response = httpcore.Response(200, headers=headers, content=compressed_body)
|
|
assert response.content == body
|
|
|
|
|
|
def test_brotli():
|
|
body = b"test 123"
|
|
compressed_body = brotli.compress(body)
|
|
|
|
headers = [(b"Content-Encoding", b"br")]
|
|
response = httpcore.Response(200, headers=headers, content=compressed_body)
|
|
assert response.content == body
|
|
|
|
|
|
def test_multi():
|
|
body = b"test 123"
|
|
|
|
deflate_compressor = zlib.compressobj(9, zlib.DEFLATED, -zlib.MAX_WBITS)
|
|
compressed_body = deflate_compressor.compress(body) + deflate_compressor.flush()
|
|
|
|
gzip_compressor = zlib.compressobj(9, zlib.DEFLATED, zlib.MAX_WBITS | 16)
|
|
compressed_body = (
|
|
gzip_compressor.compress(compressed_body) + gzip_compressor.flush()
|
|
)
|
|
|
|
headers = [(b"Content-Encoding", b"deflate, gzip")]
|
|
response = httpcore.Response(200, headers=headers, content=compressed_body)
|
|
assert response.content == body
|
|
|
|
|
|
def test_multi_with_identity():
|
|
body = b"test 123"
|
|
compressed_body = brotli.compress(body)
|
|
|
|
headers = [(b"Content-Encoding", b"br, identity")]
|
|
response = httpcore.Response(200, headers=headers, content=compressed_body)
|
|
assert response.content == body
|
|
|
|
headers = [(b"Content-Encoding", b"identity, br")]
|
|
response = httpcore.Response(200, headers=headers, content=compressed_body)
|
|
assert response.content == body
|
|
|
|
|
|
def test_streaming():
|
|
body = b"test 123"
|
|
compressor = zlib.compressobj(9, zlib.DEFLATED, zlib.MAX_WBITS | 16)
|
|
|
|
def compress(body):
|
|
yield compressor.compress(body)
|
|
yield compressor.flush()
|
|
|
|
headers = [(b"Content-Encoding", b"gzip")]
|
|
response = httpcore.Response(200, headers=headers, content=compress(body))
|
|
assert not hasattr(response, "body")
|
|
assert response.read() == body
|
|
|
|
|
|
@pytest.mark.parametrize("header_value", (b"deflate", b"gzip", b"br"))
|
|
def test_decoding_errors(header_value):
|
|
headers = [(b"Content-Encoding", header_value)]
|
|
body = b"test 123"
|
|
compressed_body = brotli.compress(body)[3:]
|
|
with pytest.raises(httpcore.exceptions.DecodingError):
|
|
response = httpcore.Response(200, headers=headers, content=compressed_body)
|
|
response.content
|