test_gzip_ignored_for_responses_with_encoding_set fails if brotlicffi is installed #1956
Answered
by
Kludex
mgorny
asked this question in
Potential Issue
-
|
If brotlicffi ( ============================================================== FAILURES ===============================================================
_____________________________________ test_gzip_ignored_for_responses_with_encoding_set[asyncio] ______________________________________
self = <httpx._decoders.BrotliDecoder object at 0x7f3d08304250>
data = b'xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx...xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx'
def decode(self, data: bytes) -> bytes:
if not data:
return b""
self.seen_data = True
try:
> return self._decompress(data)
venv/lib/python3.11/site-packages/httpx/_decoders.py:119:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <brotlicffi._api.Decompressor object at 0x7f3d08307b10>
data = b'xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx...xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx'
def decompress(self, data):
"""
Decompress part of a complete Brotli-compressed string.
:param data: A bytestring containing Brotli-compressed data.
:returns: A bytestring containing the decompressed data.
"""
chunks = []
available_in = ffi.new("size_t *", len(data))
in_buffer = ffi.new("uint8_t[]", data)
next_in = ffi.new("uint8_t **", in_buffer)
while True:
# Allocate a buffer that's hopefully overlarge, but if it's not we
# don't mind: we'll spin around again.
buffer_size = 5 * len(data)
available_out = ffi.new("size_t *", buffer_size)
out_buffer = ffi.new("uint8_t[]", buffer_size)
next_out = ffi.new("uint8_t **", out_buffer)
rc = lib.BrotliDecoderDecompressStream(self._decoder,
available_in,
next_in,
available_out,
next_out,
ffi.NULL)
# First, check for errors.
if rc == lib.BROTLI_DECODER_RESULT_ERROR:
error_code = lib.BrotliDecoderGetErrorCode(self._decoder)
error_message = lib.BrotliDecoderErrorString(error_code)
> raise error(
b"Decompression error: %s" % ffi.string(error_message)
)
E brotlicffi._api.error: b'Decompression error: PADDING_1'
venv/lib/python3.11/site-packages/brotlicffi/_api.py:404: error
The above exception was the direct cause of the following exception:
test_client_factory = functools.partial(<class 'starlette.testclient.TestClient'>, backend='asyncio', backend_options={})
def test_gzip_ignored_for_responses_with_encoding_set(test_client_factory):
def homepage(request):
async def generator(bytes, count):
for index in range(count):
yield bytes
streaming = generator(bytes=b"x" * 400, count=10)
return StreamingResponse(
streaming, status_code=200, headers={"Content-Encoding": "br"}
)
app = Starlette(
routes=[Route("/", endpoint=homepage)],
middleware=[Middleware(GZipMiddleware)],
)
client = test_client_factory(app)
> response = client.get("/", headers={"accept-encoding": "gzip, br"})
tests/middleware/test_gzip.py:98:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
starlette/testclient.py:488: in get
return super().get(
venv/lib/python3.11/site-packages/httpx/_client.py:1039: in get
return self.request(
starlette/testclient.py:454: in request
return super().request(
venv/lib/python3.11/site-packages/httpx/_client.py:815: in request
return self.send(request, auth=auth, follow_redirects=follow_redirects)
venv/lib/python3.11/site-packages/httpx/_client.py:916: in send
raise exc
venv/lib/python3.11/site-packages/httpx/_client.py:910: in send
response.read()
venv/lib/python3.11/site-packages/httpx/_models.py:792: in read
self._content = b"".join(self.iter_bytes())
venv/lib/python3.11/site-packages/httpx/_models.py:811: in iter_bytes
decoded = decoder.decode(raw_bytes)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <httpx._decoders.BrotliDecoder object at 0x7f3d08304250>
data = b'xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx...xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx'
def decode(self, data: bytes) -> bytes:
if not data:
return b""
self.seen_data = True
try:
return self._decompress(data)
except brotli.error as exc:
> raise DecodingError(str(exc)) from exc
E httpx.DecodingError: b'Decompression error: PADDING_1'
venv/lib/python3.11/site-packages/httpx/_decoders.py:121: DecodingError
_______________________________________ test_gzip_ignored_for_responses_with_encoding_set[trio] _______________________________________
self = <httpx._decoders.BrotliDecoder object at 0x7f3d02452cd0>
data = b'xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx...xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx'
def decode(self, data: bytes) -> bytes:
if not data:
return b""
self.seen_data = True
try:
> return self._decompress(data)
venv/lib/python3.11/site-packages/httpx/_decoders.py:119:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <brotlicffi._api.Decompressor object at 0x7f3d02451250>
data = b'xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx...xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx'
def decompress(self, data):
"""
Decompress part of a complete Brotli-compressed string.
:param data: A bytestring containing Brotli-compressed data.
:returns: A bytestring containing the decompressed data.
"""
chunks = []
available_in = ffi.new("size_t *", len(data))
in_buffer = ffi.new("uint8_t[]", data)
next_in = ffi.new("uint8_t **", in_buffer)
while True:
# Allocate a buffer that's hopefully overlarge, but if it's not we
# don't mind: we'll spin around again.
buffer_size = 5 * len(data)
available_out = ffi.new("size_t *", buffer_size)
out_buffer = ffi.new("uint8_t[]", buffer_size)
next_out = ffi.new("uint8_t **", out_buffer)
rc = lib.BrotliDecoderDecompressStream(self._decoder,
available_in,
next_in,
available_out,
next_out,
ffi.NULL)
# First, check for errors.
if rc == lib.BROTLI_DECODER_RESULT_ERROR:
error_code = lib.BrotliDecoderGetErrorCode(self._decoder)
error_message = lib.BrotliDecoderErrorString(error_code)
> raise error(
b"Decompression error: %s" % ffi.string(error_message)
)
E brotlicffi._api.error: b'Decompression error: PADDING_1'
venv/lib/python3.11/site-packages/brotlicffi/_api.py:404: error
The above exception was the direct cause of the following exception:
test_client_factory = functools.partial(<class 'starlette.testclient.TestClient'>, backend='trio', backend_options={})
def test_gzip_ignored_for_responses_with_encoding_set(test_client_factory):
def homepage(request):
async def generator(bytes, count):
for index in range(count):
yield bytes
streaming = generator(bytes=b"x" * 400, count=10)
return StreamingResponse(
streaming, status_code=200, headers={"Content-Encoding": "br"}
)
app = Starlette(
routes=[Route("/", endpoint=homepage)],
middleware=[Middleware(GZipMiddleware)],
)
client = test_client_factory(app)
> response = client.get("/", headers={"accept-encoding": "gzip, br"})
tests/middleware/test_gzip.py:98:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
starlette/testclient.py:488: in get
return super().get(
venv/lib/python3.11/site-packages/httpx/_client.py:1039: in get
return self.request(
starlette/testclient.py:454: in request
return super().request(
venv/lib/python3.11/site-packages/httpx/_client.py:815: in request
return self.send(request, auth=auth, follow_redirects=follow_redirects)
venv/lib/python3.11/site-packages/httpx/_client.py:916: in send
raise exc
venv/lib/python3.11/site-packages/httpx/_client.py:910: in send
response.read()
venv/lib/python3.11/site-packages/httpx/_models.py:792: in read
self._content = b"".join(self.iter_bytes())
venv/lib/python3.11/site-packages/httpx/_models.py:811: in iter_bytes
decoded = decoder.decode(raw_bytes)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <httpx._decoders.BrotliDecoder object at 0x7f3d02452cd0>
data = b'xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx...xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx'
def decode(self, data: bytes) -> bytes:
if not data:
return b""
self.seen_data = True
try:
return self._decompress(data)
except brotli.error as exc:
> raise DecodingError(str(exc)) from exc
E httpx.DecodingError: b'Decompression error: PADDING_1'
venv/lib/python3.11/site-packages/httpx/_decoders.py:121: DecodingError |
Beta Was this translation helpful? Give feedback.
Answered by
Kludex
Dec 3, 2022
Replies: 1 comment
-
|
This was fixed on #1962. |
Beta Was this translation helpful? Give feedback.
0 replies
Answer selected by
Kludex
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
This was fixed on #1962.