tests: replace medium-priority time.sleep() calls with polling

Replace fixed sleeps with condition-based polling in four test files:

- test_shard_aware.py: replace 25s of sleeps (5+10+5+5) with
  wait_until_not_raised polling for reconnection after shard connection
  close and iptables blocking
- test_metrics.py: replace 15s of sleeps (5+5+5) with polling for
  cluster recovery and node-down detection
- test_tablets.py: replace 13s of sleeps (3+10) with polling for
  metadata refresh and decommission completion
- simulacron/test_connection.py: replace 20s of sleeps (10+10) with
  polling for quiescent pool state

Total potential saving: ~73s of unconditional waiting.

Author: mykaul

tests/integration/simulacron/test_connection.py

@@ -23,7 +23,7 @@
 from cassandra.policies import HostStateListener, RoundRobinPolicy, WhiteListRoundRobinPolicy
 
 from tests import connection_class, thread_pool_executor_class
-from tests.util import late
+from tests.util import late, wait_until_not_raised
 from tests.integration import requiressimulacron, libevtest
 from tests.integration.util import assert_quiescent_pool_state
 # important to import the patch PROTOCOL_VERSION from the simulacron module
@@ -356,13 +356,15 @@ def test_retry_after_defunct(self):
         for _ in range(10):
             session.execute(query_to_prime)
 
-        # Might take some time to close the previous connections and reconnect
-        time.sleep(10)
-        assert_quiescent_pool_state(cluster)
+        # Wait for previous connections to close and pool to stabilize
+        wait_until_not_raised(
+            lambda: assert_quiescent_pool_state(cluster),
+            delay=1, max_attempts=30)
         clear_queries()
 
-        time.sleep(10)
-        assert_quiescent_pool_state(cluster)
+        wait_until_not_raised(
+            lambda: assert_quiescent_pool_state(cluster),
+            delay=1, max_attempts=30)
 
     def test_idle_connection_is_not_closed(self):
         """

tests/integration/standard/test_metrics.py

@@ -25,6 +25,7 @@
 
 from cassandra.cluster import NoHostAvailable, ExecutionProfile, EXEC_PROFILE_DEFAULT
 from tests.integration import get_cluster, get_node, use_singledc, execute_until_pass, TestCluster
+from tests.util import wait_until, wait_until_not_raised
 from cassandra import metrics
 from tests.integration import BasicSharedKeyspaceUnitTestCaseRF3WM, BasicExistingKeyspaceUnitTestCase, local
 
@@ -75,8 +76,10 @@ def test_connection_error(self):
                 self.session.execute(query)
         finally:
             get_cluster().start(wait_for_binary_proto=True, wait_other_notice=True)
-            # Give some time for the cluster to come back up, for the next test
-            time.sleep(5)
+            # Wait for the cluster to come back up for the next test
+            wait_until_not_raised(
+                lambda: self.session.execute("SELECT key FROM system.local WHERE key='local'"),
+                delay=0.5, max_attempts=30)
 
         assert self.cluster.metrics.stats.connection_errors > 0
 
@@ -156,7 +159,10 @@ def test_unavailable(self):
         # Sometimes this commands continues with the other nodes having not noticed
         # 1 is down, and a Timeout error is returned instead of an Unavailable
         get_node(1).stop(wait=True, wait_other_notice=True)
-        time.sleep(5)
+        wait_until(
+            lambda: not self.cluster.metadata.get_host('127.0.0.1') or
+                    not self.cluster.metadata.get_host('127.0.0.1').is_up,
+            delay=0.5, max_attempts=30)
         try:
             # Test write
             query = SimpleStatement("INSERT INTO test (k, v) VALUES (2, 2)", consistency_level=ConsistencyLevel.ALL)
@@ -171,8 +177,10 @@ def test_unavailable(self):
             assert self.cluster.metrics.stats.unavailables == 2
         finally:
             get_node(1).start(wait_other_notice=True, wait_for_binary_proto=True)
-            # Give some time for the cluster to come back up, for the next test
-            time.sleep(5)
+            # Wait for the cluster to come back up for the next test
+            wait_until_not_raised(
+                lambda: self.session.execute("SELECT key FROM system.local WHERE key='local'"),
+                delay=0.5, max_attempts=30)
 
         self.cluster.shutdown()
 

tests/integration/standard/test_shard_aware.py

@@ -27,6 +27,7 @@
 from cassandra import OperationTimedOut, ConsistencyLevel
 
 from tests.integration import use_cluster, get_node, PROTOCOL_VERSION
+from tests.util import wait_until_not_raised
 
 LOGGER = logging.getLogger(__name__)
 
@@ -178,11 +179,13 @@ def test_closing_connections(self):
                 continue
             shard_id = random.choice(list(pool._connections.keys()))
             pool._connections.get(shard_id).close()
-            time.sleep(5)
-            self.query_data(self.session, verify_in_tracing=False)
+            wait_until_not_raised(
+                lambda: self.query_data(self.session, verify_in_tracing=False),
+                delay=0.5, max_attempts=30)
 
-        time.sleep(10)
-        self.query_data(self.session)
+        wait_until_not_raised(
+            lambda: self.query_data(self.session),
+            delay=0.5, max_attempts=60)
 
     @pytest.mark.skip
     def test_blocking_connections(self):
@@ -212,13 +215,14 @@ def remove_iptables():
                  '--destination {node1_ip_address}/32 -j REJECT --reject-with icmp-port-unreachable'
                  ).format(node1_ip_address=node1_ip_address, node1_port=node1_port).split(' ')
                 )
-            time.sleep(5)
+            time.sleep(2)  # allow iptables rule to take effect
             try:
                 self.query_data(self.session, verify_in_tracing=False)
             except OperationTimedOut:
                 pass
             remove_iptables()
-            time.sleep(5)
-            self.query_data(self.session, verify_in_tracing=False)
+            wait_until_not_raised(
+                lambda: self.query_data(self.session, verify_in_tracing=False),
+                delay=0.5, max_attempts=30)
 
         self.query_data(self.session)

tests/integration/standard/test_tablets.py

@@ -6,6 +6,7 @@
 from cassandra.policies import ConstantReconnectionPolicy, RoundRobinPolicy, TokenAwarePolicy
 
 from tests.integration import PROTOCOL_VERSION, use_cluster, get_cluster
+from tests.util import wait_until
 from tests.unit.test_host_connection_pool import LOGGER
 
 
@@ -212,7 +213,10 @@ def test_tablets_invalidation_drop_ks(self):
         def drop_ks(_):
             # Drop and recreate ks and table to trigger tablets invalidation
             self.create_ks_and_cf(self.cluster.connect())
-            time.sleep(3)
+            # Wait for tablet metadata to be refreshed
+            wait_until(
+                lambda: 'test1' in self.cluster.metadata.keyspaces,
+                delay=0.5, max_attempts=20)
 
         self.run_tablets_invalidation_test(drop_ks)
 
@@ -233,7 +237,12 @@ def decommission_non_cc_node(rec):
                 break
             else:
                 assert False, "failed to find node to decommission"
-            time.sleep(10)
+            # Wait for decommission to complete and metadata to update
+            wait_until(
+                lambda: len([h for h in self.cluster.metadata.all_hosts() if h.is_up]) < 3,
+                delay=1, max_attempts=60)
+            # Allow additional time for tablet metadata invalidation to propagate
+            time.sleep(2)
 
         self.run_tablets_invalidation_test(decommission_non_cc_node)