test: Use test framework utils in functional tests

Replaces bare asserts with test framework utils across both the
functional tests and the test framework itself.

Also adds the `assert_not_equal`, `assert_less_than`, and
`assert_less_than_or_equal` util functions for greater readability.

    Fixes #23119.

test/functional/data/invalid_txs.py

@@ -57,6 +57,9 @@
     MIN_STANDARD_TX_NONWITNESS_SIZE,
     script_to_p2sh_script,
 )
+from test_framework.util import (
+    assert_equal,
+)
 basic_p2sh = script_to_p2sh_script(CScript([OP_0]))
 
 class BadTxTemplate:
@@ -124,8 +127,8 @@ def get_tx(self):
         tx = CTransaction()
         tx.vin.append(self.valid_txin)
         tx.vout.append(CTxOut(0, CScript([OP_RETURN] + ([OP_0] * (MIN_PADDING - 2)))))
-        assert len(tx.serialize_without_witness()) == 64
-        assert MIN_STANDARD_TX_NONWITNESS_SIZE - 1 == 64
+        assert_equal(len(tx.serialize_without_witness()), 64)
+        assert_equal(MIN_STANDARD_TX_NONWITNESS_SIZE - 1, 64)
         tx.calc_sha256()
         return tx
 

test/functional/feature_bind_port_discover.py

