test_gzip_ignored_for_responses_with_encoding_set fails if brotlicffi is installed

[Kludex]

Discussed in #1956

^{Originally posted by mgorny November 18, 2022}
If brotlicffi (brotlicffi-1.0.9.2 here) is installed, the following test failures occur:

============================================================== 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
```</div>

[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