80 lines
2.4 KiB
Python
80 lines
2.4 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
|