test: Introduce ensure_for helper

test/functional/feature_assumeutxo.py

@@ -9,7 +9,6 @@
 The assumeutxo value generated and used here is committed to in
 `CRegTestParams::m_assumeutxo_data` in `src/kernel/chainparams.cpp`.
 """
-import time
 from shutil import rmtree
 
 from dataclasses import dataclass
@@ -305,8 +304,7 @@ def test_sync_from_assumeutxo_node(self, snapshot):
         # If it does request such blocks, the snapshot_node will ignore requests it cannot fulfill, causing the ibd_node
         # to stall. This stall could last for up to 10 min, ultimately resulting in an abrupt disconnection due to the
         # ibd_node's perceived unresponsiveness.
-        time.sleep(3)  # Sleep here because we can't detect when a node avoids requesting blocks from other peer.
-        assert_equal(len(ibd_node.getpeerinfo()[0]['inflight']), 0)
+        ibd_node.ensure_for(duration=3, f=lambda: len(ibd_node.getpeerinfo()[0]['inflight']) == 0)
 
         # Now disconnect nodes and finish background chain sync
         self.disconnect_nodes(ibd_node.index, snapshot_node.index)

test/functional/feature_minchainwork.py

@@ -58,18 +58,14 @@ def run_test(self):
         hashes = self.generate(self.nodes[0], num_blocks_to_generate, sync_fun=self.no_op)
 
         self.log.info(f"Node0 current chain work: {self.nodes[0].getblockheader(hashes[-1])['chainwork']}")
-
-        # Sleep a few seconds and verify that node2 didn't get any new blocks
-        # or headers.  We sleep, rather than sync_blocks(node0, node1) because
-        # it's reasonable either way for node1 to get the blocks, or not get
-        # them (since they're below node1's minchainwork).
-        time.sleep(3)
-
         self.log.info("Verifying node 2 has no more blocks than before")
         self.log.info(f"Blockcounts: {[n.getblockcount() for n in self.nodes]}")
         # Node2 shouldn't have any new headers yet, because node1 should not
         # have relayed anything.
-        assert_equal(len(self.nodes[2].getchaintips()), 1)
+        # We wait 3 seconds, rather than sync_blocks(node0, node1) because
+        # it's reasonable either way for node1 to get the blocks, or not get
+        # them (since they're below node1's minchainwork).
+        self.nodes[2].ensure_for(duration=3, f=lambda: len(self.nodes[2].getchaintips()) == 1)
         assert_equal(self.nodes[2].getchaintips()[0]['height'], 0)
 
         assert self.nodes[1].getbestblockhash() != self.nodes[0].getbestblockhash()
@@ -81,8 +77,7 @@ def run_test(self):
         msg.locator.vHave = [int(self.nodes[2].getbestblockhash(), 16)]
         msg.hashstop = 0
         peer.send_and_ping(msg)
-        time.sleep(5)
-        assert "headers" not in peer.last_message or len(peer.last_message["headers"].headers) == 0
+        peer.ensure_for(duration=5, f=lambda: "headers" not in peer.last_message or len(peer.last_message["headers"].headers) == 0)
 
         self.log.info("Generating one more block")
         self.generate(self.nodes[0], 1)

test/functional/interface_zmq.py

@@ -394,11 +394,10 @@ def test_sequence(self):
         block_count = self.nodes[0].getblockcount()
         best_hash = self.nodes[0].getbestblockhash()
         self.nodes[0].invalidateblock(best_hash)
-        sleep(2)  # Bit of room to make sure transaction things happened
 
         # Make sure getrawmempool mempool_sequence results aren't "queued" but immediately reflective
         # of the time they were gathered.
-        assert self.nodes[0].getrawmempool(mempool_sequence=True)["mempool_sequence"] > seq_num
+        self.nodes[0].ensure_for(duration=2, f=lambda: self.nodes[0].getrawmempool(mempool_sequence=True)["mempool_sequence"] > seq_num)
 
         assert_equal((best_hash, "D", None), seq.receive_sequence())
         assert_equal((rbf_txid, "A", seq_num), seq.receive_sequence())

test/functional/mempool_unbroadcast.py

@@ -5,8 +5,6 @@
 """Test that the mempool ensures transaction delivery by periodically sending
 to peers until a GETDATA is received."""
 
-import time
-
 from test_framework.p2p import P2PTxInvStore
 from test_framework.test_framework import BitcoinTestFramework
 from test_framework.util import assert_equal
@@ -83,8 +81,8 @@ def test_broadcast(self):
 
         conn = node.add_p2p_connection(P2PTxInvStore())
         node.mockscheduler(MAX_INITIAL_BROADCAST_DELAY)
-        time.sleep(2) # allow sufficient time for possibility of broadcast
-        assert_equal(len(conn.get_invs()), 0)
+        # allow sufficient time for possibility of broadcast
+        conn.ensure_for(duration=2, f=lambda: len(conn.get_invs()) == 0)
 
         self.disconnect_nodes(0, 1)
         node.disconnect_p2ps()

test/functional/p2p_segwit.py

@@ -5,7 +5,6 @@
 """Test segwit transactions and blocks on P2P network."""
 from decimal import Decimal
 import random
-import time
 
 from test_framework.blocktools import (
     WITNESS_COMMITMENT_HEADER,
@@ -184,8 +183,7 @@ def announce_tx_and_wait_for_getdata(self, tx, success=True, use_wtxid=False):
             else:
                 self.wait_for_getdata([tx.sha256])
         else:
-            time.sleep(5)
-            assert not self.last_message.get("getdata")
+            self.ensure_for(duration=5, f=lambda: not self.last_message.get("getdata"))
 
     def announce_block_and_wait_for_getdata(self, block, use_header, timeout=60):
         with p2p_lock:

test/functional/test_framework/p2p.py

@@ -77,6 +77,7 @@
     sha256,
 )
 from test_framework.util import (
+    ensure_for_helper_internal,
     MAX_NODES,
     p2p_port,
     wait_until_helper_internal,
@@ -587,6 +588,9 @@ def test_function():
 
         wait_until_helper_internal(test_function, timeout=timeout, lock=p2p_lock, timeout_factor=self.timeout_factor)
 
+    def ensure_for(self, *, duration, f, delay=0.2):
+        return ensure_for_helper_internal(duration=duration, predicate=f, delay=delay)
+
     def wait_for_connect(self, *, timeout=60):
         test_function = lambda: self.is_connected
         self.wait_until(test_function, timeout=timeout, check_connected=False)

test/functional/test_framework/test_node.py

@@ -34,6 +34,7 @@
     assert_equal,
     append_config,
     delete_cookie_file,
+    ensure_for_helper_internal,
     get_auth_cookie,
     get_rpc_proxy,
     rpc_url,
@@ -841,6 +842,8 @@ def bumpmocktime(self, seconds):
     def wait_until(self, test_function, timeout=60):
         return wait_until_helper_internal(test_function, timeout=timeout, timeout_factor=self.timeout_factor)
 
+    def ensure_for(self, *, duration, f, delay=0.2):
+        return ensure_for_helper_internal(duration=duration, predicate=f, delay=delay)
 
 class TestNodeCLIAttr:
     def __init__(self, cli, command):

test/functional/test_framework/util.py

@@ -267,6 +267,13 @@ def satoshi_round(amount: Union[int, float, str], *, rounding: str) -> Decimal:
     """Rounds a Decimal amount to the nearest satoshi using the specified rounding mode."""
     return Decimal(amount).quantize(SATOSHI_PRECISION, rounding=rounding)
 
+def ensure_for_helper_internal(*, duration, predicate, delay=0.2):
+    time_end = time.time() + duration
+    predicate_source = "''''\n" + inspect.getsource(predicate) + "'''"
+    while time.time() < time_end:
+        if not predicate():
+            raise AssertionError("Predicate {} became false within {} seconds".format(predicate_source, duration))
+        time.sleep(delay)
 
 def wait_until_helper_internal(predicate, *, attempts=float('inf'), timeout=float('inf'), lock=None, timeout_factor=1.0):
     """Sleep until the predicate resolves to be True.

test/functional/wallet_multiwallet.py

@@ -12,7 +12,6 @@
 import platform
 import shutil
 import stat
-import time
 
 from test_framework.authproxy import JSONRPCException
 from test_framework.blocktools import COINBASE_MATURITY
@@ -373,8 +372,7 @@ def wallet_file(name):
         w2.encryptwallet('test')
         w2.walletpassphrase('test', 1)
         w2.unloadwallet()
-        time.sleep(1.1)
-        assert 'w2' not in self.nodes[0].listwallets()
+        self.nodes[0].ensure_for(duration=1.1, f=lambda: 'w2' not in self.nodes[0].listwallets())
 
         # Successfully unload all wallets
         for wallet_name in self.nodes[0].listwallets():