From e20f354f0d94ba4982c465119d96911eb6dd9870 Mon Sep 17 00:00:00 2001
From: Vladyslav Vildanov <117659936+vladvildanov@users.noreply.github.com>
Date: Fri, 4 Oct 2024 13:08:48 +0300
Subject: [PATCH] Fix bug with async pipeline and cluster fails with some
 commands (#3402)

* Fix bug with async pipeline fails with some commands

* Codestyl changes

* Remove keys option in cluster context
---
 redis/asyncio/client.py             |  4 ++++
 redis/cluster.py                    |  4 ++++
 tests/test_asyncio/test_pipeline.py | 10 ++++++++++
 tests/test_pipeline.py              | 10 ++++++++++
 4 files changed, 28 insertions(+)

diff --git a/redis/asyncio/client.py b/redis/asyncio/client.py
index 039ebfdfae..9508849703 100644
--- a/redis/asyncio/client.py
+++ b/redis/asyncio/client.py
@@ -1423,6 +1423,10 @@ async def _execute_transaction(  # noqa: C901
             if not isinstance(r, Exception):
                 args, options = cmd
                 command_name = args[0]
+
+                # Remove keys entry, it needs only for cache.
+                options.pop("keys", None)
+
                 if command_name in self.response_callbacks:
                     r = self.response_callbacks[command_name](r, **options)
                     if inspect.isawaitable(r):
diff --git a/redis/cluster.py b/redis/cluster.py
index fbf5428d40..9dcbad7fc1 100644
--- a/redis/cluster.py
+++ b/redis/cluster.py
@@ -1163,6 +1163,10 @@ def _execute_command(self, target_node, *args, **kwargs):
                     asking = False
                 connection.send_command(*args, **kwargs)
                 response = redis_node.parse_response(connection, command, **kwargs)
+
+                # Remove keys entry, it needs only for cache.
+                kwargs.pop("keys", None)
+
                 if command in self.cluster_response_callbacks:
                     response = self.cluster_response_callbacks[command](
                         response, **kwargs
diff --git a/tests/test_asyncio/test_pipeline.py b/tests/test_asyncio/test_pipeline.py
index 4b29360d72..31759d84a3 100644
--- a/tests/test_asyncio/test_pipeline.py
+++ b/tests/test_asyncio/test_pipeline.py
@@ -417,3 +417,13 @@ async def test_pipeline_discard(self, r):
             response = await pipe.execute()
         assert response[0]
         assert await r.get("foo") == b"bar"
+
+    @pytest.mark.onlynoncluster
+    async def test_send_set_commands_over_async_pipeline(self, r: redis.asyncio.Redis):
+        pipe = r.pipeline()
+        pipe.hset("hash:1", "foo", "bar")
+        pipe.hset("hash:1", "bar", "foo")
+        pipe.hset("hash:1", "baz", "bar")
+        pipe.hgetall("hash:1")
+        resp = await pipe.execute()
+        assert resp == [1, 1, 1, {b"bar": b"foo", b"baz": b"bar", b"foo": b"bar"}]
diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py
index 7f10fcad4f..be7784ad0b 100644
--- a/tests/test_pipeline.py
+++ b/tests/test_pipeline.py
@@ -412,3 +412,13 @@ def test_pipeline_discard(self, r):
             response = pipe.execute()
         assert response[0]
         assert r.get("foo") == b"bar"
+
+    @pytest.mark.onlynoncluster
+    def test_send_set_commands_over_pipeline(self, r: redis.Redis):
+        pipe = r.pipeline()
+        pipe.hset("hash:1", "foo", "bar")
+        pipe.hset("hash:1", "bar", "foo")
+        pipe.hset("hash:1", "baz", "bar")
+        pipe.hgetall("hash:1")
+        resp = pipe.execute()
+        assert resp == [1, 1, 1, {b"bar": b"foo", b"baz": b"bar", b"foo": b"bar"}]