Also test replicas

Author: shahar

tests/dragonfly/cluster_test.py

@@ -288,15 +288,24 @@ async def test_cluster_slot_ownership_changes(df_local_factory):
 
 @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 = [
+    # Start and configure cluster with 3 masters and 3 replicas
+    masters = [
         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))
+    df_local_factory.start_all(masters)
+    c_masters = [aioredis.Redis(port=master.port) for master in masters]
+    c_masters_admin = [aioredis.Redis(port=master.admin_port) for master in masters]
+    master_ids = await asyncio.gather(*(get_node_id(c) for c in c_masters_admin))
+
+    replicas = [
+        df_local_factory.create(port=BASE_PORT+100+i, admin_port=BASE_PORT+i+1100)
+        for i in range(3)
+    ]
+    df_local_factory.start_all(replicas)
+    c_replicas = [aioredis.Redis(port=replica.port) for replica in replicas]
+    c_replicas_admin = [aioredis.Redis(port=replica.admin_port) for replica in replicas]
+    replica_ids = await asyncio.gather(*(get_node_id(c) for c in c_replicas_admin))
 
     config = f"""
       [
@@ -308,11 +317,17 @@ async def test_cluster_native_client(df_local_factory):
             }}
           ],
           "master": {{
-            "id": "{node_ids[0]}",
+            "id": "{master_ids[0]}",
             "ip": "localhost",
-            "port": {nodes[0].port}
+            "port": {masters[0].port}
           }},
-          "replicas": []
+          "replicas": [
+              {{
+                "id": "{replica_ids[0]}",
+                "ip": "localhost",
+                "port": {replicas[0].port}
+              }}
+          ]
         }},
         {{
           "slot_ranges": [
@@ -322,11 +337,17 @@ async def test_cluster_native_client(df_local_factory):
             }}
           ],
           "master": {{
-            "id": "{node_ids[1]}",
+            "id": "{master_ids[1]}",
             "ip": "localhost",
-            "port": {nodes[1].port}
+            "port": {masters[1].port}
           }},
-          "replicas": []
+          "replicas": [
+              {{
+                "id": "{replica_ids[1]}",
+                "ip": "localhost",
+                "port": {replicas[1].port}
+              }}
+          ]
         }},
         {{
           "slot_ranges": [
@@ -336,23 +357,34 @@ async def test_cluster_native_client(df_local_factory):
             }}
           ],
           "master": {{
-            "id": "{node_ids[2]}",
+            "id": "{master_ids[2]}",
             "ip": "localhost",
-            "port": {nodes[2].port}
+            "port": {masters[2].port}
           }},
-          "replicas": []
+          "replicas": [
+              {{
+                "id": "{replica_ids[2]}",
+                "ip": "localhost",
+                "port": {replicas[2].port}
+              }}
+          ]
         }}
       ]
     """
-    await push_config(config, c_nodes_admin)
+    await push_config(config, c_masters_admin + c_replicas_admin)
 
-    client = aioredis.RedisCluster(decode_responses=True, host="localhost", port=nodes[0].port)
+    client = aioredis.RedisCluster(decode_responses=True, host="localhost", port=masters[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'
 
+    # Make sure that getting a value from a replica works as well.
+    replica_response = await client.execute_command(
+            'get', 'key0', target_nodes=aioredis.RedisCluster.REPLICAS)
+    assert 'value' in replica_response.values()
+
     # Push new config
     config = f"""
       [
@@ -364,11 +396,17 @@ async def test_cluster_native_client(df_local_factory):
             }}
           ],
           "master": {{
-            "id": "{node_ids[0]}",
+            "id": "{master_ids[0]}",
             "ip": "localhost",
-            "port": {nodes[0].port}
+            "port": {masters[0].port}
           }},
-          "replicas": []
+          "replicas": [
+              {{
+                "id": "{replica_ids[0]}",
+                "ip": "localhost",
+                "port": {replicas[0].port}
+              }}
+          ]
         }},
         {{
           "slot_ranges": [
@@ -378,11 +416,17 @@ async def test_cluster_native_client(df_local_factory):
             }}
           ],
           "master": {{
-            "id": "{node_ids[1]}",
+            "id": "{master_ids[1]}",
             "ip": "localhost",
-            "port": {nodes[1].port}
+            "port": {masters[1].port}
           }},
-          "replicas": []
+          "replicas": [
+              {{
+                "id": "{replica_ids[1]}",
+                "ip": "localhost",
+                "port": {replicas[1].port}
+              }}
+          ]
         }},
         {{
           "slot_ranges": [
@@ -392,15 +436,21 @@ async def test_cluster_native_client(df_local_factory):
             }}
           ],
           "master": {{
-            "id": "{node_ids[2]}",
+            "id": "{master_ids[2]}",
             "ip": "localhost",
-            "port": {nodes[2].port}
+            "port": {masters[2].port}
           }},
-          "replicas": []
+          "replicas": [
+              {{
+                "id": "{replica_ids[2]}",
+                "ip": "localhost",
+                "port": {replicas[2].port}
+              }}
+          ]
         }}
       ]
     """
-    await push_config(config, c_nodes_admin)
+    await push_config(config, c_masters_admin + c_replicas_admin)
 
     for i in range(10_000):
         key = 'key' + str(i)