test: Update wait_until usage in tests not to use the one from utils

Replace "wait_until()" usage from utils, with the ones from BitcoinTestFramework and P2PInterface.
closes bitcoin#19080

test/functional/example_test.py

@@ -26,7 +26,6 @@
 from test_framework.util import (
     assert_equal,
     connect_nodes,
-    wait_until,
 )
 
 # P2PInterface is a class containing callbacks to be executed when a P2P
@@ -203,7 +202,7 @@ def run_test(self):
 
         # wait_until() will loop until a predicate condition is met. Use it to test properties of the
         # P2PInterface objects.
-        wait_until(lambda: sorted(blocks) == sorted(list(self.nodes[2].p2p.block_receive_map.keys())), timeout=5, lock=mininode_lock)
+        self.nodes[2].p2p.wait_until(lambda: sorted(blocks) == sorted(list(self.nodes[2].p2p.block_receive_map.keys())), timeout=5)
 
         self.log.info("Check that each block was received only once")
         # The network thread uses a global lock on data access to the P2PConnection objects when sending and receiving

test/functional/feature_abortnode.py

@@ -11,7 +11,7 @@
 """
 
 from test_framework.test_framework import BitcoinTestFramework
-from test_framework.util import wait_until, get_datadir_path, connect_nodes
+from test_framework.util import get_datadir_path, connect_nodes
 import os
 
 
@@ -41,7 +41,7 @@ def run_test(self):
 
             # Check that node0 aborted
             self.log.info("Waiting for crash")
-            wait_until(lambda: self.nodes[0].is_node_stopped(), timeout=200)
+            self.nodes[0].wait_until_stopped(timeout=200)
         self.log.info("Node crashed - now verifying restart fails")
         self.nodes[0].assert_start_raises_init_error()
 

test/functional/feature_notifications.py

@@ -9,7 +9,6 @@
 from test_framework.test_framework import BitcoinTestFramework
 from test_framework.util import (
     assert_equal,
-    wait_until,
     connect_nodes,
     disconnect_nodes,
     hex_str_to_bytes,
@@ -56,15 +55,15 @@ def run_test(self):
         blocks = self.nodes[1].generatetoaddress(block_count, self.nodes[1].getnewaddress() if self.is_wallet_compiled() else ADDRESS_BCRT1_UNSPENDABLE)
 
         # wait at most 10 seconds for expected number of files before reading the content
-        wait_until(lambda: len(os.listdir(self.blocknotify_dir)) == block_count, timeout=10)
+        self.wait_until(lambda: len(os.listdir(self.blocknotify_dir)) == block_count, timeout=10)
 
         # directory content should equal the generated blocks hashes
         assert_equal(sorted(blocks), sorted(os.listdir(self.blocknotify_dir)))
 
         if self.is_wallet_compiled():
             self.log.info("test -walletnotify")
             # wait at most 10 seconds for expected number of files before reading the content
-            wait_until(lambda: len(os.listdir(self.walletnotify_dir)) == block_count, timeout=10)
+            self.wait_until(lambda: len(os.listdir(self.walletnotify_dir)) == block_count, timeout=10)
 
             # directory content should equal the generated transaction hashes
             txids_rpc = list(map(lambda t: notify_outputname(self.wallet, t['txid']), self.nodes[1].listtransactions("*", block_count)))
@@ -78,7 +77,7 @@ def run_test(self):
             self.start_node(1)
             connect_nodes(self.nodes[0], 1)
 
-            wait_until(lambda: len(os.listdir(self.walletnotify_dir)) == block_count, timeout=10)
+            self.wait_until(lambda: len(os.listdir(self.walletnotify_dir)) == block_count, timeout=10)
 
             # directory content should equal the generated transaction hashes
             txids_rpc = list(map(lambda t: notify_outputname(self.wallet, t['txid']), self.nodes[1].listtransactions("*", block_count)))
@@ -140,7 +139,7 @@ def run_test(self):
         # TODO: add test for `-alertnotify` large fork notifications
 
     def expect_wallet_notify(self, tx_ids):
-        wait_until(lambda: len(os.listdir(self.walletnotify_dir)) >= len(tx_ids), timeout=10)
+        self.wait_until(lambda: len(os.listdir(self.walletnotify_dir)) >= len(tx_ids), timeout=10)
         assert_equal(sorted(notify_outputname(self.wallet, tx_id) for tx_id in tx_ids), sorted(os.listdir(self.walletnotify_dir)))
         for tx_file in os.listdir(self.walletnotify_dir):
             os.remove(os.path.join(self.walletnotify_dir, tx_file))

test/functional/feature_pruning.py

@@ -20,7 +20,6 @@
     assert_raises_rpc_error,
     connect_nodes,
     disconnect_nodes,
-    wait_until,
 )
 
 # Rescans start at the earliest block up to 2 hours before a key timestamp, so
@@ -136,7 +135,7 @@ def test_height_min(self):
         mine_large_blocks(self.nodes[0], 25)
 
         # Wait for blk00000.dat to be pruned
-        wait_until(lambda: not os.path.isfile(os.path.join(self.prunedir, "blk00000.dat")), timeout=30)
+        self.wait_until(lambda: not os.path.isfile(os.path.join(self.prunedir, "blk00000.dat")), timeout=30)
 
         self.log.info("Success")
         usage = calc_usage(self.prunedir)
@@ -250,7 +249,7 @@ def reorg_back(self):
 
         self.log.info("Verify node 2 reorged back to the main chain, some blocks of which it had to redownload")
         # Wait for Node 2 to reorg to proper height
-        wait_until(lambda: self.nodes[2].getblockcount() >= goalbestheight, timeout=900)
+        self.wait_until(lambda: self.nodes[2].getblockcount() >= goalbestheight, timeout=900)
         assert_equal(self.nodes[2].getbestblockhash(), goalbesthash)
         # Verify we can now have the data for a block previously pruned
         assert_equal(self.nodes[2].getblock(self.forkhash)["height"], self.forkheight)

test/functional/feature_shutdown.py

@@ -5,7 +5,7 @@
 """Test bitcoind shutdown."""
 
 from test_framework.test_framework import BitcoinTestFramework
-from test_framework.util import assert_equal, get_rpc_proxy, wait_until
+from test_framework.util import assert_equal, get_rpc_proxy
 from threading import Thread
 
 def test_long_call(node):
@@ -25,7 +25,7 @@ def run_test(self):
         node.getblockcount()
         Thread(target=test_long_call, args=(node,)).start()
         # Wait until the server is executing the above `waitfornewblock`.
-        wait_until(lambda: len(self.nodes[0].getrpcinfo()['active_commands']) == 2)
+        self.wait_until(lambda: len(self.nodes[0].getrpcinfo()['active_commands']) == 2)
         # Wait 1 second after requesting shutdown but not before the `stop` call
         # finishes. This is to ensure event loop waits for current connections
         # to close.

test/functional/feature_versionbits_warning.py

@@ -14,7 +14,6 @@
 from test_framework.messages import msg_block
 from test_framework.mininode import P2PInterface, mininode_lock
 from test_framework.test_framework import BitcoinTestFramework
-from test_framework.util import wait_until
 
 VB_PERIOD = 144           # versionbits period length for regtest
 VB_THRESHOLD = 108        # versionbits activation threshold for regtest
@@ -91,14 +90,14 @@ def run_test(self):
 
         # Generating one block guarantees that we'll get out of IBD
         node.generatetoaddress(1, node_deterministic_address)
-        wait_until(lambda: not node.getblockchaininfo()['initialblockdownload'], timeout=10, lock=mininode_lock)
+        self.wait_until(lambda: not node.getblockchaininfo()['initialblockdownload'], timeout=10, lock=mininode_lock)
         # Generating one more block will be enough to generate an error.
         node.generatetoaddress(1, node_deterministic_address)
         # Check that get*info() shows the versionbits unknown rules warning
         assert WARN_UNKNOWN_RULES_ACTIVE in node.getmininginfo()["warnings"]
         assert WARN_UNKNOWN_RULES_ACTIVE in node.getnetworkinfo()["warnings"]
         # Check that the alert file shows the versionbits unknown rules warning
-        wait_until(lambda: self.versionbits_in_alert_file(), timeout=60)
+        self.wait_until(lambda: self.versionbits_in_alert_file())
 
 if __name__ == '__main__':
     VersionBitsWarningTest().main()

test/functional/mempool_packages.py

@@ -13,7 +13,6 @@
     assert_equal,
     assert_raises_rpc_error,
     satoshi_round,
-    wait_until,
 )
 
 # default limits
@@ -269,8 +268,8 @@ def run_test(self):
         # - txs from previous ancestor test (-> custom ancestor limit)
         # - parent tx for descendant test
         # - txs chained off parent tx (-> custom descendant limit)
-        wait_until(lambda: len(self.nodes[1].getrawmempool(False)) ==
-                           MAX_ANCESTORS_CUSTOM + 1 + MAX_DESCENDANTS_CUSTOM, timeout=10)
+        self.wait_until(lambda: len(self.nodes[1].getrawmempool(False)) ==
+                                MAX_ANCESTORS_CUSTOM + 1 + MAX_DESCENDANTS_CUSTOM, timeout=10)
         mempool0 = self.nodes[0].getrawmempool(False)
         mempool1 = self.nodes[1].getrawmempool(False)
         assert set(mempool1).issubset(set(mempool0))

test/functional/mempool_persist.py

@@ -47,7 +47,6 @@
     assert_raises_rpc_error,
     connect_nodes,
     disconnect_nodes,
-    wait_until,
 )
 
 
@@ -172,7 +171,7 @@ def test_persist_unbroadcast(self):
         # check that txn gets broadcast due to unbroadcast logic
         conn = node0.add_p2p_connection(P2PTxInvStore())
         node0.mockscheduler(16*60) # 15 min + 1 for buffer
-        wait_until(lambda: len(conn.get_invs()) == 1)
+        self.wait_until(lambda: len(conn.get_invs()) == 1)
 
 if __name__ == '__main__':
     MempoolPersistTest().main()

test/functional/p2p_blockfilters.py

@@ -24,7 +24,6 @@
     assert_equal,
     connect_nodes,
     disconnect_nodes,
-    wait_until,
 )
 
 class CFiltersClient(P2PInterface):
@@ -65,11 +64,11 @@ def run_test(self):
         disconnect_nodes(self.nodes[0], 1)
 
         self.nodes[0].generate(1)
-        wait_until(lambda: self.nodes[0].getblockcount() == 1000)
+        self.wait_until(lambda: self.nodes[0].getblockcount() == 1000)
         stale_block_hash = self.nodes[0].getblockhash(1000)
 
         self.nodes[1].generate(1001)
-        wait_until(lambda: self.nodes[1].getblockcount() == 2000)
+        self.wait_until(lambda: self.nodes[1].getblockcount() == 2000)
 
         # Check that nodes have signalled NODE_COMPACT_FILTERS correctly.
         assert node0.nServices & NODE_COMPACT_FILTERS != 0

test/functional/p2p_compactblocks.py

@@ -14,7 +14,7 @@
 from test_framework.mininode import mininode_lock, P2PInterface
 from test_framework.script import CScript, OP_TRUE, OP_DROP
 from test_framework.test_framework import BitcoinTestFramework
-from test_framework.util import assert_equal, wait_until, softfork_active
+from test_framework.util import assert_equal, softfork_active
 
 # TestP2PConn: A peer we use to send messages to bitcoind, and store responses.
 class TestP2PConn(P2PInterface):
@@ -73,23 +73,23 @@ def send_header_for_blocks(self, new_blocks):
     def request_headers_and_sync(self, locator, hashstop=0):
         self.clear_block_announcement()
         self.get_headers(locator, hashstop)
-        wait_until(self.received_block_announcement, timeout=30, lock=mininode_lock)
+        self.wait_until(self.received_block_announcement, timeout=30)
         self.clear_block_announcement()
 
     # Block until a block announcement for a particular block hash is
     # received.
     def wait_for_block_announcement(self, block_hash, timeout=30):
         def received_hash():
             return (block_hash in self.announced_blockhashes)
-        wait_until(received_hash, timeout=timeout, lock=mininode_lock)
+        self.wait_until(received_hash, timeout=timeout)
 
     def send_await_disconnect(self, message, timeout=30):
         """Sends a message to the node and wait for disconnect.
 
         This is used when we want to send a message into the node that we expect
         will get us disconnected, eg an invalid block."""
         self.send_message(message)
-        wait_until(lambda: not self.is_connected, timeout=timeout, lock=mininode_lock)
+        self.wait_for_disconnect(timeout)
 
 class CompactBlocksTest(BitcoinTestFramework):
     def set_test_params(self):
@@ -154,7 +154,7 @@ def test_sendcmpct(self, test_node, old_node=None):
         # Make sure we get a SENDCMPCT message from our peer
         def received_sendcmpct():
             return (len(test_node.last_sendcmpct) > 0)
-        wait_until(received_sendcmpct, timeout=30, lock=mininode_lock)
+        test_node.wait_until(received_sendcmpct, timeout=30)
         with mininode_lock:
             # Check that the first version received is the preferred one
             assert_equal(test_node.last_sendcmpct[0].version, preferred_version)
@@ -294,7 +294,7 @@ def test_compactblock_construction(self, test_node, use_witness_address=True):
         block.rehash()
 
         # Wait until the block was announced (via compact blocks)
-        wait_until(lambda: "cmpctblock" in test_node.last_message, timeout=30, lock=mininode_lock)
+        test_node.wait_until(lambda: "cmpctblock" in test_node.last_message, timeout=30)
 
         # Now fetch and check the compact block
         header_and_shortids = None
@@ -308,7 +308,7 @@ def test_compactblock_construction(self, test_node, use_witness_address=True):
         inv = CInv(MSG_CMPCT_BLOCK, block_hash)
         test_node.send_message(msg_getdata([inv]))
 
-        wait_until(lambda: "cmpctblock" in test_node.last_message, timeout=30, lock=mininode_lock)
+        test_node.wait_until(lambda: "cmpctblock" in test_node.last_message, timeout=30)
 
         # Now fetch and check the compact block
         header_and_shortids = None
@@ -378,7 +378,7 @@ def test_compactblock_requests(self, test_node, segwit=True):
 
             if announce == "inv":
                 test_node.send_message(msg_inv([CInv(MSG_BLOCK, block.sha256)]))
-                wait_until(lambda: "getheaders" in test_node.last_message, timeout=30, lock=mininode_lock)
+                test_node.wait_until(lambda: "getheaders" in test_node.last_message, timeout=30)
                 test_node.send_header_for_blocks([block])
             else:
                 test_node.send_header_for_blocks([block])
@@ -588,7 +588,7 @@ def test_getblocktxn_handler(self, test_node):
             num_to_request = random.randint(1, len(block.vtx))
             msg.block_txn_request.from_absolute(sorted(random.sample(range(len(block.vtx)), num_to_request)))
             test_node.send_message(msg)
-            wait_until(lambda: "blocktxn" in test_node.last_message, timeout=10, lock=mininode_lock)
+            test_node.wait_until(lambda: "blocktxn" in test_node.last_message, timeout=10)
 
             [tx.calc_sha256() for tx in block.vtx]
             with mininode_lock:
@@ -628,20 +628,20 @@ def test_compactblocks_not_at_tip(self, test_node):
         for _ in range(MAX_CMPCTBLOCK_DEPTH + 1):
             test_node.clear_block_announcement()
             new_blocks.append(node.generate(1)[0])
-            wait_until(test_node.received_block_announcement, timeout=30, lock=mininode_lock)
+            test_node.wait_until(test_node.received_block_announcement, timeout=30)
 
         test_node.clear_block_announcement()
         test_node.send_message(msg_getdata([CInv(MSG_CMPCT_BLOCK, int(new_blocks[0], 16))]))
-        wait_until(lambda: "cmpctblock" in test_node.last_message, timeout=30, lock=mininode_lock)
+        test_node.wait_until(lambda: "cmpctblock" in test_node.last_message, timeout=30)
 
         test_node.clear_block_announcement()
         node.generate(1)
-        wait_until(test_node.received_block_announcement, timeout=30, lock=mininode_lock)
+        test_node.wait_until(test_node.received_block_announcement, timeout=30)
         test_node.clear_block_announcement()
         with mininode_lock:
             test_node.last_message.pop("block", None)
         test_node.send_message(msg_getdata([CInv(MSG_CMPCT_BLOCK, int(new_blocks[0], 16))]))
-        wait_until(lambda: "block" in test_node.last_message, timeout=30, lock=mininode_lock)
+        test_node.wait_until(lambda: "block" in test_node.last_message, timeout=30)
         with mininode_lock:
             test_node.last_message["block"].block.calc_sha256()
             assert_equal(test_node.last_message["block"].block.sha256, int(new_blocks[0], 16))
@@ -689,7 +689,7 @@ def test_end_to_end_block_relay(self, listeners):
         node.submitblock(ToHex(block))
 
         for l in listeners:
-            wait_until(lambda: "cmpctblock" in l.last_message, timeout=30, lock=mininode_lock)
+            l.wait_until(lambda: "cmpctblock" in l.last_message, timeout=30)
         with mininode_lock:
             for l in listeners:
                 l.last_message["cmpctblock"].header_and_shortids.header.calc_sha256()

test/functional/p2p_disconnect_ban.py

@@ -10,7 +10,6 @@
     assert_equal,
     assert_raises_rpc_error,
     connect_nodes,
-    wait_until,
 )
 
 class DisconnectBanTest(BitcoinTestFramework):
@@ -28,7 +27,7 @@ def run_test(self):
         self.log.info("setban: successfully ban single IP address")
         assert_equal(len(self.nodes[1].getpeerinfo()), 2)  # node1 should have 2 connections to node0 at this point
         self.nodes[1].setban(subnet="127.0.0.1", command="add")
-        wait_until(lambda: len(self.nodes[1].getpeerinfo()) == 0, timeout=10)
+        self.wait_until(lambda: len(self.nodes[1].getpeerinfo()) == 0, timeout=10)
         assert_equal(len(self.nodes[1].getpeerinfo()), 0)  # all nodes must be disconnected at this point
         assert_equal(len(self.nodes[1].listbanned()), 1)
 
@@ -95,7 +94,7 @@ def run_test(self):
         self.log.info("disconnectnode: successfully disconnect node by address")
         address1 = self.nodes[0].getpeerinfo()[0]['addr']
         self.nodes[0].disconnectnode(address=address1)
-        wait_until(lambda: len(self.nodes[0].getpeerinfo()) == 1, timeout=10)
+        self.wait_until(lambda: len(self.nodes[0].getpeerinfo()) == 1, timeout=10)
         assert not [node for node in self.nodes[0].getpeerinfo() if node['addr'] == address1]
 
         self.log.info("disconnectnode: successfully reconnect node")
@@ -106,7 +105,7 @@ def run_test(self):
         self.log.info("disconnectnode: successfully disconnect node by node id")
         id1 = self.nodes[0].getpeerinfo()[0]['id']
         self.nodes[0].disconnectnode(nodeid=id1)
-        wait_until(lambda: len(self.nodes[0].getpeerinfo()) == 1, timeout=10)
+        self.wait_until(lambda: len(self.nodes[0].getpeerinfo()) == 1, timeout=10)
         assert not [node for node in self.nodes[0].getpeerinfo() if node['id'] == id1]
 
 if __name__ == '__main__':

test/functional/p2p_eviction.py

@@ -17,7 +17,7 @@
 
 from test_framework.test_framework import BitcoinTestFramework
 from test_framework.mininode import P2PInterface, P2PDataStore
-from test_framework.util import assert_equal, wait_until
+from test_framework.util import assert_equal
 from test_framework.blocktools import create_block, create_coinbase
 from test_framework.messages import CTransaction, FromHex, msg_pong, msg_tx
 
@@ -92,7 +92,7 @@ def run_test(self):
         for _ in range(8):
             fastpeer = node.add_p2p_connection(P2PInterface())
             current_peer += 1
-            wait_until(lambda: "ping" in fastpeer.last_message, timeout=10)
+            self.wait_until(lambda: "ping" in fastpeer.last_message, timeout=10)
 
         # Make sure by asking the node what the actual min pings are
         peerinfo = node.getpeerinfo()

test/functional/p2p_feefilter.py

@@ -39,7 +39,7 @@ def on_inv(self, message):
 
     def wait_for_invs_to_match(self, invs_expected):
         invs_expected.sort()
-        self.wait_until(lambda: invs_expected == sorted(self.txinvs), timeout=60)
+        self.wait_until(lambda: invs_expected == sorted(self.txinvs))
 
     def clear_invs(self):
         with mininode_lock: