test_gzip_ignored_for_responses_with_encoding_set fails if brotlicffi is installed

[Kludex]

Discussed in https://github.com/encode/starlette/discussions/1956

^{Originally posted by **mgorny** November 18, 2022} If brotlicffi (`brotlicffi-1.0.9.2` here) is installed, the following test failures occur: ```pytb ============================================================== FAILURES =============================================================== _____________________________________ test_gzip_ignored_for_responses_with_encoding_set[asyncio] ______________________________________ self = 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 = 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(, 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 = 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 = 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 = 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(, 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 = 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 ```

[Kludex]

Thanks @mgorny.

Ok. That makes sense... We are just "faking" the content-encoding there, we should really encode it with brotli, or choose a different encoding maybe, so we don't add the dependency?

[mgorny]

Well, it passed for me if I used a totally random text encoding, i.e.:

diff --git a/tests/middleware/test_gzip.py b/tests/middleware/test_gzip.py
index 74b09e4..b6ec1f3 100644
--- a/tests/middleware/test_gzip.py
+++ b/tests/middleware/test_gzip.py
@@ -86,7 +86,7 @@ def test_gzip_ignored_for_responses_with_encoding_set(test_client_factory):

         streaming = generator(bytes=b"x" * 400, count=10)
         return StreamingResponse(
-            streaming, status_code=200, headers={"Content-Encoding": "br"}
+            streaming, status_code=200, headers={"Content-Encoding": "text"}
         )

     app = Starlette(
@@ -95,8 +95,8 @@ def test_gzip_ignored_for_responses_with_encoding_set(test_client_factory):
     )

     client = test_client_factory(app)
-    response = client.get("/", headers={"accept-encoding": "gzip, br"})
+    response = client.get("/", headers={"accept-encoding": "gzip, text"})
     assert response.status_code == 200
     assert response.text == "x" * 4000
-    assert response.headers["Content-Encoding"] == "br"
+    assert response.headers["Content-Encoding"] == "text"
     assert "Content-Length" not in response.headers

However, I'm not sure how portable is that.

[Kludex]

Ok. PR welcome.

[Kludex]

Closed by #1962