fix(cluster): fix #2532 tests redis client close errors (#2542)

* fix(cluster): fix #2532 tests redis client close errors

* refactor: address comments

tests/dragonfly/cluster_test.py

@@ -123,6 +123,8 @@ async def test_cluster_slots_in_replicas(df_local_factory):
         ip="127.0.0.1", port=master.port, answer=res[0], rep_ip="127.0.0.1", rep_port=replica.port
     )
 
+    await close_clients(c_master, c_replica)
+
 
 @dfly_args({"cluster_mode": "emulated", "cluster_announce_ip": "127.0.0.2"})
 async def test_cluster_info(async_client):
@@ -299,6 +301,8 @@ async def test_cluster_slot_ownership_changes(df_local_factory: DflyInstanceFact
     assert (await c_nodes[0].get("KEY0")) == "value"
     assert await c_nodes[1].execute_command("DBSIZE") == 0
 
+    await close_clients(*c_nodes, *c_nodes_admin)
+
 
 # Tests that master commands to the replica are applied regardless of slot ownership
 @dfly_args({"proactor_threads": 4, "cluster_mode": "yes"})
@@ -408,6 +412,8 @@ async def test_cluster_replica_sets_non_owned_keys(df_local_factory):
         assert await c_master.execute_command("dbsize") == 0
         assert await c_replica.execute_command("dbsize") == 0
 
+        await close_clients(c_master, c_master_admin, c_replica, c_replica_admin)
+
 
 @dfly_args({"proactor_threads": 4, "cluster_mode": "yes"})
 async def test_cluster_flush_slots_after_config_change(df_local_factory: DflyInstanceFactory):
@@ -515,6 +521,8 @@ async def test_cluster_flush_slots_after_config_change(df_local_factory: DflyIns
     assert await c_master.execute_command("dbsize") == (100_000 - slot_0_size)
     assert await c_replica.execute_command("dbsize") == (100_000 - slot_0_size)
 
+    await close_clients(c_master, c_master_admin, c_replica, c_replica_admin)
+
 
 @dfly_args({"proactor_threads": 4, "cluster_mode": "yes", "admin_port": 30001})
 async def test_cluster_blocking_command(df_server):
@@ -559,6 +567,8 @@ async def test_cluster_blocking_command(df_server):
         await v2
     assert "MOVED" in str(e_info.value)
 
+    await close_clients(c_master, c_master_admin)
+
 
 @dfly_args({"proactor_threads": 4, "cluster_mode": "yes"})
 async def test_cluster_native_client(
@@ -741,7 +751,7 @@ async def test_random_keys():
     await push_config(config, c_masters_admin + c_replicas_admin)
 
     await test_random_keys()
-    await client.close()
+    await close_clients(client, *c_masters, *c_masters_admin, *c_replicas, *c_replicas_admin)
 
 
 @dfly_args({"proactor_threads": 4, "cluster_mode": "yes"})
@@ -823,8 +833,7 @@ async def test_cluster_slot_migration(df_local_factory: DflyInstanceFactory):
         c_nodes_admin,
     )
 
-    await c_nodes_admin[0].close()
-    await c_nodes_admin[1].close()
+    await close_clients(*c_nodes, *c_nodes_admin)
 
 
 @dfly_args({"proactor_threads": 4, "cluster_mode": "yes"})
@@ -943,5 +952,4 @@ async def test_cluster_data_migration(df_local_factory: DflyInstanceFactory):
     assert await c_nodes[1].get("KEY19") == "value"
     assert await c_nodes[1].execute_command("DBSIZE") == 17
 
-    await c_nodes_admin[0].close()
-    await c_nodes_admin[1].close()
+    await close_clients(*c_nodes, *c_nodes_admin)

tests/dragonfly/utility.py

@@ -590,6 +590,10 @@ async def disconnect_clients(*clients):
     await asyncio.gather(*(c.connection_pool.disconnect() for c in clients))
 
 
+async def close_clients(*clients):
+    await asyncio.gather(*(c.close() for c in clients))
+
+
 def gen_ca_cert(ca_key_path, ca_cert_path):
     # We first need to generate the tls certificates to be used by the server