@@ -7,7 +7,10 @@
 """
 
 from test_framework.test_framework import BitcoinTestFramework, SkipTest
-from test_framework.util import assert_equal
+from test_framework.util import (
+    assert_equal,
+    assert_not_equal,
+)
 
 # We need to bind to a routable address for this test to exercise the relevant code
 # and also must have another routable address on another interface which must not
@@ -71,7 +74,7 @@ def run_test(self):
             if local['address'] == ADDR1:
                 found_addr1 = True
                 assert_equal(local['port'], BIND_PORT)
-            assert local['address'] != ADDR2
+            assert_not_equal(local['address'], ADDR2)
         assert found_addr1
 
 if __name__ == '__main__':

test/functional/feature_block.py

@@ -51,6 +51,8 @@
 from test_framework.test_framework import BitcoinTestFramework
 from test_framework.util import (
     assert_equal,
+    assert_less_than,
+    assert_less_than_or_equal,
     assert_greater_than,
 )
 from test_framework.wallet_util import generate_keypair
@@ -527,7 +529,7 @@ def run_test(self):
         b40 = self.next_block(40, spend=out[12])
         sigops = get_legacy_sigopcount_block(b40)
         numTxes = (MAX_BLOCK_SIGOPS - sigops) // b39_sigops_per_output
-        assert_equal(numTxes <= b39_outputs, True)
+        assert_less_than_or_equal(numTxes, b39_outputs)
 
         lastOutpoint = COutPoint(b40.vtx[1].sha256, 0)
         new_txs = []
@@ -798,7 +800,7 @@ def run_test(self):
         self.move_tip(57)
         self.next_block(58, spend=out[17])
         tx = CTransaction()
-        assert len(out[17].vout) < 42
+        assert_less_than(len(out[17].vout), 42)
         tx.vin.append(CTxIn(COutPoint(out[17].sha256, 42), CScript([OP_TRUE]), SEQUENCE_FINAL))
         tx.vout.append(CTxOut(0, b""))
         tx.calc_sha256()

test/functional/feature_coinstatsindex.py

@@ -28,6 +28,8 @@
 from test_framework.test_framework import BitcoinTestFramework
 from test_framework.util import (
     assert_equal,
+    assert_greater_than,
+    assert_not_equal,
     assert_raises_rpc_error,
 )
 from test_framework.wallet import (
@@ -226,7 +228,7 @@ def _test_coin_stats_index(self):
 
         self.generate(index_node, 1, sync_fun=self.no_op)
         res10 = index_node.gettxoutsetinfo('muhash')
-        assert res8['txouts'] < res10['txouts']
+        assert_greater_than(res10['txouts'], res8['txouts'])
 
         self.log.info("Test that the index works with -reindex")
 
@@ -272,12 +274,12 @@ def _test_reorg_index(self):
         res2 = index_node.gettxoutsetinfo(hash_type='muhash', hash_or_height=112)
         assert_equal(res["bestblock"], block)
         assert_equal(res["muhash"], res2["muhash"])
-        assert res["muhash"] != res_invalid["muhash"]
+        assert_not_equal(res["muhash"], res_invalid["muhash"])
 
         # Test that requesting reorged out block by hash is still returning correct results
         res_invalid2 = index_node.gettxoutsetinfo(hash_type='muhash', hash_or_height=reorg_block)
         assert_equal(res_invalid2["muhash"], res_invalid["muhash"])
-        assert res["muhash"] != res_invalid2["muhash"]
+        assert_not_equal(res["muhash"], res_invalid2["muhash"])
 
         # Add another block, so we don't depend on reconsiderblock remembering which
         # blocks were touched by invalidateblock

test/functional/feature_csv_activation.py

@@ -53,6 +53,7 @@
 from test_framework.test_framework import BitcoinTestFramework
 from test_framework.util import (
     assert_equal,
+    assert_greater_than_or_equal,
     softfork_active,
 )
 from test_framework.wallet import (
@@ -131,7 +132,7 @@ def send_generic_input_tx(self, coinbases):
     def create_bip68txs(self, bip68inputs, txversion, locktime_delta=0):
         """Returns a list of bip68 transactions with different bits set."""
         txs = []
-        assert len(bip68inputs) >= 16
+        assert_greater_than_or_equal(len(bip68inputs), 16)
         for i, (sdf, srhb, stf, srlb) in enumerate(product(*[[True, False]] * 4)):
             locktime = relative_locktime(sdf, srhb, stf, srlb)
             tx = self.create_self_transfer_from_utxo(bip68inputs[i])
@@ -145,7 +146,7 @@ def create_bip68txs(self, bip68inputs, txversion, locktime_delta=0):
     def create_bip112txs(self, bip112inputs, varyOP_CSV, txversion, locktime_delta=0):
         """Returns a list of bip68 transactions with different bits set."""
         txs = []
-        assert len(bip112inputs) >= 16
+        assert_greater_than_or_equal(len(bip112inputs), 16)
         for i, (sdf, srhb, stf, srlb) in enumerate(product(*[[True, False]] * 4)):
             locktime = relative_locktime(sdf, srhb, stf, srlb)
             tx = self.create_self_transfer_from_utxo(bip112inputs[i])

test/functional/feature_dbcrash.py

@@ -37,6 +37,8 @@
 from test_framework.test_framework import BitcoinTestFramework
 from test_framework.util import (
     assert_equal,
+    assert_greater_than,
+    assert_not_equal,
 )
 from test_framework.wallet import (
     MiniWallet,
@@ -274,10 +276,10 @@ def run_test(self):
         self.log.info(f"Restarted nodes: {self.restart_counts}; crashes on restart: {self.crashed_on_restart}")
 
         # If no nodes were restarted, we didn't test anything.
-        assert self.restart_counts != [0, 0, 0]
+        assert_not_equal(self.restart_counts, [0, 0, 0])
 
         # Make sure we tested the case of crash-during-recovery.
-        assert self.crashed_on_restart > 0
+        assert_greater_than(self.crashed_on_restart, 0)
 
         # Warn if any of the nodes escaped restart.
         for i in range(3):

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,
+    assert_not_equal,
+)
 
 # 2 hashes required per regtest block (with no difficulty adjustment)
 REGTEST_WORK_PER_BLOCK = 2
@@ -72,7 +75,7 @@ def run_test(self):
         assert_equal(len(self.nodes[2].getchaintips()), 1)
         assert_equal(self.nodes[2].getchaintips()[0]['height'], 0)
 
-        assert self.nodes[1].getbestblockhash() != self.nodes[0].getbestblockhash()
+        assert_not_equal(self.nodes[1].getbestblockhash(), self.nodes[0].getbestblockhash())
         assert_equal(self.nodes[2].getblockcount(), starting_blockcount)
 
         self.log.info("Check that getheaders requests to node2 are ignored")

test/functional/feature_posix_fs_permissions.py

@@ -9,6 +9,7 @@
 import stat
 
 from test_framework.test_framework import BitcoinTestFramework
+from test_framework.util import assert_equal
 
 
 class PosixFsPermissionsTest(BitcoinTestFramework):
@@ -22,12 +23,12 @@ def skip_test_if_missing_module(self):
     def check_directory_permissions(self, dir):
         mode = os.lstat(dir).st_mode
         self.log.info(f"{stat.filemode(mode)} {dir}")
-        assert mode == (stat.S_IFDIR | stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR)
+        assert_equal(mode, (stat.S_IFDIR | stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR))
 
     def check_file_permissions(self, file):
         mode = os.lstat(file).st_mode
         self.log.info(f"{stat.filemode(mode)} {file}")
-        assert mode == (stat.S_IFREG | stat.S_IRUSR | stat.S_IWUSR)
+        assert_equal(mode, (stat.S_IFREG | stat.S_IRUSR | stat.S_IWUSR))
 
     def run_test(self):
         self.stop_node(0)

test/functional/feature_pruning.py

@@ -197,7 +197,7 @@ def reorg_test(self):
             self.nodes[1].invalidateblock(curhash)
             curhash = self.nodes[1].getblockhash(self.forkheight - 1)
 
-        assert self.nodes[1].getblockcount() == self.forkheight - 1
+        assert_equal(self.nodes[1].getblockcount(), self.forkheight - 1)
         self.log.info(f"New best height: {self.nodes[1].getblockcount()}")
 
         # Disconnect node1 and generate the new chain

test/functional/feature_rbf.py

@@ -14,6 +14,8 @@
 from test_framework.test_framework import BitcoinTestFramework
 from test_framework.util import (
     assert_equal,
+    assert_greater_than,
+    assert_greater_than_or_equal,
     assert_raises_rpc_error,
 )
 from test_framework.wallet import MiniWallet
@@ -102,7 +104,7 @@ def make_utxo(self, node, amount, *, confirmed=True, scriptPubKey=None):
                 new_size = len(node.getrawmempool())
                 # Error out if we have something stuck in the mempool, as this
                 # would likely be a bug.
-                assert new_size < mempool_size
+                assert_greater_than(mempool_size, new_size)
                 mempool_size = new_size
 
         return self.wallet.get_utxo(txid=tx["txid"], vout=tx["sent_vout"])
@@ -445,9 +447,9 @@ def test_too_many_replacements_with_default_mempool_params(self):
             num_txs_invalidated = len(root_utxos) + (num_tx_graphs * txs_per_graph)
 
             if failure_expected:
-                assert num_txs_invalidated > MAX_REPLACEMENT_LIMIT
+                assert_greater_than(num_txs_invalidated, MAX_REPLACEMENT_LIMIT)
             else:
-                assert num_txs_invalidated <= MAX_REPLACEMENT_LIMIT
+                assert_greater_than_or_equal(MAX_REPLACEMENT_LIMIT, num_txs_invalidated)
 
             # Now attempt to submit a tx that double-spends all the root tx inputs, which
             # would invalidate `num_txs_invalidated` transactions.

test/functional/feature_segwit.py

@@ -46,6 +46,7 @@
     assert_equal,
     assert_greater_than_or_equal,
     assert_is_hex_string,
+    assert_not_equal,
     assert_raises_rpc_error,
     try_rpc,
 )
@@ -225,16 +226,16 @@ def run_test(self):
         self.fail_accept(self.nodes[0], "mandatory-script-verify-flag-failed (Witness program was passed an empty witness)", p2sh_ids[NODE_0][P2WSH][0], sign=False, redeem_script=witness_script(True, self.pubkey[0]))
 
         self.log.info("Verify block and transaction serialization rpcs return differing serializations depending on rpc serialization flag")
-        assert self.nodes[2].getblock(blockhash, False) != self.nodes[0].getblock(blockhash, False)
-        assert self.nodes[1].getblock(blockhash, False) == self.nodes[2].getblock(blockhash, False)
+        assert_not_equal(self.nodes[2].getblock(blockhash, False), self.nodes[0].getblock(blockhash, False))
+        assert_equal(self.nodes[1].getblock(blockhash, False), self.nodes[2].getblock(blockhash, False))
 
         for tx_id in segwit_tx_list:
             tx = tx_from_hex(self.nodes[2].gettransaction(tx_id)["hex"])
-            assert self.nodes[2].getrawtransaction(tx_id, False, blockhash) != self.nodes[0].getrawtransaction(tx_id, False, blockhash)
-            assert self.nodes[1].getrawtransaction(tx_id, False, blockhash) == self.nodes[2].getrawtransaction(tx_id, False, blockhash)
-            assert self.nodes[0].getrawtransaction(tx_id, False, blockhash) != self.nodes[2].gettransaction(tx_id)["hex"]
-            assert self.nodes[1].getrawtransaction(tx_id, False, blockhash) == self.nodes[2].gettransaction(tx_id)["hex"]
-            assert self.nodes[0].getrawtransaction(tx_id, False, blockhash) == tx.serialize_without_witness().hex()
+            assert_not_equal(self.nodes[2].getrawtransaction(tx_id, False, blockhash), self.nodes[0].getrawtransaction(tx_id, False, blockhash))
+            assert_equal(self.nodes[1].getrawtransaction(tx_id, False, blockhash), self.nodes[2].getrawtransaction(tx_id, False, blockhash))
+            assert_not_equal(self.nodes[0].getrawtransaction(tx_id, False, blockhash), self.nodes[2].gettransaction(tx_id)["hex"])
+            assert_equal(self.nodes[1].getrawtransaction(tx_id, False, blockhash), self.nodes[2].gettransaction(tx_id)["hex"])
+            assert_equal(self.nodes[0].getrawtransaction(tx_id, False, blockhash), tx.serialize_without_witness().hex())
 
         # Coinbase contains the witness commitment nonce, check that RPC shows us
         coinbase_txid = self.nodes[2].getblock(blockhash)['tx'][0]

test/functional/feature_taproot.py

@@ -95,6 +95,8 @@
 from test_framework.util import (
     assert_raises_rpc_error,
     assert_equal,
+    assert_greater_than_or_equal,
+    assert_not_equal,
     random_bytes,
 )
 from test_framework.wallet_util import generate_keypair
@@ -819,7 +821,7 @@ def spenders_taproot_active():
             for witlen in [20, 31, 32, 33]:
                 def mutate(spk):
                     prog = spk[2:]
-                    assert len(prog) == 32
+                    assert_equal(len(prog), 32)
                     if witlen < 32:
                         prog = prog[0:witlen]
                     elif witlen > 32:
@@ -1405,7 +1407,7 @@ def test_spenders(self, node, spenders, input_counts):
         self.log.info("- Running %i spending tests" % done)
         random.shuffle(normal_utxos)
         random.shuffle(mismatching_utxos)
-        assert done == len(normal_utxos) + len(mismatching_utxos)
+        assert_equal(done, len(normal_utxos) + len(mismatching_utxos))
 
         left = done
         while left:
@@ -1457,7 +1459,8 @@ def test_spenders(self, node, spenders, input_counts):
 
             # Add 1 to 4 random outputs (but constrained by inputs that require mismatching outputs)
             num_outputs = random.choice(range(1, 1 + min(4, 4 if first_mismatch_input is None else first_mismatch_input)))
-            assert in_value >= 0 and fee - num_outputs * DUST_LIMIT >= MIN_FEE
+            assert_greater_than_or_equal(in_value, 0)
+            assert_greater_than_or_equal(fee - num_outputs * DUST_LIMIT, MIN_FEE)
             for i in range(num_outputs):
                 tx.vout.append(CTxOut())
                 if in_value <= DUST_LIMIT:
@@ -1470,7 +1473,7 @@ def test_spenders(self, node, spenders, input_counts):
                 tx.vout[-1].scriptPubKey = random.choice(host_spks)
                 sigops_weight += CScript(tx.vout[-1].scriptPubKey).GetSigOpCount(False) * WITNESS_SCALE_FACTOR
             fee += in_value
-            assert fee >= 0
+            assert_greater_than_or_equal(fee, 0)
 
             # Select coinbase pubkey
             cb_pubkey = random.choice(host_pubkeys)
@@ -1518,9 +1521,9 @@ def test_spenders(self, node, spenders, input_counts):
             if (len(spenders) - left) // 200 > (len(spenders) - left - len(input_utxos)) // 200:
                 self.log.info("  - %i tests done" % (len(spenders) - left))
 
-        assert left == 0
-        assert len(normal_utxos) == 0
-        assert len(mismatching_utxos) == 0
+        assert_equal(left, 0)
+        assert_equal(len(normal_utxos), 0)
+        assert_equal(len(mismatching_utxos), 0)
         self.log.info("  - Done")
 
     def gen_test_vectors(self):
@@ -1536,7 +1539,7 @@ def gen_test_vectors(self):
         coinbase.vout = [CTxOut(5000000000, CScript([OP_1]))]
         coinbase.nLockTime = 0
         coinbase.rehash()
-        assert coinbase.hash == "f60c73405d499a956d3162e3483c395526ef78286458a4cb17b125aa92e49b20"
+        assert_equal(coinbase.hash, "f60c73405d499a956d3162e3483c395526ef78286458a4cb17b125aa92e49b20")
         # Mine it
         block = create_block(hashprev=int(self.nodes[0].getbestblockhash(), 16), coinbase=coinbase)
         block.rehash()
@@ -1572,20 +1575,20 @@ def gen_test_vectors(self):
         # Require negated taps[0]
         assert taps[0].negflag
         # Require one negated and one non-negated in taps 1 and 2.
-        assert taps[1].negflag != taps[2].negflag
+        assert_not_equal(taps[1].negflag, taps[2].negflag)
         # Require one negated and one non-negated in taps 3 and 4.
-        assert taps[3].negflag != taps[4].negflag
+        assert_not_equal(taps[3].negflag, taps[4].negflag)
         # Require one negated and one non-negated in taps 5 and 6.
-        assert taps[5].negflag != taps[6].negflag
+        assert_not_equal(taps[5].negflag, taps[6].negflag)
 
         cblks = [{leaf: get({**DEFAULT_CONTEXT, 'tap': taps[i], 'leaf': leaf}, 'controlblock') for leaf in taps[i].leaves} for i in range(7)]
         # Require one swapped and one unswapped in taps 3 and 4.
-        assert (cblks[3]['0'][33:65] < cblks[3]['1'][33:65]) != (cblks[4]['0'][33:65] < cblks[4]['1'][33:65])
+        assert_not_equal((cblks[3]['0'][33:65] < cblks[3]['1'][33:65]), (cblks[4]['0'][33:65] < cblks[4]['1'][33:65]))
         # Require one swapped and one unswapped in taps 5 and 6, both at the top and child level.
-        assert (cblks[5]['0'][33:65] < cblks[5]['1'][65:]) != (cblks[6]['0'][33:65] < cblks[6]['1'][65:])
-        assert (cblks[5]['1'][33:65] < cblks[5]['2'][33:65]) != (cblks[6]['1'][33:65] < cblks[6]['2'][33:65])
+        assert_not_equal((cblks[5]['0'][33:65] < cblks[5]['1'][65:]), (cblks[6]['0'][33:65] < cblks[6]['1'][65:]))
+        assert_not_equal((cblks[5]['1'][33:65] < cblks[5]['2'][33:65]), (cblks[6]['1'][33:65] < cblks[6]['2'][33:65]))
         # Require within taps 5 (and thus also 6) that one level is swapped and the other is not.
-        assert (cblks[5]['0'][33:65] < cblks[5]['1'][65:]) != (cblks[5]['1'][33:65] < cblks[5]['2'][33:65])
+        assert_not_equal((cblks[5]['0'][33:65] < cblks[5]['1'][65:]), (cblks[5]['1'][33:65] < cblks[5]['2'][33:65]))
 
         # Compute a deterministic set of scriptPubKeys
         tap_spks = []

test/functional/interface_usdt_utxocache.py

@@ -15,7 +15,10 @@
     pass
 from test_framework.messages import COIN
 from test_framework.test_framework import BitcoinTestFramework
-from test_framework.util import assert_equal
+from test_framework.util import (
+    assert_equal,
+    assert_greater_than,
+)
 from test_framework.wallet import MiniWallet
 
 utxocache_changes_program = """
@@ -354,8 +357,8 @@ def handle_utxocache_flush(_, data, __):
             })
             # sanity checks only
             try:
-                assert event.memory > 0
-                assert event.duration > 0
+                assert_greater_than(event.memory, 0)
+                assert_greater_than(event.duration, 0)
             except AssertionError:
                 self.log.exception("Assertion error")
             else: