test(cluster): Test that cluster works with a standard cluster client.

In this case, `redis.RedisCluster`.

To be double sure I also looked at the actual packets and saw that the
client asks for `CLUSTER SLOTS`, and then after the redistribution of
slots, following a few `MOVED` replied, it asks for the new slots again.

Author: shahar

tests/dragonfly/cluster_test.py

@@ -284,3 +284,127 @@ async def test_cluster_slot_ownership_changes(df_local_factory):
     assert await c_nodes[0].execute_command("DBSIZE") == 1
     assert (await c_nodes[0].get("KEY0")).decode() == "value"
     assert await c_nodes[1].execute_command("DBSIZE") == 0
+
+
+@dfly_args({"proactor_threads": 4, "cluster_mode": "yes"})
+async def test_cluster_native_client(df_local_factory):
+    # Start and configure cluster with 3 nodes
+    nodes = [
+        df_local_factory.create(port=BASE_PORT+i, admin_port=BASE_PORT+i+1000)
+        for i in range(3)
+    ]
+    df_local_factory.start_all(nodes)
+    c_nodes = [aioredis.Redis(port=node.port) for node in nodes]
+    c_nodes_admin = [aioredis.Redis(port=node.admin_port) for node in nodes]
+    node_ids = await asyncio.gather(*(get_node_id(c) for c in c_nodes_admin))
+
+    config = f"""
+      [
+        {{
+          "slot_ranges": [
+            {{
+              "start": 0,
+              "end": 5000
+            }}
+          ],
+          "master": {{
+            "id": "{node_ids[0]}",
+            "ip": "localhost",
+            "port": {nodes[0].port}
+          }},
+          "replicas": []
+        }},
+        {{
+          "slot_ranges": [
+            {{
+              "start": 5001,
+              "end": 10000
+            }}
+          ],
+          "master": {{
+            "id": "{node_ids[1]}",
+            "ip": "localhost",
+            "port": {nodes[1].port}
+          }},
+          "replicas": []
+        }},
+        {{
+          "slot_ranges": [
+            {{
+              "start": 10001,
+              "end": 16383
+            }}
+          ],
+          "master": {{
+            "id": "{node_ids[2]}",
+            "ip": "localhost",
+            "port": {nodes[2].port}
+          }},
+          "replicas": []
+        }}
+      ]
+    """
+    await push_config(config, c_nodes_admin)
+
+    client = aioredis.RedisCluster(decode_responses=True, host="localhost", port=nodes[0].port)
+
+    for i in range(10_000):
+        key = 'key' + str(i)
+        assert await client.set(key, 'value') == True
+        assert await client.get(key) == 'value'
+
+    # Push new config
+    config = f"""
+      [
+        {{
+          "slot_ranges": [
+            {{
+              "start": 0,
+              "end": 4000
+            }}
+          ],
+          "master": {{
+            "id": "{node_ids[0]}",
+            "ip": "localhost",
+            "port": {nodes[0].port}
+          }},
+          "replicas": []
+        }},
+        {{
+          "slot_ranges": [
+            {{
+              "start": 4001,
+              "end": 14000
+            }}
+          ],
+          "master": {{
+            "id": "{node_ids[1]}",
+            "ip": "localhost",
+            "port": {nodes[1].port}
+          }},
+          "replicas": []
+        }},
+        {{
+          "slot_ranges": [
+            {{
+              "start": 14001,
+              "end": 16383
+            }}
+          ],
+          "master": {{
+            "id": "{node_ids[2]}",
+            "ip": "localhost",
+            "port": {nodes[2].port}
+          }},
+          "replicas": []
+        }}
+      ]
+    """
+    await push_config(config, c_nodes_admin)
+
+    for i in range(10_000):
+        key = 'key' + str(i)
+        assert await client.set(key, 'value') == True
+        assert await client.get(key) == 'value'
+
+    await client.close()