Merge bitcoin/bitcoin#30893: test: Introduce ensure_for helper

111465d test: Remove unused attempts parameter from wait_until (Fabian Jahr)
5468a23 test: Add check_interval parameter to wait_until (Fabian Jahr)
16c87d9 test: Introduce ensure_for helper (Fabian Jahr)

Pull request description:

  A repeating pattern in the functional tests is that the test sleeps for a while to ensure that a certain condition is still true after some amount of time has elapsed. Most recently a new case of this was added in #30807. This PR here introduces an `ensure` helper to streamline this functionality.

  Some approach considerations:
  - It is possible to construct this by reusing `wait_until` and wrapping it in `try` internally. However, the logger output of the failing wait would still be printed which seems irritating. So I opted for simplified but similar internals to `wait_until`.
  - This implementation starts for a failure in the condition right away which has the nice side-effect that it might give feedback on a failure earlier than is currently the case. However, in some cases, it may be expected that the condition may still be false at the beginning and then turns true until time has run out, something that would work when the test sleeps without checking in a loop. I decided against this design (and even against adding it as an option) because such a test design seems like it would be racy either way.
  - I have also been going back and forth on naming. To me `ensure` works well but I am also not a native speaker, happy consider a different name if others don't think it's clear enough.

ACKs for top commit:
  maflcko:
    re-ACK 111465d 🍋
  achow101:
    ACK 111465d
  tdb3:
    code review re ACK 111465d
  furszy:
    utACK 111465d

Tree-SHA512: ce01a4f3531995375a6fbf01b27d51daa9d4c3d7cd10381be6e86ec5925d2965861000f7cb4796b8d40aabe3b64c4c27e2811270e4e3c9916689575b8ba4a2aa

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
@@ -31,6 +30,7 @@
     assert_approx,
     assert_equal,
     assert_raises_rpc_error,
+    ensure_for,
     sha256sum_file,
     try_rpc,
 )
@@ -305,8 +305,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)
+        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

@@ -19,7 +19,10 @@
 
 from test_framework.p2p import P2PInterface, msg_getheaders
 from test_framework.test_framework import BitcoinTestFramework
-from test_framework.util import assert_equal
+from test_framework.util import (
+    assert_equal,
+    ensure_for,
+)
 
 # 2 hashes required per regtest block (with no difficulty adjustment)
 REGTEST_WORK_PER_BLOCK = 2
@@ -58,18 +61,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).
+        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 +80,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
+        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

@@ -6,7 +6,6 @@
 import os
 import struct
 import tempfile
