Fix DUMP binary data corruption in pipeline with decode_responses=True (#1884)

redis/asyncio/client.py

@@ -1768,8 +1768,11 @@ async def _execute_transaction(  # noqa: C901
                     errors.append((i, err))
 
         # parse the EXEC.
+        # Read with disable_decoding=True so that NEVER_DECODE commands
+        # (e.g. DUMP) preserve their binary data when decode_responses=True.
+        # Individual elements are selectively decoded below.
         try:
-            response = await self.parse_response(connection, "_")
+            response = await connection.read_response(disable_decoding=True)
         except ExecAbortError as err:
             if errors:
                 raise errors[0][1] from err
@@ -1806,6 +1809,14 @@ async def _execute_transaction(  # noqa: C901
                 # Remove keys entry, it needs only for cache.
                 options.pop("keys", None)
 
+                # Selectively decode responses: NEVER_DECODE commands (e.g. DUMP)
+                # keep their binary data, all others are decoded normally.
+                if isinstance(r, bytes):
+                    if NEVER_DECODE in options:
+                        options.pop(NEVER_DECODE)
+                    elif connection.encoder.decode_responses:
+                        r = connection.encoder.decode(r)
+
                 if command_name in self.response_callbacks:
                     r = self.response_callbacks[command_name](r, **options)
                     if inspect.isawaitable(r):

redis/client.py

@@ -1855,8 +1855,11 @@ def _execute_transaction(
                     errors.append((i, e))
 
         # parse the EXEC.
+        # Read with disable_decoding=True so that NEVER_DECODE commands
+        # (e.g. DUMP) preserve their binary data when decode_responses=True.
+        # Individual elements are selectively decoded below.
         try:
-            response = self.parse_response(connection, "_")
+            response = connection.read_response(disable_decoding=True)
         except ExecAbortError:
             if errors:
                 raise errors[0][1]
@@ -1890,6 +1893,13 @@ def _execute_transaction(
                 # Remove keys entry, it needs only for cache.
                 options.pop("keys", None)
                 command_name = args[0]
+                # Selectively decode responses: NEVER_DECODE commands (e.g. DUMP)
+                # keep their binary data, all others are decoded normally.
+                if isinstance(r, bytes):
+                    if NEVER_DECODE in options:
+                        options.pop(NEVER_DECODE)
+                    elif connection.encoder.decode_responses:
+                        r = connection.encoder.decode(r)
                 if command_name in self.response_callbacks:
                     r = self.response_callbacks[command_name](r, **options)
             data.append(r)

tests/test_asyncio/test_pipeline.py

@@ -486,6 +486,30 @@ async def test_pipeline_json_module_access(self, r):
         assert "JSON.SET" in r.response_callbacks
         assert "JSON.GET" in r.response_callbacks
 
+    async def test_pipeline_dump_with_decode_responses(self, decoded_r):
+        """DUMP returns binary data that should not be decoded in a pipeline.
+
+        Regression test for https://github.com/redis/redis-py/issues/1884
+        """
+        await decoded_r.set("dumpee", "some_value")
+        async with decoded_r.pipeline(transaction=False) as pipe:
+            pipe.dump("dumpee")
+            result = await pipe.execute()
+            assert len(result) == 1
+            assert isinstance(result[0], bytes)
+
+    async def test_pipeline_transaction_dump_with_decode_responses(self, decoded_r):
+        """DUMP returns binary data that should not be decoded in a transactional pipeline.
+
+        Regression test for https://github.com/redis/redis-py/issues/1884
+        """
+        await decoded_r.set("dumpee", "some_value")
+        async with decoded_r.pipeline(transaction=True) as pipe:
+            pipe.dump("dumpee")
+            result = await pipe.execute()
+            assert len(result) == 1
+            assert isinstance(result[0], bytes)
+
 
 @pytest.mark.asyncio
 class TestAsyncPipelineOperationDurationMetricsRecording:

tests/test_pipeline.py

@@ -460,6 +460,31 @@ def test_pipeline_with_msetex(self, r):
             )
 
 
+    def test_pipeline_dump_with_decode_responses(self, decoded_r):
+        """DUMP returns binary data that should not be decoded in a pipeline.
+
+        Regression test for https://github.com/redis/redis-py/issues/1884
+        """
+        decoded_r.set("dumpee", "some_value")
+        with decoded_r.pipeline(transaction=False) as pipe:
+            pipe.dump("dumpee")
+            result = pipe.execute()
+            assert len(result) == 1
+            assert isinstance(result[0], bytes)
+
+    def test_pipeline_transaction_dump_with_decode_responses(self, decoded_r):
+        """DUMP returns binary data that should not be decoded in a transactional pipeline.
+
+        Regression test for https://github.com/redis/redis-py/issues/1884
+        """
+        decoded_r.set("dumpee", "some_value")
+        with decoded_r.pipeline(transaction=True) as pipe:
+            pipe.dump("dumpee")
+            result = pipe.execute()
+            assert len(result) == 1
+            assert isinstance(result[0], bytes)
+
+
 class TestPipelineMetricsRecording:
     """
     Unit tests that verify metrics are properly recorded from Pipeline