scripted-diff: replace remaining Python test equality asserts
What changed, and why it matters
This commit is a bulk cleanup of Bitcoin Core's Python functional tests. It mechanically replaces plain `assert x == y` statements with a project-specific `assert_equal(x, y)` helper. The change only affects test code, not the Bitcoin node software that users run, so it cannot directly impact live Bitcoin operations, wallets, or consensus. It improves the quality of test failure messages but introduces no security vulnerability and fixes no exploitable bug.
No security action required. Treat as a normal code-quality/test-maintenance commit. Reviewers may optionally verify that the excluded files were correctly left untouched and that no runtime semantics changed (e.g., `assert_equal` is defined and raises AssertionError on mismatch, matching the prior behavior).
Security signals we found
No strong security signals were identified.
Evidence from the diff
The commit applies a scripted Perl substitution across 37 Python files under test/functional, converting remaining equality assertions from the built-in assert form to the test framework’s assert_equal() helper. The helper typically prints both compared values on failure, making test diagnostics clearer. The substitution excludes three files where a naive line-based rewrite would be unsafe (wallet_bumpfee.py, test_framework/netutil.py, test_framework/authproxy.py). All changes are syntactic and confined to the test suite; no production C++/Python node code, RPC behavior, consensus logic, or networking protocol is modified.
Changed components
test/functional/*.pytest/functional/test_framework/*.pyInspect captured patch +113 / −113
diff --git a/test/functional/data/invalid_txs.py b/test/functional/data/invalid_txs.py
index 7a769922..1d3a9399 100644
--- a/test/functional/data/invalid_txs.py
+++ b/test/functional/data/invalid_txs.py
@@ -123,8 +123,8 @@ class SizeTooSmall(BadTxTemplate):
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)
return tx
# reject a transaction that contains a witness
diff --git a/test/functional/feature_assumeutxo.py b/test/functional/feature_assumeutxo.py
index 5033cbb8..886d104e 100755
--- a/test/functional/feature_assumeutxo.py
+++ b/test/functional/feature_assumeutxo.py
@@ -413,7 +413,7 @@ class AssumeutxoTest(BitcoinTestFramework):
# Generate a series of blocks that `n0` will have in the snapshot,
# but that n1 and n2 don't yet see.
- assert n0.getblockcount() == START_HEIGHT
+ assert_equal(n0.getblockcount(), START_HEIGHT)
blocks = {START_HEIGHT: Block(n0.getbestblockhash(), 1, START_HEIGHT + 1)}
for i in range(100):
block_tx = 1
diff --git a/test/functional/feature_posix_fs_permissions.py b/test/functional/feature_posix_fs_permissions.py
index 5fc82c26..15087a9c 100755
--- a/test/functional/feature_posix_fs_permissions.py
+++ b/test/functional/feature_posix_fs_permissions.py
@@ -23,12 +23,12 @@ class PosixFsPermissionsTest(BitcoinTestFramework):
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)
diff --git a/test/functional/feature_pruning.py b/test/functional/feature_pruning.py
index ddc3fc59..c935ea95 100755
--- a/test/functional/feature_pruning.py
+++ b/test/functional/feature_pruning.py
@@ -195,7 +195,7 @@ class PruneTest(BitcoinTestFramework):
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
diff --git a/test/functional/feature_reindex_readonly.py b/test/functional/feature_reindex_readonly.py
index 6f088e4a..889d0c2a 100755
--- a/test/functional/feature_reindex_readonly.py
+++ b/test/functional/feature_reindex_readonly.py
@@ -78,7 +78,7 @@ class BlockstoreReindexTest(BitcoinTestFramework):
self.log.debug("Attempt to restart and reindex the node with the unwritable block file")
with self.nodes[0].assert_debug_log(["Reindexing finished"], timeout=60):
self.start_node(0, extra_args=['-reindex', '-fastprune'])
- assert block_count == self.nodes[0].getblockcount()
+ assert_equal(block_count, self.nodes[0].getblockcount())
undo_immutable()
filename.chmod(0o777)
diff --git a/test/functional/feature_taproot.py b/test/functional/feature_taproot.py
index 8891e0f9..b2d1d0f5 100755
--- a/test/functional/feature_taproot.py
+++ b/test/functional/feature_taproot.py
@@ -850,7 +850,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:
@@ -1532,7 +1532,7 @@ class TaprootTest(BitcoinTestFramework):
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:
@@ -1645,9 +1645,9 @@ class TaprootTest(BitcoinTestFramework):
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):
@@ -1662,7 +1662,7 @@ class TaprootTest(BitcoinTestFramework):
coinbase.vin = [CTxIn(COutPoint(0, 0xffffffff), CScript([OP_1, OP_1]), SEQUENCE_FINAL)]
coinbase.vout = [CTxOut(5000000000, CScript([OP_1]))]
coinbase.nLockTime = 0
- assert coinbase.txid_hex == "f60c73405d499a956d3162e3483c395526ef78286458a4cb17b125aa92e49b20"
+ assert_equal(coinbase.txid_hex, "f60c73405d499a956d3162e3483c395526ef78286458a4cb17b125aa92e49b20")
# Mine it
block = create_block(hashprev=int(self.nodes[0].getbestblockhash(), 16), coinbase=coinbase)
block.solve()
diff --git a/test/functional/mempool_cluster.py b/test/functional/mempool_cluster.py
index 26205290..0dfa6b09 100755
--- a/test/functional/mempool_cluster.py
+++ b/test/functional/mempool_cluster.py
@@ -72,7 +72,7 @@ class MempoolClusterTest(BitcoinTestFramework):
all_txids.append(next_tx["txid"])
utxo_to_spend = next_tx["new_utxo"]
- assert node.getmempoolcluster(parent_tx['txid'])['txcount'] == cluster_count
+ assert_equal(node.getmempoolcluster(parent_tx['txid'])['txcount'], cluster_count)
return all_results
def check_feerate_diagram(self, node):
diff --git a/test/functional/mining_getblocktemplate_longpoll.py b/test/functional/mining_getblocktemplate_longpoll.py
index db547971..e36396b3 100755
--- a/test/functional/mining_getblocktemplate_longpoll.py
+++ b/test/functional/mining_getblocktemplate_longpoll.py
@@ -37,7 +37,7 @@ class GetBlockTemplateLPTest(BitcoinTestFramework):
template = self.nodes[0].getblocktemplate({'rules': ['segwit']})
longpollid = template['longpollid']
template2 = self.nodes[0].getblocktemplate({'rules': ['segwit']})
- assert template2['longpollid'] == longpollid
+ assert_equal(template2['longpollid'], longpollid)
self.log.info("Test that longpoll waits if we do nothing")
thr = LongpollThread(self.nodes[0])
diff --git a/test/functional/p2p_addrfetch.py b/test/functional/p2p_addrfetch.py
index 49c131f1..fd781f0a 100755
--- a/test/functional/p2p_addrfetch.py
+++ b/test/functional/p2p_addrfetch.py
@@ -50,8 +50,8 @@ class P2PAddrFetch(BitcoinTestFramework):
self.log.info("Check that we send getaddr but don't try to sync headers with the addr-fetch peer")
peer.sync_with_ping()
with p2p_lock:
- assert peer.message_count['getaddr'] == 1
- assert peer.message_count['getheaders'] == 0
+ assert_equal(peer.message_count['getaddr'], 1)
+ assert_equal(peer.message_count['getheaders'], 0)
self.log.info("Check that answering the getaddr with a single address does not lead to disconnect")
# This prevents disconnecting on self-announcements
diff --git a/test/functional/p2p_blockfilters.py b/test/functional/p2p_blockfilters.py
index 32a3db7b..375a794b 100755
--- a/test/functional/p2p_blockfilters.py
+++ b/test/functional/p2p_blockfilters.py
@@ -71,11 +71,11 @@ class CompactFiltersTest(BitcoinTestFramework):
# Check that nodes have signalled NODE_COMPACT_FILTERS correctly.
assert_not_equal(peer_0.nServices & NODE_COMPACT_FILTERS, 0)
- assert peer_1.nServices & NODE_COMPACT_FILTERS == 0
+ assert_equal(peer_1.nServices & NODE_COMPACT_FILTERS, 0)
# Check that the localservices is as expected.
assert_not_equal(int(self.nodes[0].getnetworkinfo()['localservices'], 16) & NODE_COMPACT_FILTERS, 0)
- assert int(self.nodes[1].getnetworkinfo()['localservices'], 16) & NODE_COMPACT_FILTERS == 0
+ assert_equal(int(self.nodes[1].getnetworkinfo()['localservices'], 16) & NODE_COMPACT_FILTERS, 0)
self.log.info("get cfcheckpt on chain to be re-orged out.")
request = msg_getcfcheckpt(
diff --git a/test/functional/p2p_handshake.py b/test/functional/p2p_handshake.py
index a05c072b..b46c89ca 100755
--- a/test/functional/p2p_handshake.py
+++ b/test/functional/p2p_handshake.py
@@ -63,7 +63,7 @@ class P2PHandshakeTest(BitcoinTestFramework):
with node.assert_debug_log([expected_debug_log]):
self.add_outbound_connection(node, conn_type, services, wait_for_disconnect=True)
else:
- assert (services & desirable_service_flags) == desirable_service_flags
+ assert_equal((services & desirable_service_flags), desirable_service_flags)
self.add_outbound_connection(node, conn_type, services, wait_for_disconnect=False)
def generate_at_mocktime(self, time):
diff --git a/test/functional/p2p_headers_sync_with_minchainwork.py b/test/functional/p2p_headers_sync_with_minchainwork.py
index e90dccd4..1dc38faa 100755
--- a/test/functional/p2p_headers_sync_with_minchainwork.py
+++ b/test/functional/p2p_headers_sync_with_minchainwork.py
@@ -68,7 +68,7 @@ class RejectLowDifficultyHeadersTest(BitcoinTestFramework):
def check_node3_chaintips(num_tips, tip_hash, height):
node3_chaintips = self.nodes[3].getchaintips()
- assert len(node3_chaintips) == num_tips
+ assert_equal(len(node3_chaintips), num_tips)
assert {
'height': height,
'hash': tip_hash,
@@ -80,7 +80,7 @@ class RejectLowDifficultyHeadersTest(BitcoinTestFramework):
for node in self.nodes[1:3]:
chaintips = node.getchaintips()
- assert len(chaintips) == 1
+ assert_equal(len(chaintips), 1)
assert {
'height': 0,
'hash': '0f9188f13cb7b2c71f2a335e3a4fc328bf5beb436012afca590b1a11466e2206',
@@ -103,7 +103,7 @@ class RejectLowDifficultyHeadersTest(BitcoinTestFramework):
'status': 'active',
} in self.nodes[2].getchaintips()
- assert len(self.nodes[2].getchaintips()) == 1
+ assert_equal(len(self.nodes[2].getchaintips()), 1)
self.log.info("Check that node3 accepted these headers as well")
check_node3_chaintips(2, self.nodes[0].getbestblockhash(), NODE1_BLOCKS_REQUIRED)
diff --git a/test/functional/p2p_invalid_messages.py b/test/functional/p2p_invalid_messages.py
index ec0859f5..8521d673 100755
--- a/test/functional/p2p_invalid_messages.py
+++ b/test/functional/p2p_invalid_messages.py
@@ -331,7 +331,7 @@ class InvalidMessagesTest(BitcoinTestFramework):
conn = self.nodes[0].add_p2p_connection(P2PDataStore(), supports_v2_p2p=False)
conn2 = self.nodes[0].add_p2p_connection(P2PDataStore(), supports_v2_p2p=False)
msg_at_size = msg_unrecognized(str_data="b" * VALID_DATA_LIMIT)
- assert len(msg_at_size.serialize()) == MAX_PROTOCOL_MESSAGE_LENGTH
+ assert_equal(len(msg_at_size.serialize()), MAX_PROTOCOL_MESSAGE_LENGTH)
self.log.info("(a) Send 80 messages, each of maximum valid data size (4MB)")
for _ in range(80):
diff --git a/test/functional/p2p_segwit.py b/test/functional/p2p_segwit.py
index 85fe0412..29e9fc32 100755
--- a/test/functional/p2p_segwit.py
+++ b/test/functional/p2p_segwit.py
@@ -374,14 +374,14 @@ class SegWitTest(BitcoinTestFramework):
self.test_node.send_without_ping(msg_headers())
self.test_node.announce_block_and_wait_for_getdata(block1, use_header=False)
- assert self.test_node.last_message["getdata"].inv[0].type == blocktype
+ assert_equal(self.test_node.last_message["getdata"].inv[0].type, blocktype)
test_witness_block(self.nodes[0], self.test_node, block1, True)
block2 = self.build_next_block()
block2.solve()
self.test_node.announce_block_and_wait_for_getdata(block2, use_header=True)
- assert self.test_node.last_message["getdata"].inv[0].type == blocktype
+ assert_equal(self.test_node.last_message["getdata"].inv[0].type, blocktype)
test_witness_block(self.nodes[0], self.test_node, block2, True)
# Check that we can getdata for witness blocks or regular blocks,
@@ -410,8 +410,8 @@ class SegWitTest(BitcoinTestFramework):
block = self.build_next_block()
self.update_witness_block_with_transactions(block, [])
# This gives us a witness commitment.
- assert len(block.vtx[0].wit.vtxinwit) == 1
- assert len(block.vtx[0].wit.vtxinwit[0].scriptWitness.stack) == 1
+ assert_equal(len(block.vtx[0].wit.vtxinwit), 1)
+ assert_equal(len(block.vtx[0].wit.vtxinwit[0].scriptWitness.stack), 1)
test_witness_block(self.nodes[0], self.test_node, block, accepted=True)
# Now try to retrieve it...
rpc_block = self.nodes[0].getblock(block.hash_hex, False)
@@ -529,7 +529,7 @@ class SegWitTest(BitcoinTestFramework):
# Verify that if a peer doesn't set nServices to include NODE_WITNESS,
# the getdata is just for the non-witness portion.
self.old_node.announce_tx_and_wait_for_getdata(tx)
- assert self.old_node.last_message["getdata"].inv[0].type == MSG_TX
+ assert_equal(self.old_node.last_message["getdata"].inv[0].type, MSG_TX)
# Since we haven't delivered the tx yet, inv'ing the same tx from
# a witness transaction ought not result in a getdata.
@@ -831,7 +831,7 @@ class SegWitTest(BitcoinTestFramework):
assert block.get_weight() < MAX_BLOCK_WEIGHT
assert_equal(None, self.nodes[0].submitblock(block.serialize().hex()))
- assert self.nodes[0].getbestblockhash() == block.hash_hex
+ assert_equal(self.nodes[0].getbestblockhash(), block.hash_hex)
# Build a relayable-but-invalid block
relayable_block = self.build_next_block()
@@ -943,7 +943,7 @@ class SegWitTest(BitcoinTestFramework):
block.vtx[0].vout.pop()
add_witness_commitment(block)
block.solve()
- assert block.get_weight() == MAX_BLOCK_WEIGHT
+ assert_equal(block.get_weight(), MAX_BLOCK_WEIGHT)
test_witness_block(self.nodes[0], self.test_node, block, accepted=True)
@@ -1101,7 +1101,7 @@ class SegWitTest(BitcoinTestFramework):
# This script is 19 max pushes (9937 bytes), then 64 more opcode-bytes.
long_witness_script = CScript([b'a' * MAX_SCRIPT_ELEMENT_SIZE] * 19 + [OP_DROP] * 63 + [OP_TRUE])
- assert len(long_witness_script) == MAX_WITNESS_SCRIPT_LENGTH + 1
+ assert_equal(len(long_witness_script), MAX_WITNESS_SCRIPT_LENGTH + 1)
long_script_pubkey = script_to_p2wsh_script(long_witness_script)
block = self.build_next_block()
@@ -1123,7 +1123,7 @@ class SegWitTest(BitcoinTestFramework):
# Try again with one less byte in the witness script
witness_script = CScript([b'a' * MAX_SCRIPT_ELEMENT_SIZE] * 19 + [OP_DROP] * 62 + [OP_TRUE])
- assert len(witness_script) == MAX_WITNESS_SCRIPT_LENGTH
+ assert_equal(len(witness_script), MAX_WITNESS_SCRIPT_LENGTH)
script_pubkey = script_to_p2wsh_script(witness_script)
tx.vout[0] = CTxOut(tx.vout[0].nValue, script_pubkey)
@@ -1358,7 +1358,7 @@ class SegWitTest(BitcoinTestFramework):
temp_utxo.append(UTXO(tx.txid_int, 0, tx.vout[0].nValue))
self.generate(self.nodes[0], 1) # Mine all the transactions
- assert len(self.nodes[0].getrawmempool()) == 0
+ assert_equal(len(self.nodes[0].getrawmempool()), 0)
# Finally, verify that version 0 -> version 2 transactions
# are standard
diff --git a/test/functional/p2p_tx_download.py b/test/functional/p2p_tx_download.py
index f1a42f5a..a46b6a6b 100755
--- a/test/functional/p2p_tx_download.py
+++ b/test/functional/p2p_tx_download.py
@@ -123,7 +123,7 @@ class TxDownloadTest(BitcoinTestFramework):
# peer, plus
# * the first time it is re-requested from the outbound peer, plus
# * 2 seconds to avoid races
- assert self.nodes[1].getpeerinfo()[0]['inbound'] == False
+ assert_equal(self.nodes[1].getpeerinfo()[0]['inbound'], False)
timeout = 2 + NONPREF_PEER_TX_DELAY + GETDATA_TX_INTERVAL
self.log.info("Tx should be received at node 1 after {} seconds".format(timeout))
self.nodes[0].bumpmocktime(timeout)
diff --git a/test/functional/rpc_getdescriptoractivity.py b/test/functional/rpc_getdescriptoractivity.py
index 97599025..49db18ee 100755
--- a/test/functional/rpc_getdescriptoractivity.py
+++ b/test/functional/rpc_getdescriptoractivity.py
@@ -111,11 +111,11 @@ class GetBlocksActivityTest(BitcoinTestFramework):
[a1] = [a for a in result['activity'] if a['output_spk']['address'] == addr_1]
[a2] = [a for a in result['activity'] if a['output_spk']['address'] == addr_2]
- assert a1['blockhash'] == blockhash
- assert a1['amount'] == 1.0
+ assert_equal(a1['blockhash'], blockhash)
+ assert_equal(a1['amount'], 1.0)
- assert a2['blockhash'] == blockhash
- assert a2['amount'] == 2.0
+ assert_equal(a2['blockhash'], blockhash)
+ assert_equal(a2['amount'], 2.0)
def test_invalid_blockhash(self, node, wallet):
self.log.info("Test that passing an invalid blockhash raises appropriate RPC error")
@@ -159,8 +159,8 @@ class GetBlocksActivityTest(BitcoinTestFramework):
assert_equal(len(activity), 2)
[confirmed] = [a for a in activity if a.get('blockhash') == blockhash]
- assert confirmed['txid'] == txid_1
- assert confirmed['height'] == node.getblockchaininfo()['blocks']
+ assert_equal(confirmed['txid'], txid_1)
+ assert_equal(confirmed['height'], node.getblockchaininfo()['blocks'])
[unconfirmed] = [a for a in activity if not a.get('blockhash')]
assert 'blockhash' not in unconfirmed
@@ -185,15 +185,15 @@ class GetBlocksActivityTest(BitcoinTestFramework):
assert_equal(len(result['activity']), 4)
- assert result['activity'][1]['type'] == 'receive'
- assert result['activity'][1]['txid'] == sent1['txid']
- assert result['activity'][1]['blockhash'] == blockhash_1
+ assert_equal(result['activity'][1]['type'], 'receive')
+ assert_equal(result['activity'][1]['txid'], sent1['txid'])
+ assert_equal(result['activity'][1]['blockhash'], blockhash_1)
- assert result['activity'][2]['type'] == 'spend'
- assert result['activity'][2]['spend_txid'] == sent2['txid']
- assert result['activity'][2]['spend_vin'] == 0
- assert result['activity'][2]['prevout_txid'] == sent1['txid']
- assert result['activity'][2]['blockhash'] == blockhash_2
+ assert_equal(result['activity'][2]['type'], 'spend')
+ assert_equal(result['activity'][2]['spend_txid'], sent2['txid'])
+ assert_equal(result['activity'][2]['spend_vin'], 0)
+ assert_equal(result['activity'][2]['prevout_txid'], sent1['txid'])
+ assert_equal(result['activity'][2]['blockhash'], blockhash_2)
# Test that reversing the blockorder yields the same result.
assert_equal(result, node.getdescriptoractivity(
@@ -221,17 +221,17 @@ class GetBlocksActivityTest(BitcoinTestFramework):
a1 = result['activity'][0]
a2 = result['activity'][1]
- assert a1['type'] == "spend"
- assert a1['blockhash'] == blockhash
+ assert_equal(a1['type'], "spend")
+ assert_equal(a1['blockhash'], blockhash)
# sPK lacks address.
assert_equal(list(a1['prevout_spk'].keys()), ['asm', 'desc', 'hex', 'type'])
- assert a1['amount'] == no_addr_tx["fee"] + Decimal(no_addr_tx["tx"].vout[0].nValue) / COIN
+ assert_equal(a1['amount'], no_addr_tx["fee"] + Decimal(no_addr_tx["tx"].vout[0].nValue) / COIN)
- assert a2['type'] == "receive"
- assert a2['blockhash'] == blockhash
+ assert_equal(a2['type'], "receive")
+ assert_equal(a2['blockhash'], blockhash)
# sPK lacks address.
assert_equal(list(a2['output_spk'].keys()), ['asm', 'desc', 'hex', 'type'])
- assert a2['amount'] == Decimal(no_addr_tx["tx"].vout[0].nValue) / COIN
+ assert_equal(a2['amount'], Decimal(no_addr_tx["tx"].vout[0].nValue) / COIN)
def test_required_args(self, node):
self.log.info("Test that required arguments must be passed")
diff --git a/test/functional/rpc_help.py b/test/functional/rpc_help.py
index f8265c32..d76a6fa6 100755
--- a/test/functional/rpc_help.py
+++ b/test/functional/rpc_help.py
@@ -13,8 +13,8 @@ import re
def parse_string(s):
- assert s[0] == '"'
- assert s[-1] == '"'
+ assert_equal(s[0], '"')
+ assert_equal(s[-1], '"')
return s[1:-1]
def process_mapping(fname):
diff --git a/test/functional/rpc_net.py b/test/functional/rpc_net.py
index 24769e2b..405d21fc 100755
--- a/test/functional/rpc_net.py
+++ b/test/functional/rpc_net.py
@@ -38,7 +38,7 @@ def assert_net_servicesnames(servicesflag, servicenames):
servicesflag_generated = 0
for servicename in servicenames:
servicesflag_generated |= getattr(test_framework.messages, 'NODE_' + servicename)
- assert servicesflag_generated == servicesflag
+ assert_equal(servicesflag_generated, servicesflag)
def seed_addrman(node):
@@ -500,7 +500,7 @@ class NetTest(BitcoinTestFramework):
for bucket_position in getrawaddrman[table_name].keys():
entry = getrawaddrman[table_name][bucket_position]
expected_entry = list(filter(lambda e: e["address"] == entry["address"], table_info))[0]
- assert bucket_position == expected_entry["bucket_position"]
+ assert_equal(bucket_position, expected_entry["bucket_position"])
check_addr_information(entry, expected_entry)
# we expect 4 new and 4 tried table entries in the addrman which were added using seed_addrman()
diff --git a/test/functional/rpc_psbt.py b/test/functional/rpc_psbt.py
index 56c00514..6b1bc414 100755
--- a/test/functional/rpc_psbt.py
+++ b/test/functional/rpc_psbt.py
@@ -490,7 +490,7 @@ class PSBTTest(BitcoinTestFramework):
finalized_psbt = processed_finalized_psbt['psbt']
finalized_psbt_hex = processed_finalized_psbt['hex']
assert_not_equal(signed_psbt, finalized_psbt)
- assert finalized_psbt_hex == finalized_hex
+ assert_equal(finalized_psbt_hex, finalized_hex)
# Manually selected inputs can be locked:
assert_equal(len(self.nodes[0].listlockunspent()), 0)
diff --git a/test/functional/test_framework/address.py b/test/functional/test_framework/address.py
index f07019d2..d4aa26e5 100644
--- a/test/functional/test_framework/address.py
+++ b/test/functional/test_framework/address.py
@@ -100,12 +100,12 @@ def base58_to_byte(s):
def keyhash_to_p2pkh(hash, main=False):
- assert len(hash) == 20
+ assert_equal(len(hash), 20)
version = 0 if main else 111
return byte_to_base58(hash, version)
def scripthash_to_p2sh(hash, main=False):
- assert len(hash) == 20
+ assert_equal(len(hash), 20)
version = 5 if main else 196
return byte_to_base58(hash, version)
@@ -144,7 +144,7 @@ def script_to_p2sh_p2wsh(script, main=False):
return script_to_p2sh(p2shscript, main)
def output_key_to_p2tr(key, main=False):
- assert len(key) == 32
+ assert_equal(len(key), 32)
return program_to_witness(1, key, main)
def p2a(main=False):
diff --git a/test/functional/test_framework/crypto/ellswift.py b/test/functional/test_framework/crypto/ellswift.py
index db48dd24..e73da525 100644
--- a/test/functional/test_framework/crypto/ellswift.py
+++ b/test/functional/test_framework/crypto/ellswift.py
@@ -105,7 +105,7 @@ class TestFrameworkEllSwift(unittest.TestCase):
(FE(42), FE(0)), # t = 0
(FE(5), FE(-132).sqrt()), # u^3 + t^2 + 7 = 0
]
- assert undefined_inputs[-1][0]**3 + undefined_inputs[-1][1]**2 + 7 == 0
+ assert_equal(undefined_inputs[-1][0]**3 + undefined_inputs[-1][1]**2 + 7, 0)
for u, t in undefined_inputs:
x = xswiftec(u, t)
self.assertTrue(GE.is_valid_x(x))
@@ -155,7 +155,7 @@ class TestFrameworkEllSwift(unittest.TestCase):
reader = csv.DictReader(csvfile)
for row in reader:
encoding = bytes.fromhex(row['ellswift'])
- assert len(encoding) == 64
+ assert_equal(len(encoding), 64)
expected_x = FE(int(row['x'], 16))
u = FE(int.from_bytes(encoding[:32], 'big'))
t = FE(int.from_bytes(encoding[32:], 'big'))
diff --git a/test/functional/test_framework/crypto/secp256k1.py b/test/functional/test_framework/crypto/secp256k1.py
index b7193d46..e89f2018 100644
--- a/test/functional/test_framework/crypto/secp256k1.py
+++ b/test/functional/test_framework/crypto/secp256k1.py
@@ -179,7 +179,7 @@ class GE:
# Initialize as point on the curve (and check that it is).
fx = FE(x)
fy = FE(y)
- assert fy**2 == fx**3 + 7
+ assert_equal(fy**2, fx**3 + 7)
self.infinity = False
self.x = fx
self.y = fy
@@ -194,7 +194,7 @@ class GE:
if self.x == a.x:
if self.y != a.y:
# A point added to its own negation is infinity.
- assert self.y + a.y == 0
+ assert_equal(self.y + a.y, 0)
return GE()
else:
# For identical inputs, use the tangent (doubling formula).
@@ -292,7 +292,7 @@ class GE:
@staticmethod
def from_bytes_xonly(b):
"""Convert a point given in xonly encoding to a group element."""
- assert len(b) == 32
+ assert_equal(len(b), 32)
x = FE.from_bytes(b)
if x is None:
return None
diff --git a/test/functional/test_framework/key.py b/test/functional/test_framework/key.py
index 1e963dd3..796e6e35 100644
--- a/test/functional/test_framework/key.py
+++ b/test/functional/test_framework/key.py
@@ -132,7 +132,7 @@ class ECKey:
def set(self, secret, compressed):
"""Construct a private key object with given 32-byte secret and compressed flag."""
- assert len(secret) == 32
+ assert_equal(len(secret), 32)
secret = int.from_bytes(secret, 'big')
self.valid = (secret > 0 and secret < ORDER)
if self.valid:
@@ -194,7 +194,7 @@ def compute_xonly_pubkey(key):
This also returns whether the resulting public key was negated.
"""
- assert len(key) == 32
+ assert_equal(len(key), 32)
x = int.from_bytes(key, 'big')
if x == 0 or x >= ORDER:
return (None, None)
@@ -204,8 +204,8 @@ def compute_xonly_pubkey(key):
def tweak_add_privkey(key, tweak):
"""Tweak a private key (after negating it if needed)."""
- assert len(key) == 32
- assert len(tweak) == 32
+ assert_equal(len(key), 32)
+ assert_equal(len(tweak), 32)
x = int.from_bytes(key, 'big')
if x == 0 or x >= ORDER:
@@ -223,8 +223,8 @@ def tweak_add_privkey(key, tweak):
def tweak_add_pubkey(key, tweak):
"""Tweak a public key and return whether the result had to be negated."""
- assert len(key) == 32
- assert len(tweak) == 32
+ assert_equal(len(key), 32)
+ assert_equal(len(tweak), 32)
P = secp256k1.GE.from_bytes_xonly(key)
if P is None:
@@ -244,8 +244,8 @@ def verify_schnorr(key, sig, msg):
- sig is a 64-byte Schnorr signature
- msg is a variable-length message
"""
- assert len(key) == 32
- assert len(sig) == 64
+ assert_equal(len(key), 32)
+ assert_equal(len(sig), 64)
P = secp256k1.GE.from_bytes_xonly(key)
if P is None:
@@ -270,8 +270,8 @@ def sign_schnorr(key, msg, aux=None, flip_p=False, flip_r=False):
if aux is None:
aux = bytes(32)
- assert len(key) == 32
- assert len(aux) == 32
+ assert_equal(len(key), 32)
+ assert_equal(len(aux), 32)
sec = int.from_bytes(key, 'big')
if sec == 0 or sec >= ORDER:
diff --git a/test/functional/test_framework/messages.py b/test/functional/test_framework/messages.py
index f6dfc461..012049c2 100755
--- a/test/functional/test_framework/messages.py
+++ b/test/functional/test_framework/messages.py
@@ -284,7 +284,7 @@ def from_binary(cls, stream):
obj = cls()
obj.deserialize(stream)
if was_bytes:
- assert len(stream.read()) == 0
+ assert_equal(len(stream.read()), 0)
return obj
@@ -343,7 +343,7 @@ class CAddress:
def serialize(self, *, with_time=True):
"""Serialize in addrv1 format (pre-BIP155)"""
- assert self.net == self.NET_IPV4
+ assert_equal(self.net, self.NET_IPV4)
r = b""
if with_time:
# VERSION messages serialize CAddress objects without time
@@ -364,7 +364,7 @@ class CAddress:
assert self.net in self.ADDRV2_NET_NAME
address_length = deser_compact_size(f)
- assert address_length == self.ADDRV2_ADDRESS_LENGTH[self.net]
+ assert_equal(address_length, self.ADDRV2_ADDRESS_LENGTH[self.net])
addr_bytes = f.read(address_length)
if self.net == self.NET_IPV4:
diff --git a/test/functional/test_framework/psbt.py b/test/functional/test_framework/psbt.py
index 5790d634..0a57be31 100644
--- a/test/functional/test_framework/psbt.py
+++ b/test/functional/test_framework/psbt.py
@@ -111,7 +111,7 @@ class PSBT:
self.tx = None
def deserialize(self, f):
- assert f.read(5) == b"psbt\xff"
+ assert_equal(f.read(5), b"psbt\xff")
self.g = from_binary(PSBTMap, f)
assert PSBT_GLOBAL_UNSIGNED_TX in self.g.map
self.tx = from_binary(CTransaction, self.g.map[PSBT_GLOBAL_UNSIGNED_TX])
@@ -125,8 +125,8 @@ class PSBT:
assert isinstance(self.o, list) and all(isinstance(x, PSBTMap) for x in self.o)
assert PSBT_GLOBAL_UNSIGNED_TX in self.g.map
tx = from_binary(CTransaction, self.g.map[PSBT_GLOBAL_UNSIGNED_TX])
- assert len(tx.vin) == len(self.i)
- assert len(tx.vout) == len(self.o)
+ assert_equal(len(tx.vin), len(self.i))
+ assert_equal(len(tx.vout), len(self.o))
psbt = [x.serialize() for x in [self.g] + self.i + self.o]
return b"psbt\xff" + b"".join(psbt)
diff --git a/test/functional/test_framework/script.py b/test/functional/test_framework/script.py
index 0097998b..fb168928 100644
--- a/test/functional/test_framework/script.py
+++ b/test/functional/test_framework/script.py
@@ -109,7 +109,7 @@ class CScriptOp(int):
try:
return _opcode_instances[n]
except IndexError:
- assert len(_opcode_instances) == n
+ assert_equal(len(_opcode_instances), n)
_opcode_instances.append(super().__new__(cls, n))
return _opcode_instances[n]
@@ -854,7 +854,7 @@ def TaprootSignatureMsg(txTo, spent_utxos, hash_type, input_index=0, *, scriptpa
ss += TaggedHash("TapLeaf", bytes([leaf_ver]) + ser_string(leaf_script))
ss += bytes([0])
ss += codeseparator_pos.to_bytes(4, "little", signed=False)
- assert len(ss) == 175 - (in_type == SIGHASH_ANYONECANPAY) * 49 - (out_type != SIGHASH_ALL and out_type != SIGHASH_SINGLE) * 32 + (annex is not None) * 32 + scriptpath * 37
+ assert_equal(len(ss), 175 - (in_type == SIGHASH_ANYONECANPAY) * 49 - (out_type != SIGHASH_ALL and out_type != SIGHASH_SINGLE) * 32 + (annex is not None) * 32 + scriptpath * 37)
return ss
def TaprootSignatureHash(*args, **kwargs):
@@ -875,7 +875,7 @@ def taproot_tree_helper(scripts):
code = script[1]
if len(script) == 3:
version = script[2]
- assert version & 1 == 0
+ assert_equal(version & 1, 0)
assert isinstance(code, bytes)
h = TaggedHash("TapLeaf", bytes([version]) + ser_string(code))
if name is None:
diff --git a/test/functional/test_framework/script_util.py b/test/functional/test_framework/script_util.py
index 812d9bbd..02be3538 100755
--- a/test/functional/test_framework/script_util.py
+++ b/test/functional/test_framework/script_util.py
@@ -61,12 +61,12 @@ MAX_STD_P2SH_SIGOPS = 15
# least 5 bytes.
MIN_STANDARD_TX_NONWITNESS_SIZE = 65
MIN_PADDING = MIN_STANDARD_TX_NONWITNESS_SIZE - 10 - 41 - 9
-assert MIN_PADDING == 5
+assert_equal(MIN_PADDING, 5)
# This script cannot be spent, allowing dust output values under
# standardness checks
DUMMY_MIN_OP_RETURN_SCRIPT = CScript([OP_RETURN] + ([OP_0] * (MIN_PADDING - 1)))
-assert len(DUMMY_MIN_OP_RETURN_SCRIPT) == MIN_PADDING
+assert_equal(len(DUMMY_MIN_OP_RETURN_SCRIPT), MIN_PADDING)
PAY_TO_ANCHOR = CScript([OP_1, bytes.fromhex("4e73")])
ANCHOR_ADDRESS = "bcrt1pfeesnyr2tx"
@@ -86,12 +86,12 @@ def keys_to_multisig_script(keys, *, k=None):
def keyhash_to_p2pkh_script(hash):
- assert len(hash) == 20
+ assert_equal(len(hash), 20)
return CScript([OP_DUP, OP_HASH160, hash, OP_EQUALVERIFY, OP_CHECKSIG])
def scripthash_to_p2sh_script(hash):
- assert len(hash) == 20
+ assert_equal(len(hash), 20)
return CScript([OP_HASH160, hash, OP_EQUAL])
@@ -147,7 +147,7 @@ def bulk_vout(tx, target_vsize):
assert_equal(tx.get_vsize(), target_vsize)
def output_key_to_p2tr_script(key):
- assert len(key) == 32
+ assert_equal(len(key), 32)
return program_to_witness_script(1, key)
diff --git a/test/functional/test_framework/test_node.py b/test/functional/test_framework/test_node.py
index 541c758b..bf0250f7 100755
--- a/test/functional/test_framework/test_node.py
+++ b/test/functional/test_framework/test_node.py
@@ -226,7 +226,7 @@ class TestNode():
def get_deterministic_priv_key(self):
"""Return a deterministic priv key in base58, that only depends on the node's index"""
- assert len(self.PRIV_KEYS) == MAX_NODES
+ assert_equal(len(self.PRIV_KEYS), MAX_NODES)
return self.PRIV_KEYS[self.index]
def _node_msg(self, msg: str) -> str:
diff --git a/test/functional/tool_wallet.py b/test/functional/tool_wallet.py
index ecef5e13..737aaf54 100755
--- a/test/functional/tool_wallet.py
+++ b/test/functional/tool_wallet.py
@@ -93,7 +93,7 @@ class ToolWalletTest(BitcoinTestFramework):
def assert_is_sqlite(self, filename):
with open(filename, 'rb') as f:
file_magic = f.read(16)
- assert file_magic == b'SQLite format 3\x00'
+ assert_equal(file_magic, b'SQLite format 3\x00')
def write_dump(self, dump, filename, magic=None, skip_checksum=False):
if magic is None:
diff --git a/test/functional/wallet_address_types.py b/test/functional/wallet_address_types.py
index 4c6e6ad7..2ba8eb9d 100755
--- a/test/functional/wallet_address_types.py
+++ b/test/functional/wallet_address_types.py
@@ -176,12 +176,12 @@ class AddressTypeTest(BitcoinTestFramework):
# Verify the descriptor checksum against the Python implementation
assert descsum_check(info['desc'])
# Verify that stripping the checksum and recreating it using Python roundtrips
- assert info['desc'] == descsum_create(info['desc'][:-9])
+ assert_equal(info['desc'], descsum_create(info['desc'][:-9]))
# Verify that stripping the checksum and feeding it to getdescriptorinfo roundtrips
- assert info['desc'] == self.nodes[0].getdescriptorinfo(info['desc'][:-9])['descriptor']
+ assert_equal(info['desc'], self.nodes[0].getdescriptorinfo(info['desc'][:-9])['descriptor'])
assert_equal(info['desc'][-8:], self.nodes[0].getdescriptorinfo(info['desc'][:-9])['checksum'])
# Verify that keeping the checksum and feeding it to getdescriptorinfo roundtrips
- assert info['desc'] == self.nodes[0].getdescriptorinfo(info['desc'])['descriptor']
+ assert_equal(info['desc'], self.nodes[0].getdescriptorinfo(info['desc'])['descriptor'])
assert_equal(info['desc'][-8:], self.nodes[0].getdescriptorinfo(info['desc'])['checksum'])
if not multisig and typ == 'legacy':
diff --git a/test/functional/wallet_backwards_compatibility.py b/test/functional/wallet_backwards_compatibility.py
index 13c3ef74..f00f2958 100755
--- a/test/functional/wallet_backwards_compatibility.py
+++ b/test/functional/wallet_backwards_compatibility.py
@@ -240,8 +240,8 @@ class BackwardsCompatibilityTest(BitcoinTestFramework):
node_master.createwallet(wallet_name="w2", disable_private_keys=True)
wallet = node_master.get_wallet_rpc("w2")
info = wallet.getwalletinfo()
- assert info['private_keys_enabled'] == False
- assert info['keypoolsize'] == 0
+ assert_equal(info['private_keys_enabled'], False)
+ assert_equal(info['keypoolsize'], 0)
# w3: blank wallet, created on master: update this
# test when default blank wallets can no longer be opened by older versions.
@@ -249,7 +249,7 @@ class BackwardsCompatibilityTest(BitcoinTestFramework):
wallet = node_master.get_wallet_rpc("w3")
info = wallet.getwalletinfo()
assert info['private_keys_enabled']
- assert info['keypoolsize'] == 0
+ assert_equal(info['keypoolsize'], 0)
# Unload wallets and copy to older nodes:
node_master_wallets_dir = node_master.wallets_path
@@ -279,7 +279,7 @@ class BackwardsCompatibilityTest(BitcoinTestFramework):
wallet = n.get_wallet_rpc(wallet_name)
info = wallet.getwalletinfo()
if wallet_name == "w1":
- assert info['private_keys_enabled'] == True
+ assert_equal(info['private_keys_enabled'], True)
assert info['keypoolsize'] > 0
txs = wallet.listtransactions()
assert_equal(len(txs), 5)
@@ -294,11 +294,11 @@ class BackwardsCompatibilityTest(BitcoinTestFramework):
assert_equal(txs[3]["replaced_by_txid"], tx4_id)
assert not hasattr(txs[3], "blockindex")
elif wallet_name == "w2":
- assert info['private_keys_enabled'] == False
- assert info['keypoolsize'] == 0
+ assert_equal(info['private_keys_enabled'], False)
+ assert_equal(info['keypoolsize'], 0)
else:
- assert info['private_keys_enabled'] == True
- assert info['keypoolsize'] == 0
+ assert_equal(info['private_keys_enabled'], True)
+ assert_equal(info['keypoolsize'], 0)
# Copy back to master
wallet.unloadwallet()
diff --git a/test/functional/wallet_disable.py b/test/functional/wallet_disable.py
index 46b92ad5..17b327e1 100755
--- a/test/functional/wallet_disable.py
+++ b/test/functional/wallet_disable.py
@@ -22,9 +22,9 @@ class DisableWalletTest (BitcoinTestFramework):
# Make sure wallet is really disabled
assert_raises_rpc_error(-32601, 'Method not found', self.nodes[0].getwalletinfo)
x = self.nodes[0].validateaddress('3J98t1WpEZ73CNmQviecrnyiWrnqRhWNLy')
- assert x['isvalid'] == False
+ assert_equal(x['isvalid'], False)
x = self.nodes[0].validateaddress('mneYUmWYsuk7kySiURxCi3AGxrAqZxLgPZ')
- assert x['isvalid'] == True
+ assert_equal(x['isvalid'], True)
if __name__ == '__main__':
diff --git a/test/functional/wallet_fundrawtransaction.py b/test/functional/wallet_fundrawtransaction.py
index 0d435a35..a00e7e68 100755
--- a/test/functional/wallet_fundrawtransaction.py
+++ b/test/functional/wallet_fundrawtransaction.py
@@ -785,7 +785,7 @@ class RawTransactionsTest(BitcoinTestFramework):
result = wwatch.fundrawtransaction(rawtx, changeAddress=w3.getrawchangeaddress(), subtractFeeFromOutputs=[0])
res_dec = self.nodes[0].decoderawtransaction(result["hex"])
assert_equal(len(res_dec["vin"]), 1)
- assert res_dec["vin"][0]["txid"] == self.watchonly_utxo['txid']
+ assert_equal(res_dec["vin"][0]["txid"], self.watchonly_utxo['txid'])
assert_greater_than(result["fee"], 0)
assert_equal(result["changepos"], -1)
@@ -1481,7 +1481,7 @@ class RawTransactionsTest(BitcoinTestFramework):
assert_equal(len(tx1_inputs), 1)
utxo1 = tx1_inputs[0]
- assert unconfirmed_txid == utxo1['txid']
+ assert_equal(unconfirmed_txid, utxo1['txid'])
final_tx1 = wallet.signrawtransactionwithwallet(funded_tx1)['hex']
txid1 = self.nodes[0].sendrawtransaction(final_tx1)
diff --git a/test/functional/wallet_keypool.py b/test/functional/wallet_keypool.py
index a935b854..78c760db 100755
--- a/test/functional/wallet_keypool.py
+++ b/test/functional/wallet_keypool.py
@@ -110,7 +110,7 @@ class KeyPoolTest(BitcoinTestFramework):
addr.add(nodes[0].getnewaddress(address_type="bech32"))
addr.add(nodes[0].getnewaddress(address_type="bech32"))
addr.add(nodes[0].getnewaddress(address_type="bech32"))
- assert len(addr) == 6
+ assert_equal(len(addr), 6)
# remember keypool sizes
wi = nodes[0].getwalletinfo()
kp_size_before = [wi['keypoolsize_hd_internal'], wi['keypoolsize']]
diff --git a/test/functional/wallet_listdescriptors.py b/test/functional/wallet_listdescriptors.py
index 8fae9290..98938f1b 100755
--- a/test/functional/wallet_listdescriptors.py
+++ b/test/functional/wallet_listdescriptors.py
@@ -46,7 +46,7 @@ class ListDescriptorsTest(BitcoinTestFramework):
assert_equal(4, len([d for d in result['descriptors'] if d['internal']]))
for item in result['descriptors']:
assert_not_equal(item['desc'], '')
- assert item['next_index'] == 0
+ assert_equal(item['next_index'], 0)
assert_equal(item['range'], [0, 0])
assert item['timestamp'] is not None
diff --git a/test/functional/wallet_miniscript.py b/test/functional/wallet_miniscript.py
index f3e5a0f8..b6b45296 100755
--- a/test/functional/wallet_miniscript.py
+++ b/test/functional/wallet_miniscript.py
@@ -301,7 +301,7 @@ class WalletMiniscriptTest(BitcoinTestFramework):
res = self.ms_sig_wallet.walletprocesspsbt(psbt=psbt, finalize=False)
psbtin = self.nodes[0].decodepsbt(res["psbt"])["inputs"][0]
sigs_field_name = "taproot_script_path_sigs" if is_taproot else "partial_signatures"
- assert len(psbtin[sigs_field_name]) == sigs_count
+ assert_equal(len(psbtin[sigs_field_name]), sigs_count)
res = self.ms_sig_wallet.finalizepsbt(res["psbt"])
assert_equal(res["complete"], stack_size is not None)
diff --git a/test/functional/wallet_txn_doublespend.py b/test/functional/wallet_txn_doublespend.py
index 4b61fa89..8a0bf28b 100755
--- a/test/functional/wallet_txn_doublespend.py
+++ b/test/functional/wallet_txn_doublespend.py
@@ -43,7 +43,7 @@ class TxnMallTest(BitcoinTestFramework):
# blockchain sync later in the test when nodes are connected, due to
# timing issues.
for n in self.nodes:
- assert n.getblockchaininfo()["initialblockdownload"] == False
+ assert_equal(n.getblockchaininfo()["initialblockdownload"], False)
for i in range(3):
assert_equal(self.nodes[i].getbalance(), starting_balance)
Why this scored 15/100
Community notes
Notes can correct, qualify, or add evidence to the AI analysis. Every note shown here has been validated by a human moderator.
The AI analysis stands alone for now. Submit a note if you can add evidence or important context.