wait for both OK and same topo

Author: GuyAv46

RLTest/redis_cluster.py

@@ -50,25 +50,40 @@ def getInformationBeforeDispose(self):
     def getInformationAfterDispose(self):
         return [shard.getInformationAfterDispose() for shard in self.shards]
 
+    def _agreeOk(self):
+        ok = 0
+        for shard in self.shards:
+            con = shard.getConnection()
+            try:
+                status = con.execute_command('CLUSTER', 'INFO')
+            except Exception as e:
+                print('got error on cluster slots, will try again, %s' % str(e))
+                continue
+            if 'cluster_state:ok' in str(status):
+                ok += 1
+        return ok == len(self.shards)
+
+    def _agreeSlots(self):
+        ok = 0
+        first_view = None
+        for shard in self.shards:
+            con = shard.getConnection()
+            try:
+                slots_view = con.execute_command('CLUSTER', 'SLOTS')
+            except Exception as e:
+                print('got error on cluster slots, will try again, %s' % str(e))
+                continue
+            if first_view is None:
+                first_view = slots_view
+            if slots_view == first_view:
+                ok += 1
+        return ok == len(self.shards)
+
     def waitCluster(self, timeout_sec=40):
         st = time.time()
-        ok = 0
 
         while st + timeout_sec > time.time():
-            ok = 0
-            first_view = None
-            for shard in self.shards:
-                con = shard.getConnection()
-                try:
-                    slots_pov = con.execute_command('CLUSTER', 'SLOTS')
-                except Exception as e:
-                    print('got error on cluster info, will try again, %s' % str(e))
-                    continue
-                if first_view is None:
-                    first_view = slots_pov
-                if slots_pov == first_view:
-                    ok += 1
-            if ok == len(self.shards):
+            if self._agreeOk() and self._agreeSlots():
                 for shard in self.shards:
                     try:
                         shard.getConnection().execute_command('SEARCH.CLUSTERREFRESH')