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

Author: vladvildanov

Diff for: 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 for: 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 for: 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 for: 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"}]