tests: replace fixed time.sleep() calls with polling (~17s saving)

- test_cluster.py: replace sleep(1) x10 iterations with
  connect(wait_for_all_pools=True) for deterministic pool readiness
- test_query.py: replace sleep(5) with wait_until polling for
  'Preparing all known prepared statements' log message
- test_connection.py: replace sleep(2) with wait_until polling for
  host_down listener notification

Author: mykaul

tests/integration/standard/test_cluster.py

@@ -1121,8 +1121,7 @@ def test_stale_connections_after_shutdown(self):
         """
         for _ in range(10):
             with TestCluster(protocol_version=3) as cluster:
-                cluster.connect().execute("SELECT * FROM system_schema.keyspaces")
-                time.sleep(1)
+                cluster.connect(wait_for_all_pools=True).execute("SELECT * FROM system_schema.keyspaces")
 
             with TestCluster(protocol_version=3) as cluster:
                 session = cluster.connect()

tests/integration/standard/test_connection.py

@@ -32,6 +32,7 @@
 from tests import is_monkey_patched
 from tests.integration import use_singledc, get_node, CASSANDRA_IP, local, \
     requiresmallclockgranularity, greaterthancass20, TestCluster
+from tests.util import wait_until
 
 try:
     import cassandra.io.asyncorereactor
@@ -140,9 +141,10 @@ def test_heart_beat_timeout(self):
             # Wait for connections associated with this host go away
             self.wait_for_no_connections(host, self.cluster)
 
-            # Wait to seconds for the driver to be notified
-            time.sleep(2)
-            assert test_listener.host_down
+            # Wait for the driver to detect the host is down
+            wait_until(
+                lambda: test_listener.host_down,
+                delay=0.5, max_attempts=20)
             # Resume paused node
         finally:
             node.resume()

tests/integration/standard/test_query.py

@@ -29,7 +29,7 @@
     USE_CASS_EXTERNAL, greaterthanorequalcass40, TestCluster, xfail_scylla
 from tests import notwindows
 from tests.integration import greaterthanorequalcass30, get_node
-from tests.util import assertListEqual
+from tests.util import assertListEqual, wait_until
 
 import time
 import random
@@ -1571,9 +1571,10 @@ def test_reprepare_after_host_is_down(self):
 
             get_node(1).start(wait_for_binary_proto=True, wait_other_notice=True)
 
-            # We wait for cluster._prepare_all_queries to be called
-            time.sleep(5)
-            assert 1 == mock_handler.get_message_count('debug', 'Preparing all known prepared statements')
+            # Wait for cluster._prepare_all_queries to be called
+            wait_until(
+                lambda: mock_handler.get_message_count('debug', 'Preparing all known prepared statements') >= 1,
+                delay=0.5, max_attempts=20)
 
             results = self.session.execute(prepared_statement, (1,), execution_profile="only_first")
             assert results.one() == (1, )