Fix bug with async pipeline fails with some commands

Author: vladvildanov

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):

tests/test_asyncio/test_pipeline.py

@@ -417,3 +417,16 @@ 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'}
+        ]

tests/test_pipeline.py

@@ -412,3 +412,16 @@ 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'}
+        ]