-from time import sleep
 from io import BytesIO
 
 from test_framework.address import (
@@ -27,6 +26,7 @@
 from test_framework.util import (
     assert_equal,
     assert_raises_rpc_error,
+    ensure_for,
     p2p_port,
 )
 from test_framework.wallet import (
@@ -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
+        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,11 +5,12 @@
 """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
+from test_framework.util import (
+    assert_equal,
+    ensure_for,
+)
 from test_framework.wallet import MiniWallet
 
 MAX_INITIAL_BROADCAST_DELAY = 15 * 60 # 15 minutes in seconds
@@ -83,8 +84,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
+        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,
@@ -83,8 +82,9 @@
 from test_framework.test_framework import BitcoinTestFramework
 from test_framework.util import (
     assert_equal,
-    softfork_active,
     assert_raises_rpc_error,
+    ensure_for,
+    softfork_active,
 )
 from test_framework.wallet import MiniWallet
 from test_framework.wallet_util import generate_keypair
@@ -184,8 +184,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")
+            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

@@ -585,13 +585,13 @@ def on_version(self, message):
 
     # Connection helper methods
 
-    def wait_until(self, test_function_in, *, timeout=60, check_connected=True):
+    def wait_until(self, test_function_in, *, timeout=60, check_connected=True, check_interval=0.05):
         def test_function():
             if check_connected:
                 assert self.is_connected
             return test_function_in()
 
-        wait_until_helper_internal(test_function, timeout=timeout, lock=p2p_lock, timeout_factor=self.timeout_factor)
+        wait_until_helper_internal(test_function, timeout=timeout, lock=p2p_lock, timeout_factor=self.timeout_factor, check_interval=check_interval)
 
     def wait_for_connect(self, *, timeout=60):
         test_function = lambda: self.is_connected

test/functional/test_framework/test_framework.py

@@ -787,8 +787,8 @@ def sync_all(self, nodes=None):
         self.sync_blocks(nodes)
         self.sync_mempools(nodes)
 
-    def wait_until(self, test_function, timeout=60):
-        return wait_until_helper_internal(test_function, timeout=timeout, timeout_factor=self.options.timeout_factor)
+    def wait_until(self, test_function, timeout=60, check_interval=0.05):
+        return wait_until_helper_internal(test_function, timeout=timeout, timeout_factor=self.options.timeout_factor, check_interval=check_interval)
 
     # Private helper methods. These should not be accessed by the subclass test scripts.
 

test/functional/test_framework/test_node.py

@@ -837,8 +837,8 @@ def bumpmocktime(self, seconds):
         self.mocktime += seconds
         self.setmocktime(self.mocktime)
 
-    def wait_until(self, test_function, timeout=60):
-        return wait_until_helper_internal(test_function, timeout=timeout, timeout_factor=self.timeout_factor)
+    def wait_until(self, test_function, timeout=60, check_interval=0.05):
+        return wait_until_helper_internal(test_function, timeout=timeout, timeout_factor=self.timeout_factor, check_interval=check_interval)
 
 
 class TestNodeCLIAttr:

test/functional/test_framework/util.py

@@ -268,7 +268,28 @@ def satoshi_round(amount: Union[int, float, str], *, rounding: str) -> Decimal:
     return Decimal(amount).quantize(SATOSHI_PRECISION, rounding=rounding)
 
 
-def wait_until_helper_internal(predicate, *, attempts=float('inf'), timeout=float('inf'), lock=None, timeout_factor=1.0):
+def ensure_for(*, duration, f, check_interval=0.2):
+    """Check if the predicate keeps returning True for duration.
+
+    check_interval can be used to configure the wait time between checks.
+    Setting check_interval to 0 will allow to have two checks: one in the
+    beginning and one after duration.
+    """
+    # If check_interval is 0 or negative or larger than duration, we fall back
+    # to checking once in the beginning and once at the end of duration
+    if check_interval <= 0 or check_interval > duration:
+        check_interval = duration
+    time_end = time.time() + duration
+    predicate_source = "''''\n" + inspect.getsource(f) + "'''"
+    while True:
+        if not f():
+            raise AssertionError(f"Predicate {predicate_source} became false within {duration} seconds")
+        if time.time() > time_end:
+            return
+        time.sleep(check_interval)
+
+
+def wait_until_helper_internal(predicate, *, timeout=60, lock=None, timeout_factor=1.0, check_interval=0.05):
     """Sleep until the predicate resolves to be True.
 
     Warning: Note that this method is not recommended to be used in tests as it is
@@ -277,31 +298,23 @@ def wait_until_helper_internal(predicate, *, attempts=float('inf'), timeout=floa
     properly scaled. Furthermore, `wait_until()` from `P2PInterface` class in
     `p2p.py` has a preset lock.
     """
-    if attempts == float('inf') and timeout == float('inf'):
-        timeout = 60
     timeout = timeout * timeout_factor
-    attempt = 0
     time_end = time.time() + timeout
 
-    while attempt < attempts and time.time() < time_end:
+    while time.time() < time_end:
         if lock:
             with lock:
                 if predicate():
                     return
         else:
             if predicate():
                 return
-        attempt += 1
-        time.sleep(0.05)
+        time.sleep(check_interval)
 
     # Print the cause of the timeout
     predicate_source = "''''\n" + inspect.getsource(predicate) + "'''"
     logger.error("wait_until() failed. Predicate: {}".format(predicate_source))
-    if attempt >= attempts:
-        raise AssertionError("Predicate {} not true after {} attempts".format(predicate_source, attempts))
-    elif time.time() >= time_end:
-        raise AssertionError("Predicate {} not true after {} seconds".format(predicate_source, timeout))
-    raise RuntimeError('Unreachable')
+    raise AssertionError("Predicate {} not true after {} seconds".format(predicate_source, timeout))
 
 
 def sha256sum_file(filename):

test/functional/wallet_inactive_hdchains.py

@@ -6,7 +6,6 @@
 Test Inactive HD Chains.
 """
 import shutil
-import time
 
 from test_framework.authproxy import JSONRPCException
 from test_framework.test_framework import BitcoinTestFramework
@@ -75,12 +74,13 @@ def do_inactive_test(self, base_wallet, test_wallet, encrypt=False):
         self.generate(self.nodes[0], 1)
 
         # Wait for the test wallet to see the transaction
-        while True:
+        def is_tx_available(txid):
             try:
                 test_wallet.gettransaction(txid)
-                break
+                return True
             except JSONRPCException:
-                time.sleep(0.1)
+                return False
+        self.nodes[0].wait_until(lambda: is_tx_available(txid), timeout=10, check_interval=0.1)
 
         if encrypt:
             # The test wallet will not be able to generate the topped up keypool

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
@@ -21,6 +20,7 @@
 from test_framework.util import (
     assert_equal,
     assert_raises_rpc_error,
+    ensure_for,
     get_rpc_proxy,
 )
 
@@ -373,8 +373,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()
+        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():