"""
def __init__(self) -> None:
- self.decompressor = zlib.decompressobj(-zlib.MAX_WBITS)
+ self.first_attempt = True
+ self.decompressor = zlib.decompressobj()
def decode(self, data: bytes) -> bytes:
+ was_first_attempt = self.first_attempt
+ self.first_attempt = False
try:
return self.decompressor.decompress(data)
except zlib.error as exc:
+ if was_first_attempt:
+ self.decompressor = zlib.decompressobj(-zlib.MAX_WBITS)
+ return self.decode(data)
raise DecodingError from exc
def flush(self) -> bytes:
def test_deflate():
+ """
+ Deflate encoding may use either 'zlib' or 'deflate' in the wild.
+
+ https://stackoverflow.com/questions/1838699/how-can-i-decompress-a-gzip-stream-with-zlib#answer-22311297
+ """
body = b"test 123"
compressor = zlib.compressobj(9, zlib.DEFLATED, -zlib.MAX_WBITS)
compressed_body = compressor.compress(body) + compressor.flush()
assert response.content == body
+def test_zlib():
+ """
+ Deflate encoding may use either 'zlib' or 'deflate' in the wild.
+
+ https://stackoverflow.com/questions/1838699/how-can-i-decompress-a-gzip-stream-with-zlib#answer-22311297
+ """
+ body = b"test 123"
+ compressed_body = zlib.compress(body)
+
+ headers = [(b"Content-Encoding", b"deflate")]
+ response = httpx.Response(
+ 200, headers=headers, content=compressed_body, request=REQUEST
+ )
+ assert response.content == body
+
+
def test_gzip():
body = b"test 123"
compressor = zlib.compressobj(9, zlib.DEFLATED, zlib.MAX_WBITS | 16)