Fix double compression issue when enable_compression() is called on pre-encoded responses

CHANGES/10968.bugfix.rst

@@ -0,0 +1 @@
+Fixed double compression issue when :py:meth:`~aiohttp.web.StreamResponse.enable_compression` is called on a response with pre-existing Content-Encoding header -- by :user:`bdraco`.

aiohttp/web_response.py

@@ -186,6 +186,13 @@ def enable_compression(
         strategy: Optional[int] = None,
     ) -> None:
         """Enables response compression encoding."""
+        # Don't enable compression if content is already encoded.
+        # This prevents double compression and provides a safe, predictable behavior
+        # without breaking existing code that may call enable_compression() on
+        # responses that already have Content-Encoding set (e.g., FileResponse
+        # serving pre-compressed files).
+        if hdrs.CONTENT_ENCODING in self._headers:
+            return
         self._compression = True
         self._compression_force = force
         self._compression_strategy = strategy

tests/test_web_response.py

@@ -627,6 +627,31 @@ async def write_headers(status_line: str, headers: CIMultiDict[str]) -> None:
     assert resp.content_length == 6
 
 
+async def test_enable_compression_with_existing_encoding() -> None:
+    """Test that enable_compression does not override existing content encoding."""
+    writer = mock.Mock()
+
+    async def write_headers(status_line: str, headers: CIMultiDict[str]) -> None:
+        # Should preserve the existing content encoding
+        assert headers[hdrs.CONTENT_ENCODING] == "gzip"
+        # Should not have double encoding
+        assert headers.get(hdrs.CONTENT_ENCODING) != "gzip, deflate"
+
+    writer.write_headers.side_effect = write_headers
+    req = make_request("GET", "/", writer=writer)
+    resp = web.Response(body=b"answer")
+
+    # Manually set content encoding (simulating FileResponse with pre-compressed file)
+    resp.headers[hdrs.CONTENT_ENCODING] = "gzip"
+
+    # Try to enable compression - should be ignored
+    resp.enable_compression(web.ContentCoding.deflate)
+
+    await resp.prepare(req)
+    # Verify compression was not enabled due to existing encoding
+    assert not resp.compression
+
+
 @pytest.mark.usefixtures("parametrize_zlib_backend")
 async def test_rm_content_length_if_compression_http11() -> None:
     writer = mock.Mock()