diff --git a/tests/integration/standard/test_metadata.py b/tests/integration/standard/test_metadata.py
index 562f457a32..084b0376a3 100644
--- a/tests/integration/standard/test_metadata.py
+++ b/tests/integration/standard/test_metadata.py
@@ -16,7 +16,6 @@
 
 import logging
 import sys
-import time
 import os
 
 from packaging.version import Version
@@ -2387,13 +2386,11 @@ def test_base_table_column_addition_mv(self):
         assert "fouls" in score_table.columns
 
         # This is a workaround for mv notifications being separate from base table schema responses.
-        # This maybe fixed with future protocol changes
-        for i in range(10):
-            mv_alltime = self.cluster.metadata.keyspaces[self.keyspace_name].views["alltimehigh"]
-            if("fouls" in mv_alltime.columns):
-                break
-            time.sleep(.2)
-
+        # This maybe fixed with future protocol changes. CI load can push the lag well past a couple
+        # seconds, so poll for up to 30s (see https://github.com/scylladb/python-driver/issues/1020).
+        wait_until(lambda: "fouls" in self.cluster.metadata.keyspaces[self.keyspace_name].views["alltimehigh"].columns,
+                   delay=.5, max_attempts=60)
+        mv_alltime = self.cluster.metadata.keyspaces[self.keyspace_name].views["alltimehigh"]
         assert "fouls" in mv_alltime.columns
 
         mv_alltime_fouls_comumn = self.cluster.metadata.keyspaces[self.keyspace_name].views["alltimehigh"].columns['fouls']
@@ -2443,13 +2440,12 @@ def test_base_table_type_alter_mv(self):
         score_column = self.cluster.metadata.keyspaces[self.keyspace_name].tables['scores'].columns['score']
         assert score_column.cql_type == 'blob'
 
-        # until CASSANDRA-9920+CASSANDRA-10500 MV updates are only available later with an async event
-        for i in range(10):
-            score_mv_column = self.cluster.metadata.keyspaces[self.keyspace_name].views["monthlyhigh"].columns['score']
-            if "blob" == score_mv_column.cql_type:
-                break
-            time.sleep(.2)
-
+        # until CASSANDRA-9920+CASSANDRA-10500 MV updates are only available later with an async event.
+        # CI load can push the lag well past a couple seconds, so poll for up to 30s
+        # (see https://github.com/scylladb/python-driver/issues/1020).
+        wait_until(lambda: self.cluster.metadata.keyspaces[self.keyspace_name].views["monthlyhigh"].columns['score'].cql_type == 'blob',
+                   delay=.5, max_attempts=60)
+        score_mv_column = self.cluster.metadata.keyspaces[self.keyspace_name].views["monthlyhigh"].columns['score']
         assert score_mv_column.cql_type == 'blob'
 
     def test_metadata_with_quoted_identifiers(self):
diff --git a/tests/unit/test_wait_until.py b/tests/unit/test_wait_until.py
new file mode 100644
index 0000000000..ef2047ce53
--- /dev/null
+++ b/tests/unit/test_wait_until.py
@@ -0,0 +1,46 @@
+# Copyright DataStax, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import unittest
+from unittest.mock import patch
+
+from tests.util import wait_until
+
+
+class WaitUntilTests(unittest.TestCase):
+
+    def test_succeeds_immediately(self):
+        wait_until(lambda: True, delay=0, max_attempts=3)
+
+    def test_succeeds_on_final_poll(self):
+        """
+        The condition becoming true on the very last poll (after the last sleep,
+        with the attempt counter at max_attempts) must count as success, not a
+        timeout - see https://github.com/scylladb/python-driver/pull/1021.
+        """
+        calls = []
+
+        def condition():
+            calls.append(None)
+            return len(calls) > 3
+
+        with patch('tests.util.time.sleep'):
+            wait_until(condition, delay=0, max_attempts=3)
+
+        self.assertEqual(len(calls), 4)
+
+    def test_raises_after_exhausting_attempts(self):
+        with patch('tests.util.time.sleep'):
+            with self.assertRaises(Exception):
+                wait_until(lambda: False, delay=0, max_attempts=3)
diff --git a/tests/util.py b/tests/util.py
index 2439e20fd5..94a85a8a76 100644
--- a/tests/util.py
+++ b/tests/util.py
@@ -30,11 +30,13 @@ def wait_until(condition, delay, max_attempts):
                          of this function is delay*max_attempts
     """
     attempt = 0
-    while not condition() and attempt < max_attempts:
+    success = condition()
+    while not success and attempt < max_attempts:
         attempt += 1
         time.sleep(delay)
+        success = condition()
 
-    if attempt >= max_attempts:
+    if not success:
         raise Exception("Condition is still False after {} attempts.".format(max_attempts))