95 lines
2.9 KiB
Python
95 lines
2.9 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, body=compressed_body)
|
|
assert response.body == 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, body=compressed_body)
|
|
assert response.body == 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, body=compressed_body)
|
|
assert response.body == 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, body=compressed_body)
|
|
assert response.body == 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, body=compressed_body)
|
|
assert response.body == body
|
|
|
|
headers = [(b"Content-Encoding", b"identity, br")]
|
|
response = httpcore.Response(200, headers=headers, body=compressed_body)
|
|
assert response.body == body
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_streaming():
|
|
body = b"test 123"
|
|
compressor = zlib.compressobj(9, zlib.DEFLATED, zlib.MAX_WBITS | 16)
|
|
|
|
async def compress(body):
|
|
yield compressor.compress(body)
|
|
yield compressor.flush()
|
|
|
|
headers = [(b"Content-Encoding", b"gzip")]
|
|
response = httpcore.Response(200, headers=headers, body=compress(body))
|
|
assert not hasattr(response, "body")
|
|
assert await response.read() == body
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
'header_value, expected_exception',
|
|
[
|
|
(b"deflate", httpcore.DeflateDecodingError),
|
|
(b"gzip", httpcore.GzipDecodingError),
|
|
(b"br", httpcore.BrotliDecodingError),
|
|
])
|
|
def test_decoding_errors(header_value, expected_exception):
|
|
headers = [(b"Content-Encoding", header_value)]
|
|
body = b"test 123"
|
|
compressed_body = brotli.compress(body)[3:]
|
|
with pytest.raises(expected_exception):
|
|
response = httpcore.Response(200, headers=headers, body=compressed_body)
|