test: introduce `tx_from_hex` helper for tx deserialization

`FromHex` is mostly used for transactions, so we introduce a
shortcut `tx_from_hex` for `FromHex(CTransaction, hex_str)`.

21 files changed, 255 insertions, 150 deletions

diff --git a/test/functional/feature_bip68_sequence.py b/test/functional/feature_bip68_sequence.py
index 6c5857c5ce..2b2341fb03 100755
--- a/test/functional/feature_bip68_sequence.py
+++ b/test/functional/feature_bip68_sequence.py
@@ -6,8 +6,20 @@
 
 import time
 
-from test_framework.blocktools import create_block, NORMAL_GBT_REQUEST_PARAMS, add_witness_commitment
-from test_framework.messages import COIN, COutPoint, CTransaction, CTxIn, CTxOut, FromHex, ToHex
+from test_framework.blocktools import (
+    NORMAL_GBT_REQUEST_PARAMS,
+    add_witness_commitment,
+    create_block,
+)
+from test_framework.messages import (
+    COIN,
+    COutPoint,
+    CTransaction,
+    CTxIn,
+    CTxOut,
+    ToHex,
+    tx_from_hex,
+)
 from test_framework.test_framework import BitcoinTestFramework
 from test_framework.util import (
     assert_equal,
@@ -215,7 +227,7 @@ class BIP68Test(BitcoinTestFramework):
 
         # Create a mempool tx.
         txid = self.nodes[0].sendtoaddress(self.nodes[0].getnewaddress(), 2)
-        tx1 = FromHex(CTransaction(), self.nodes[0].getrawtransaction(txid))
+        tx1 = tx_from_hex(self.nodes[0].getrawtransaction(txid))
         tx1.rehash()
 
         # Anyone-can-spend mempool tx.
@@ -225,7 +237,7 @@ class BIP68Test(BitcoinTestFramework):
         tx2.vin = [CTxIn(COutPoint(tx1.sha256, 0), nSequence=0)]
         tx2.vout = [CTxOut(int(tx1.vout[0].nValue - self.relayfee*COIN), DUMMY_P2WPKH_SCRIPT)]
         tx2_raw = self.nodes[0].signrawtransactionwithwallet(ToHex(tx2))["hex"]
-        tx2 = FromHex(tx2, tx2_raw)
+        tx2 = tx_from_hex(tx2_raw)
         tx2.rehash()
 
         self.nodes[0].sendrawtransaction(tx2_raw)
@@ -348,7 +360,7 @@ class BIP68Test(BitcoinTestFramework):
         assert not softfork_active(self.nodes[0], 'csv')
         txid = self.nodes[0].sendtoaddress(self.nodes[0].getnewaddress(), 2)
 
-        tx1 = FromHex(CTransaction(), self.nodes[0].getrawtransaction(txid))
+        tx1 = tx_from_hex(self.nodes[0].getrawtransaction(txid))
         tx1.rehash()
 
         # Make an anyone-can-spend transaction
@@ -359,7 +371,7 @@ class BIP68Test(BitcoinTestFramework):
 
         # sign tx2
         tx2_raw = self.nodes[0].signrawtransactionwithwallet(ToHex(tx2))["hex"]
-        tx2 = FromHex(tx2, tx2_raw)
+        tx2 = tx_from_hex(tx2_raw)
         tx2.rehash()
 
         self.nodes[0].sendrawtransaction(ToHex(tx2))
@@ -404,7 +416,7 @@ class BIP68Test(BitcoinTestFramework):
         outputs = { self.nodes[1].getnewaddress() : 1.0 }
         rawtx = self.nodes[1].createrawtransaction(inputs, outputs)
         rawtxfund = self.nodes[1].fundrawtransaction(rawtx)['hex']
-        tx = FromHex(CTransaction(), rawtxfund)
+        tx = tx_from_hex(rawtxfund)
         tx.nVersion = 2
         tx_signed = self.nodes[1].signrawtransactionwithwallet(ToHex(tx))["hex"]
         self.nodes[1].sendrawtransaction(tx_signed)
diff --git a/test/functional/feature_segwit.py b/test/functional/feature_segwit.py
index ad8767556b..0f62701305 100755
--- a/test/functional/feature_segwit.py
+++ b/test/functional/feature_segwit.py
@@ -5,7 +5,6 @@
 """Test the SegWit changeover logic."""
 
 from decimal import Decimal
-from io import BytesIO
 
 from test_framework.address import (
     key_to_p2pkh,
@@ -14,9 +13,35 @@ from test_framework.address import (
     script_to_p2sh_p2wsh,
     script_to_p2wsh,
 )
-from test_framework.blocktools import witness_script, send_to_witness
-from test_framework.messages import COIN, COutPoint, CTransaction, CTxIn, CTxOut, FromHex, sha256, ToHex
-from test_framework.script import CScript, OP_HASH160, OP_CHECKSIG, OP_0, hash160, OP_EQUAL, OP_DUP, OP_EQUALVERIFY, OP_1, OP_2, OP_CHECKMULTISIG, OP_TRUE, OP_DROP
+from test_framework.blocktools import (
+    send_to_witness,
+    witness_script,
+)
+from test_framework.messages import (
+    COIN,
+    COutPoint,
+    CTransaction,
+    CTxIn,
+    CTxOut,
+    ToHex,
+    sha256,
+    tx_from_hex,
+)
+from test_framework.script import (
+    CScript,
+    OP_0,
+    OP_1,
+    OP_2,
+    OP_CHECKMULTISIG,
+    OP_CHECKSIG,
+    OP_DROP,
+    OP_DUP,
+    OP_EQUAL,
+    OP_EQUALVERIFY,
+    OP_HASH160,
+    OP_TRUE,
+    hash160,
+)
 from test_framework.test_framework import BitcoinTestFramework
 from test_framework.util import (
     assert_equal,
@@ -179,7 +204,7 @@ class SegWitTest(BitcoinTestFramework):
         assert self.nodes[1].getblock(blockhash, False) == self.nodes[2].getblock(blockhash, False)
 
         for tx_id in segwit_tx_list:
-            tx = FromHex(CTransaction(), self.nodes[2].gettransaction(tx_id)["hex"])
+            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"]
@@ -225,12 +250,12 @@ class SegWitTest(BitcoinTestFramework):
         # tx1 is allowed to appear in the block, but no others.
         txid1 = send_to_witness(1, self.nodes[0], find_spendable_utxo(self.nodes[0], 50), self.pubkey[0], False, Decimal("49.996"))
         hex_tx = self.nodes[0].gettransaction(txid)['hex']
-        tx = FromHex(CTransaction(), hex_tx)
+        tx = tx_from_hex(hex_tx)
         assert tx.wit.is_null()  # This should not be a segwit input
         assert txid1 in self.nodes[0].getrawmempool()
 
         tx1_hex = self.nodes[0].gettransaction(txid1)['hex']
-        tx1 = FromHex(CTransaction(), tx1_hex)
+        tx1 = tx_from_hex(tx1_hex)
 
         # Check that wtxid is properly reported in mempool entry (txid1)
         assert_equal(int(self.nodes[0].getmempoolentry(txid1)["wtxid"], 16), tx1.calc_sha256(True))
@@ -245,7 +270,7 @@ class SegWitTest(BitcoinTestFramework):
         tx.vout.append(CTxOut(int(49.99 * COIN), CScript([OP_TRUE, OP_DROP] * 15 + [OP_TRUE])))
         tx2_hex = self.nodes[0].signrawtransactionwithwallet(ToHex(tx))['hex']
         txid2 = self.nodes[0].sendrawtransaction(tx2_hex)
-        tx = FromHex(CTransaction(), tx2_hex)
+        tx = tx_from_hex(tx2_hex)
         assert not tx.wit.is_null()
 
         # Check that wtxid is properly reported in mempool entry (txid2)
@@ -611,10 +636,8 @@ class SegWitTest(BitcoinTestFramework):
     def create_and_mine_tx_from_txids(self, txids, success=True):
         tx = CTransaction()
         for i in txids:
-            txtmp = CTransaction()
             txraw = self.nodes[0].getrawtransaction(i, 0, txs_mined[i])
-            f = BytesIO(hex_str_to_bytes(txraw))
-            txtmp.deserialize(f)
+            txtmp = tx_from_hex(txraw)
             for j in range(len(txtmp.vout)):
                 tx.vin.append(CTxIn(COutPoint(int('0x' + i, 0), j)))
         tx.vout.append(CTxOut(0, CScript()))
diff --git a/test/functional/interface_zmq.py b/test/functional/interface_zmq.py
index 94e162b748..0534618d06 100755
--- a/test/functional/interface_zmq.py
+++ b/test/functional/interface_zmq.py
@@ -8,7 +8,11 @@ import struct
 from test_framework.address import ADDRESS_BCRT1_UNSPENDABLE, ADDRESS_BCRT1_P2WSH_OP_TRUE
 from test_framework.blocktools import create_block, create_coinbase, add_witness_commitment
 from test_framework.test_framework import BitcoinTestFramework
-from test_framework.messages import CTransaction, hash256, FromHex
+from test_framework.messages import (
+    CTransaction,
+    hash256,
+    tx_from_hex,
+)
 from test_framework.util import (
     assert_equal,
     assert_raises_rpc_error,
@@ -393,10 +397,10 @@ class ZMQTest (BitcoinTestFramework):
             bump_info = self.nodes[0].bumpfee(orig_txid)
             # Mine the pre-bump tx
             block = create_block(int(self.nodes[0].getbestblockhash(), 16), create_coinbase(self.nodes[0].getblockcount()+1))
-            tx = FromHex(CTransaction(), raw_tx)
+            tx = tx_from_hex(raw_tx)
             block.vtx.append(tx)
             for txid in more_tx:
-                tx = FromHex(CTransaction(), self.nodes[0].getrawtransaction(txid))
+                tx = tx_from_hex(self.nodes[0].getrawtransaction(txid))
                 block.vtx.append(tx)
             add_witness_commitment(block)
             block.solve()
diff --git a/test/functional/mempool_accept.py b/test/functional/mempool_accept.py
index 12aac3ab65..1705d957aa 100755
--- a/test/functional/mempool_accept.py
+++ b/test/functional/mempool_accept.py
@@ -5,7 +5,6 @@
 """Test mempool acceptance of raw transactions."""
 
 from decimal import Decimal
-from io import BytesIO
 import math
 
 from test_framework.test_framework import BitcoinTestFramework
@@ -14,10 +13,10 @@ from test_framework.messages import (
     BIP125_SEQUENCE_NUMBER,
     COIN,
     COutPoint,
-    CTransaction,
     CTxOut,
     MAX_BLOCK_BASE_SIZE,
     MAX_MONEY,
+    tx_from_hex,
 )
 from test_framework.script import (
     hash160,
@@ -33,7 +32,6 @@ from test_framework.script import (
 from test_framework.util import (
     assert_equal,
     assert_raises_rpc_error,
-    hex_str_to_bytes,
 )
 
 
@@ -91,8 +89,7 @@ class MempoolAcceptanceTest(BitcoinTestFramework):
             inputs=[{"txid": txid_in_block, "vout": 0, "sequence": BIP125_SEQUENCE_NUMBER}],  # RBF is used later
             outputs=[{node.getnewaddress(): Decimal('0.3') - fee}],
         ))['hex']
-        tx = CTransaction()
-        tx.deserialize(BytesIO(hex_str_to_bytes(raw_tx_0)))
+        tx = tx_from_hex(raw_tx_0)
         txid_0 = tx.rehash()
         self.check_mempool_result(
             result_expected=[{'txid': txid_0, 'allowed': True, 'vsize': tx.get_vsize(), 'fees': {'base': fee}}],
@@ -107,7 +104,7 @@ class MempoolAcceptanceTest(BitcoinTestFramework):
             outputs=[{node.getnewaddress(): output_amount}],
             locktime=node.getblockcount() + 2000,  # Can be anything
         ))['hex']
-        tx.deserialize(BytesIO(hex_str_to_bytes(raw_tx_final)))
+        tx = tx_from_hex(raw_tx_final)
         fee_expected = coin['amount'] - output_amount
         self.check_mempool_result(
             result_expected=[{'txid': tx.rehash(), 'allowed': True, 'vsize': tx.get_vsize(), 'fees': {'base': fee_expected}}],
@@ -126,11 +123,11 @@ class MempoolAcceptanceTest(BitcoinTestFramework):
         )
 
         self.log.info('A transaction that replaces a mempool transaction')
-        tx.deserialize(BytesIO(hex_str_to_bytes(raw_tx_0)))
+        tx = tx_from_hex(raw_tx_0)
         tx.vout[0].nValue -= int(fee * COIN)  # Double the fee
         tx.vin[0].nSequence = BIP125_SEQUENCE_NUMBER + 1  # Now, opt out of RBF
         raw_tx_0 = node.signrawtransactionwithwallet(tx.serialize().hex())['hex']
-        tx.deserialize(BytesIO(hex_str_to_bytes(raw_tx_0)))
+        tx = tx_from_hex(raw_tx_0)
         txid_0 = tx.rehash()
         self.check_mempool_result(
             result_expected=[{'txid': txid_0, 'allowed': True, 'vsize': tx.get_vsize(), 'fees': {'base': (2 * fee)}}],
@@ -141,7 +138,7 @@ class MempoolAcceptanceTest(BitcoinTestFramework):
         # Send the transaction that replaces the mempool transaction and opts out of replaceability
         node.sendrawtransaction(hexstring=tx.serialize().hex(), maxfeerate=0)
         # take original raw_tx_0
-        tx.deserialize(BytesIO(hex_str_to_bytes(raw_tx_0)))
+        tx = tx_from_hex(raw_tx_0)
         tx.vout[0].nValue -= int(4 * fee * COIN)  # Set more fee
         # skip re-signing the tx
         self.check_mempool_result(
@@ -151,7 +148,7 @@ class MempoolAcceptanceTest(BitcoinTestFramework):
         )
 
         self.log.info('A transaction with missing inputs, that never existed')
-        tx.deserialize(BytesIO(hex_str_to_bytes(raw_tx_0)))
+        tx = tx_from_hex(raw_tx_0)
         tx.vin[0].prevout = COutPoint(hash=int('ff' * 32, 16), n=14)
         # skip re-signing the tx
         self.check_mempool_result(
@@ -160,7 +157,7 @@ class MempoolAcceptanceTest(BitcoinTestFramework):
         )
 
         self.log.info('A transaction with missing inputs, that existed once in the past')
-        tx.deserialize(BytesIO(hex_str_to_bytes(raw_tx_0)))
+        tx = tx_from_hex(raw_tx_0)
         tx.vin[0].prevout.n = 1  # Set vout to 1, to spend the other outpoint (49 coins) of the in-chain-tx we want to double spend
         raw_tx_1 = node.signrawtransactionwithwallet(tx.serialize().hex())['hex']
         txid_1 = node.sendrawtransaction(hexstring=raw_tx_1, maxfeerate=0)
@@ -190,7 +187,7 @@ class MempoolAcceptanceTest(BitcoinTestFramework):
             inputs=[{'txid': txid_spend_both, 'vout': 0}],
             outputs=[{node.getnewaddress(): 0.05}],
         ))['hex']
-        tx.deserialize(BytesIO(hex_str_to_bytes(raw_tx_reference)))
+        tx = tx_from_hex(raw_tx_reference)
         # Reference tx should be valid on itself
         self.check_mempool_result(
             result_expected=[{'txid': tx.rehash(), 'allowed': True, 'vsize': tx.get_vsize(), 'fees': { 'base': Decimal('0.1') - Decimal('0.05')}}],
@@ -199,17 +196,17 @@ class MempoolAcceptanceTest(BitcoinTestFramework):
         )
 
         self.log.info('A transaction with no outputs')
-        tx.deserialize(BytesIO(hex_str_to_bytes(raw_tx_reference)))
+        tx = tx_from_hex(raw_tx_reference)
         tx.vout = []
         # Skip re-signing the transaction for context independent checks from now on
-        # tx.deserialize(BytesIO(hex_str_to_bytes(node.signrawtransactionwithwallet(tx.serialize().hex())['hex'])))
+        # tx = tx_from_hex(node.signrawtransactionwithwallet(tx.serialize().hex())['hex'])
         self.check_mempool_result(
             result_expected=[{'txid': tx.rehash(), 'allowed': False, 'reject-reason': 'bad-txns-vout-empty'}],
             rawtxs=[tx.serialize().hex()],
         )
 
         self.log.info('A really large transaction')
-        tx.deserialize(BytesIO(hex_str_to_bytes(raw_tx_reference)))
+        tx = tx_from_hex(raw_tx_reference)
         tx.vin = [tx.vin[0]] * math.ceil(MAX_BLOCK_BASE_SIZE / len(tx.vin[0].serialize()))
         self.check_mempool_result(
             result_expected=[{'txid': tx.rehash(), 'allowed': False, 'reject-reason': 'bad-txns-oversize'}],
@@ -217,7 +214,7 @@ class MempoolAcceptanceTest(BitcoinTestFramework):
         )
 
         self.log.info('A transaction with negative output value')
-        tx.deserialize(BytesIO(hex_str_to_bytes(raw_tx_reference)))
+        tx = tx_from_hex(raw_tx_reference)
         tx.vout[0].nValue *= -1
         self.check_mempool_result(
             result_expected=[{'txid': tx.rehash(), 'allowed': False, 'reject-reason': 'bad-txns-vout-negative'}],
@@ -226,7 +223,7 @@ class MempoolAcceptanceTest(BitcoinTestFramework):
 
         # The following two validations prevent overflow of the output amounts (see CVE-2010-5139).
         self.log.info('A transaction with too large output value')
-        tx.deserialize(BytesIO(hex_str_to_bytes(raw_tx_reference)))
+        tx = tx_from_hex(raw_tx_reference)
         tx.vout[0].nValue = MAX_MONEY + 1
         self.check_mempool_result(
             result_expected=[{'txid': tx.rehash(), 'allowed': False, 'reject-reason': 'bad-txns-vout-toolarge'}],
@@ -234,7 +231,7 @@ class MempoolAcceptanceTest(BitcoinTestFramework):
         )
 
         self.log.info('A transaction with too large sum of output values')
-        tx.deserialize(BytesIO(hex_str_to_bytes(raw_tx_reference)))
+        tx = tx_from_hex(raw_tx_reference)
         tx.vout = [tx.vout[0]] * 2
         tx.vout[0].nValue = MAX_MONEY
         self.check_mempool_result(
@@ -243,7 +240,7 @@ class MempoolAcceptanceTest(BitcoinTestFramework):
         )
 
         self.log.info('A transaction with duplicate inputs')
-        tx.deserialize(BytesIO(hex_str_to_bytes(raw_tx_reference)))
+        tx = tx_from_hex(raw_tx_reference)
         tx.vin = [tx.vin[0]] * 2
         self.check_mempool_result(
             result_expected=[{'txid': tx.rehash(), 'allowed': False, 'reject-reason': 'bad-txns-inputs-duplicate'}],
@@ -253,26 +250,26 @@ class MempoolAcceptanceTest(BitcoinTestFramework):
         self.log.info('A coinbase transaction')
         # Pick the input of the first tx we signed, so it has to be a coinbase tx
         raw_tx_coinbase_spent = node.getrawtransaction(txid=node.decoderawtransaction(hexstring=raw_tx_in_block)['vin'][0]['txid'])
-        tx.deserialize(BytesIO(hex_str_to_bytes(raw_tx_coinbase_spent)))
+        tx = tx_from_hex(raw_tx_coinbase_spent)
         self.check_mempool_result(
             result_expected=[{'txid': tx.rehash(), 'allowed': False, 'reject-reason': 'coinbase'}],
             rawtxs=[tx.serialize().hex()],
         )
 
         self.log.info('Some nonstandard transactions')
-        tx.deserialize(BytesIO(hex_str_to_bytes(raw_tx_reference)))
+        tx = tx_from_hex(raw_tx_reference)
         tx.nVersion = 3  # A version currently non-standard
         self.check_mempool_result(
             result_expected=[{'txid': tx.rehash(), 'allowed': False, 'reject-reason': 'version'}],
             rawtxs=[tx.serialize().hex()],
         )
-        tx.deserialize(BytesIO(hex_str_to_bytes(raw_tx_reference)))
+        tx = tx_from_hex(raw_tx_reference)
         tx.vout[0].scriptPubKey = CScript([OP_0])  # Some non-standard script
         self.check_mempool_result(
             result_expected=[{'txid': tx.rehash(), 'allowed': False, 'reject-reason': 'scriptpubkey'}],
             rawtxs=[tx.serialize().hex()],
         )
-        tx.deserialize(BytesIO(hex_str_to_bytes(raw_tx_reference)))
+        tx = tx_from_hex(raw_tx_reference)
         key = ECKey()
         key.generate()
         pubkey = key.get_pubkey().get_bytes()
@@ -281,19 +278,19 @@ class MempoolAcceptanceTest(BitcoinTestFramework):
             result_expected=[{'txid': tx.rehash(), 'allowed': False, 'reject-reason': 'bare-multisig'}],
             rawtxs=[tx.serialize().hex()],
         )
-        tx.deserialize(BytesIO(hex_str_to_bytes(raw_tx_reference)))
+        tx = tx_from_hex(raw_tx_reference)
         tx.vin[0].scriptSig = CScript([OP_HASH160])  # Some not-pushonly scriptSig
         self.check_mempool_result(
             result_expected=[{'txid': tx.rehash(), 'allowed': False, 'reject-reason': 'scriptsig-not-pushonly'}],
             rawtxs=[tx.serialize().hex()],
         )
-        tx.deserialize(BytesIO(hex_str_to_bytes(raw_tx_reference)))
+        tx = tx_from_hex(raw_tx_reference)
         tx.vin[0].scriptSig = CScript([b'a' * 1648]) # Some too large scriptSig (>1650 bytes)
         self.check_mempool_result(
             result_expected=[{'txid': tx.rehash(), 'allowed': False, 'reject-reason': 'scriptsig-size'}],
             rawtxs=[tx.serialize().hex()],
         )
-        tx.deserialize(BytesIO(hex_str_to_bytes(raw_tx_reference)))
+        tx = tx_from_hex(raw_tx_reference)
         output_p2sh_burn = CTxOut(nValue=540, scriptPubKey=CScript([OP_HASH160, hash160(b'burn'), OP_EQUAL]))
         num_scripts = 100000 // len(output_p2sh_burn.serialize())  # Use enough outputs to make the tx too large for our policy
         tx.vout = [output_p2sh_burn] * num_scripts
@@ -301,14 +298,14 @@ class MempoolAcceptanceTest(BitcoinTestFramework):
             result_expected=[{'txid': tx.rehash(), 'allowed': False, 'reject-reason': 'tx-size'}],
             rawtxs=[tx.serialize().hex()],
         )
-        tx.deserialize(BytesIO(hex_str_to_bytes(raw_tx_reference)))
+        tx = tx_from_hex(raw_tx_reference)
         tx.vout[0] = output_p2sh_burn
         tx.vout[0].nValue -= 1  # Make output smaller, such that it is dust for our policy
         self.check_mempool_result(
             result_expected=[{'txid': tx.rehash(), 'allowed': False, 'reject-reason': 'dust'}],
             rawtxs=[tx.serialize().hex()],
         )
-        tx.deserialize(BytesIO(hex_str_to_bytes(raw_tx_reference)))
+        tx = tx_from_hex(raw_tx_reference)
         tx.vout[0].scriptPubKey = CScript([OP_RETURN, b'\xff'])
         tx.vout = [tx.vout[0]] * 2
         self.check_mempool_result(
@@ -317,7 +314,7 @@ class MempoolAcceptanceTest(BitcoinTestFramework):
         )
 
         self.log.info('A timelocked transaction')
-        tx.deserialize(BytesIO(hex_str_to_bytes(raw_tx_reference)))
+        tx = tx_from_hex(raw_tx_reference)
         tx.vin[0].nSequence -= 1  # Should be non-max, so locktime is not ignored
         tx.nLockTime = node.getblockcount() + 1
         self.check_mempool_result(
@@ -326,7 +323,7 @@ class MempoolAcceptanceTest(BitcoinTestFramework):
         )
 
         self.log.info('A transaction that is locked by BIP68 sequence logic')
-        tx.deserialize(BytesIO(hex_str_to_bytes(raw_tx_reference)))
+        tx = tx_from_hex(raw_tx_reference)
         tx.vin[0].nSequence = 2  # We could include it in the second block mined from now, but not the very next one
         # Can skip re-signing the tx because of early rejection
         self.check_mempool_result(
diff --git a/test/functional/p2p_compactblocks.py b/test/functional/p2p_compactblocks.py
index 3e4f2f974d..3c771df0aa 100755
--- a/test/functional/p2p_compactblocks.py
+++ b/test/functional/p2p_compactblocks.py
@@ -15,9 +15,52 @@ from test_framework.blocktools import (
     add_witness_commitment,
     create_block,
 )
-from test_framework.messages import BlockTransactions, BlockTransactionsRequest, calculate_shortid, CBlock, CBlockHeader, CInv, COutPoint, CTransaction, CTxIn, CTxInWitness, CTxOut, FromHex, HeaderAndShortIDs, msg_no_witness_block, msg_no_witness_blocktxn, msg_cmpctblock, msg_getblocktxn, msg_getdata, msg_getheaders, msg_headers, msg_inv, msg_sendcmpct, msg_sendheaders, msg_tx, msg_block, msg_blocktxn, MSG_BLOCK, MSG_CMPCT_BLOCK, MSG_WITNESS_FLAG, NODE_NETWORK, P2PHeaderAndShortIDs, PrefilledTransaction, ser_uint256, ToHex
-from test_framework.p2p import p2p_lock, P2PInterface
-from test_framework.script import CScript, OP_TRUE, OP_DROP
+from test_framework.messages import (
+    BlockTransactions,
+    BlockTransactionsRequest,
+    CBlock,
+    CBlockHeader,
+    CInv,
+    COutPoint,
+    CTransaction,
+    CTxIn,
+    CTxInWitness,
+    CTxOut,
+    FromHex,
+    HeaderAndShortIDs,
+    MSG_BLOCK,
+    MSG_CMPCT_BLOCK,
+    MSG_WITNESS_FLAG,
+    NODE_NETWORK,
+    P2PHeaderAndShortIDs,
+    PrefilledTransaction,
+    ToHex,
+    calculate_shortid,
+    msg_block,
+    msg_blocktxn,
+    msg_cmpctblock,
+    msg_getblocktxn,
+    msg_getdata,
+    msg_getheaders,
+    msg_headers,
+    msg_inv,
+    msg_no_witness_block,
+    msg_no_witness_blocktxn,
+    msg_sendcmpct,
+    msg_sendheaders,
+    msg_tx,
+    ser_uint256,
+    tx_from_hex,
+)
+from test_framework.p2p import (
+    P2PInterface,
+    p2p_lock,
+)
+from test_framework.script import (
+    CScript,
+    OP_DROP,
+    OP_TRUE,
+)
 from test_framework.test_framework import BitcoinTestFramework
 from test_framework.util import assert_equal, softfork_active
 
@@ -257,7 +300,7 @@ class CompactBlocksTest(BitcoinTestFramework):
         for _ in range(num_transactions):
             txid = node.sendtoaddress(address, 0.1)
             hex_tx = node.gettransaction(txid)["hex"]
-            tx = FromHex(CTransaction(), hex_tx)
+            tx = tx_from_hex(hex_tx)
             if not tx.wit.is_null():
                 segwit_tx_generated = True
 
diff --git a/test/functional/p2p_eviction.py b/test/functional/p2p_eviction.py
index a525996493..35bce7c69e 100755
--- a/test/functional/p2p_eviction.py
+++ b/test/functional/p2p_eviction.py
@@ -20,7 +20,11 @@ from test_framework.blocktools import (
     create_block,
     create_coinbase,
 )
-from test_framework.messages import CTransaction, FromHex, msg_pong, msg_tx
+from test_framework.messages import (
+    msg_pong,
+    msg_tx,
+    tx_from_hex,
+)
 from test_framework.p2p import P2PDataStore, P2PInterface
 from test_framework.test_framework import BitcoinTestFramework
 from test_framework.util import assert_equal
@@ -89,7 +93,7 @@ class P2PEvict(BitcoinTestFramework):
                     'scriptPubKey': prevtx['vout'][0]['scriptPubKey']['hex'],
                 }],
             )['hex']
-            txpeer.send_message(msg_tx(FromHex(CTransaction(), sigtx)))
+            txpeer.send_message(msg_tx(tx_from_hex(sigtx)))
             protected_peers.add(current_peer)
 
         self.log.info("Create 8 peers and protect them from eviction by having faster pings")
diff --git a/test/functional/p2p_permissions.py b/test/functional/p2p_permissions.py
index 62652d949d..594a28d662 100755
--- a/test/functional/p2p_permissions.py
+++ b/test/functional/p2p_permissions.py
@@ -9,9 +9,8 @@ Test that permissions are correctly calculated and applied
 
 from test_framework.address import ADDRESS_BCRT1_P2WSH_OP_TRUE
 from test_framework.messages import (
-    CTransaction,
     CTxInWitness,
-    FromHex,
+    tx_from_hex,
 )
 from test_framework.p2p import P2PDataStore
 from test_framework.script import (
@@ -105,8 +104,7 @@ class P2PPermissionsTests(BitcoinTestFramework):
         p2p_rebroadcast_wallet = self.nodes[1].add_p2p_connection(P2PDataStore())
 
         self.log.debug("Send a tx from the wallet initially")
-        tx = FromHex(
-            CTransaction(),
+        tx = tx_from_hex(
             self.nodes[0].createrawtransaction(
                 inputs=[{
                     'txid': block_op_true['tx'][0],
diff --git a/test/functional/p2p_segwit.py b/test/functional/p2p_segwit.py
index 9d32c1cb86..95c7aec318 100755
--- a/test/functional/p2p_segwit.py
+++ b/test/functional/p2p_segwit.py
@@ -40,8 +40,8 @@ from test_framework.messages import (
     ser_uint256,
     ser_vector,
     sha256,
+    tx_from_hex,
     uint256_from_str,
-    FromHex,
 )
 from test_framework.p2p import (
     P2PInterface,
@@ -2122,14 +2122,14 @@ class SegWitTest(BitcoinTestFramework):
         unspent = next(u for u in self.nodes[0].listunspent() if u['spendable'] and u['address'].startswith('bcrt'))
 
         raw = self.nodes[0].createrawtransaction([{"txid": unspent['txid'], "vout": unspent['vout']}], {self.nodes[0].getnewaddress(): 1})
-        tx = FromHex(CTransaction(), raw)
+        tx = tx_from_hex(raw)
         assert_raises_rpc_error(-22, "TX decode failed", self.nodes[0].decoderawtransaction, hexstring=serialize_with_bogus_witness(tx).hex(), iswitness=True)
         with self.nodes[0].assert_debug_log(['Superfluous witness record']):
             self.test_node.send_and_ping(msg_bogus_tx(tx))
         raw = self.nodes[0].signrawtransactionwithwallet(raw)
         assert raw['complete']
         raw = raw['hex']
-        tx = FromHex(CTransaction(), raw)
+        tx = tx_from_hex(raw)
         assert_raises_rpc_error(-22, "TX decode failed", self.nodes[0].decoderawtransaction, hexstring=serialize_with_bogus_witness(tx).hex(), iswitness=True)
         with self.nodes[0].assert_debug_log(['Unknown transaction optional data']):
             self.test_node.send_and_ping(msg_bogus_tx(tx))
diff --git a/test/functional/p2p_tx_download.py b/test/functional/p2p_tx_download.py
index 4bf96cb0e6..3e962b4450 100755
--- a/test/functional/p2p_tx_download.py
+++ b/test/functional/p2p_tx_download.py
@@ -8,13 +8,12 @@ Test transaction download behavior
 
 from test_framework.messages import (
     CInv,
-    CTransaction,
-    FromHex,
     MSG_TX,
     MSG_TYPE_MASK,
     MSG_WTX,
     msg_inv,
     msg_notfound,
+    tx_from_hex,
 )
 from test_framework.p2p import (
     P2PInterface,
@@ -100,7 +99,7 @@ class TxDownloadTest(BitcoinTestFramework):
             hexstring=tx,
             privkeys=[self.nodes[0].get_deterministic_priv_key().key],
         )['hex']
-        ctx = FromHex(CTransaction(), tx)
+        ctx = tx_from_hex(tx)
         txid = int(ctx.rehash(), 16)
 
         self.log.info(
diff --git a/test/functional/rpc_addresses_deprecation.py b/test/functional/rpc_addresses_deprecation.py
index bc0559f3b5..ac430f5b39 100755
--- a/test/functional/rpc_addresses_deprecation.py
+++ b/test/functional/rpc_addresses_deprecation.py
@@ -4,9 +4,9 @@
 # file COPYING or http://www.opensource.org/licenses/mit-license.php.
 """Test deprecation of reqSigs and addresses RPC fields."""
 
-from io import BytesIO
-
-from test_framework.messages import CTransaction
+from test_framework.messages import (
+    tx_from_hex,
+)
 from test_framework.test_framework import BitcoinTestFramework
 from test_framework.util import (
     assert_equal,
@@ -35,8 +35,7 @@ class AddressesDeprecationTest(BitcoinTestFramework):
         signed = node.signrawtransactionwithwallet(raw)['hex']
 
         # This transaction is derived from test/util/data/txcreatemultisig1.json
-        tx = CTransaction()
-        tx.deserialize(BytesIO(hex_str_to_bytes(signed)))
+        tx = tx_from_hex(signed)
         tx.vout[0].scriptPubKey = hex_str_to_bytes("522102a5613bd857b7048924264d1e70e08fb2a7e6527d32b7ab1bb993ac59964ff39721021ac43c7ff740014c3b33737ede99c967e4764553d1b2b83db77c83b8715fa72d2102df2089105c77f266fa11a9d33f05c735234075f2e8780824c6b709415f9fb48553ae")
         tx_signed = node.signrawtransactionwithwallet(tx.serialize().hex())['hex']
         txid = node.sendrawtransaction(hexstring=tx_signed, maxfeerate=0)
diff --git a/test/functional/rpc_decodescript.py b/test/functional/rpc_decodescript.py
index 01b8cb1854..f6643c7167 100755
--- a/test/functional/rpc_decodescript.py
+++ b/test/functional/rpc_decodescript.py
@@ -4,11 +4,16 @@
 # file COPYING or http://www.opensource.org/licenses/mit-license.php.
 """Test decoding scripts via decodescript RPC command."""
 
-from test_framework.messages import CTransaction, sha256
+from test_framework.messages import (
+    sha256,
+    tx_from_hex,
+)
 from test_framework.test_framework import BitcoinTestFramework
-from test_framework.util import assert_equal, hex_str_to_bytes
+from test_framework.util import (
+    assert_equal,
+    hex_str_to_bytes,
+)
 
-from io import BytesIO
 
 class DecodeScriptTest(BitcoinTestFramework):
     def set_test_params(self):
@@ -179,8 +184,7 @@ class DecodeScriptTest(BitcoinTestFramework):
         assert_equal('0 3045022100ae3b4e589dfc9d48cb82d41008dc5fa6a86f94d5c54f9935531924602730ab8002202f88cf464414c4ed9fa11b773c5ee944f66e9b05cc1e51d97abc22ce098937ea[ALL] 3045022100b44883be035600e9328a01b66c7d8439b74db64187e76b99a68f7893b701d5380220225bf286493e4c4adcf928c40f785422572eb232f84a0b83b0dea823c3a19c75[ALL] 5221020743d44be989540d27b1b4bbbcfd17721c337cb6bc9af20eb8a32520b393532f2102c0120a1dda9e51a938d39ddd9fe0ebc45ea97e1d27a7cbd671d5431416d3dd87210213820eb3d5f509d7438c9eeecb4157b2f595105e7cd564b3cdbb9ead3da41eed53ae', rpc_result['vin'][0]['scriptSig']['asm'])
         assert_equal('OP_DUP OP_HASH160 dc863734a218bfe83ef770ee9d41a27f824a6e56 OP_EQUALVERIFY OP_CHECKSIG', rpc_result['vout'][0]['scriptPubKey']['asm'])
         assert_equal('OP_HASH160 2a5edea39971049a540474c6a99edf0aa4074c58 OP_EQUAL', rpc_result['vout'][1]['scriptPubKey']['asm'])
-        txSave = CTransaction()
-        txSave.deserialize(BytesIO(hex_str_to_bytes(tx)))
+        txSave = tx_from_hex(tx)
 
         # make sure that a specifically crafted op_return value will not pass all the IsDERSignature checks and then get decoded as a sighash type
         tx = '01000000015ded05872fdbda629c7d3d02b194763ce3b9b1535ea884e3c8e765d42e316724020000006b48304502204c10d4064885c42638cbff3585915b322de33762598321145ba033fc796971e2022100bb153ad3baa8b757e30a2175bd32852d2e1cb9080f84d7e32fcdfd667934ef1b012103163c0ff73511ea1743fb5b98384a2ff09dd06949488028fd819f4d83f56264efffffffff0200000000000000000b6a0930060201000201000180380100000000001976a9141cabd296e753837c086da7a45a6c2fe0d49d7b7b88ac00000000'
diff --git a/test/functional/rpc_packages.py b/test/functional/rpc_packages.py
index d0848d9bd6..4b2ed20958 100755
--- a/test/functional/rpc_packages.py
+++ b/test/functional/rpc_packages.py
@@ -5,7 +5,6 @@
 """RPCs that handle raw transaction packages."""
 
 from decimal import Decimal
-from io import BytesIO
 import random
 
 from test_framework.address import ADDRESS_BCRT1_P2WSH_OP_TRUE
@@ -13,8 +12,8 @@ from test_framework.test_framework import BitcoinTestFramework
 from test_framework.messages import (
     BIP125_SEQUENCE_NUMBER,
     COIN,
-    CTransaction,
     CTxInWitness,
+    tx_from_hex,
 )
 from test_framework.script import (
     CScript,
@@ -22,7 +21,6 @@ from test_framework.script import (
 )
 from test_framework.util import (
     assert_equal,
-    hex_str_to_bytes,
 )
 
 class RPCPackagesTest(BitcoinTestFramework):
@@ -97,9 +95,8 @@ class RPCPackagesTest(BitcoinTestFramework):
             "amount": parent_value,
         }] if parent_locking_script else None
         signedtx = node.signrawtransactionwithkey(hexstring=rawtx, privkeys=self.privkeys, prevtxs=prevtxs)
-        tx = CTransaction()
         assert signedtx["complete"]
-        tx.deserialize(BytesIO(hex_str_to_bytes(signedtx["hex"])))
+        tx = tx_from_hex(signedtx["hex"])
         return (tx, signedtx["hex"], my_value, tx.vout[0].scriptPubKey.hex())
 
     def test_independent(self):
@@ -110,8 +107,7 @@ class RPCPackagesTest(BitcoinTestFramework):
 
         self.log.info("Test an otherwise valid package with an extra garbage tx appended")
         garbage_tx = node.createrawtransaction([{"txid": "00" * 32, "vout": 5}], {self.address: 1})
-        tx = CTransaction()
-        tx.deserialize(BytesIO(hex_str_to_bytes(garbage_tx)))
+        tx = tx_from_hex(garbage_tx)
         # Only the txid and wtxids are returned because validation is incomplete for the independent txns.
         # Package validation is atomic: if the node cannot find a UTXO for any single tx in the package,
         # it terminates immediately to avoid unnecessary, expensive signature verification.
@@ -123,8 +119,7 @@ class RPCPackagesTest(BitcoinTestFramework):
         coin = self.coins.pop()
         tx_bad_sig_hex = node.createrawtransaction([{"txid": coin["txid"], "vout": 0}],
                                            {self.address : coin["amount"] - Decimal("0.0001")})
-        tx_bad_sig = CTransaction()
-        tx_bad_sig.deserialize(BytesIO(hex_str_to_bytes(tx_bad_sig_hex)))
+        tx_bad_sig = tx_from_hex(tx_bad_sig_hex)
         testres_bad_sig = node.testmempoolaccept(self.independent_txns_hex + [tx_bad_sig_hex])
         # By the time the signature for the last transaction is checked, all the other transactions
         # have been fully validated, which is why the node returns full validation results for all
@@ -141,8 +136,7 @@ class RPCPackagesTest(BitcoinTestFramework):
                                            {self.address : coin["amount"] - Decimal("0.999")})
         tx_high_fee_signed = node.signrawtransactionwithkey(hexstring=tx_high_fee_raw, privkeys=self.privkeys)
         assert tx_high_fee_signed["complete"]
-        tx_high_fee = CTransaction()
-        tx_high_fee.deserialize(BytesIO(hex_str_to_bytes(tx_high_fee_signed["hex"])))
+        tx_high_fee = tx_from_hex(tx_high_fee_signed["hex"])
         testres_high_fee = node.testmempoolaccept([tx_high_fee_signed["hex"]])
         assert_equal(testres_high_fee, [
             {"txid": tx_high_fee.rehash(), "wtxid": tx_high_fee.getwtxid(), "allowed": False, "reject-reason": "max-fee-exceeded"}
@@ -198,9 +192,8 @@ class RPCPackagesTest(BitcoinTestFramework):
         rawtx = node.createrawtransaction(inputs, outputs)
 
         parent_signed = node.signrawtransactionwithkey(hexstring=rawtx, privkeys=self.privkeys)
-        parent_tx = CTransaction()
         assert parent_signed["complete"]
-        parent_tx.deserialize(BytesIO(hex_str_to_bytes(parent_signed["hex"])))
+        parent_tx = tx_from_hex(parent_signed["hex"])
         parent_txid = parent_tx.rehash()
         assert node.testmempoolaccept([parent_signed["hex"]])[0]["allowed"]
 
@@ -213,8 +206,7 @@ class RPCPackagesTest(BitcoinTestFramework):
 
         # Child B
         rawtx_b = node.createrawtransaction([{"txid": parent_txid, "vout": 1}], {self.address : child_value})
-        tx_child_b = CTransaction()
-        tx_child_b.deserialize(BytesIO(hex_str_to_bytes(rawtx_b)))
+        tx_child_b = tx_from_hex(rawtx_b)
         tx_child_b.wit.vtxinwit = [CTxInWitness()]
         tx_child_b.wit.vtxinwit[0].scriptWitness.stack = [CScript([OP_TRUE])]
         tx_child_b_hex = tx_child_b.serialize().hex()
@@ -293,10 +285,8 @@ class RPCPackagesTest(BitcoinTestFramework):
         rawtx2 = node.createrawtransaction(inputs, output2)
         signedtx1 = node.signrawtransactionwithkey(hexstring=rawtx1, privkeys=self.privkeys)
         signedtx2 = node.signrawtransactionwithkey(hexstring=rawtx2, privkeys=self.privkeys)
-        tx1 = CTransaction()
-        tx1.deserialize(BytesIO(hex_str_to_bytes(signedtx1["hex"])))
-        tx2 = CTransaction()
-        tx2.deserialize(BytesIO(hex_str_to_bytes(signedtx2["hex"])))
+        tx1 = tx_from_hex(signedtx1["hex"])
+        tx2 = tx_from_hex(signedtx2["hex"])
         assert signedtx1["complete"]
         assert signedtx2["complete"]
 
@@ -327,19 +317,17 @@ class RPCPackagesTest(BitcoinTestFramework):
         raw_replaceable_tx = node.createrawtransaction(inputs, output)
         signed_replaceable_tx = node.signrawtransactionwithkey(hexstring=raw_replaceable_tx, privkeys=self.privkeys)
         testres_replaceable = node.testmempoolaccept([signed_replaceable_tx["hex"]])
-        replaceable_tx = CTransaction()
-        replaceable_tx.deserialize(BytesIO(hex_str_to_bytes(signed_replaceable_tx["hex"])))
+        replaceable_tx = tx_from_hex(signed_replaceable_tx["hex"])
         assert_equal(testres_replaceable, [
             {"txid": replaceable_tx.rehash(), "wtxid": replaceable_tx.getwtxid(),
             "allowed": True, "vsize": replaceable_tx.get_vsize(), "fees": { "base": fee }}
         ])
 
         # Replacement transaction is identical except has double the fee
-        replacement_tx = CTransaction()
-        replacement_tx.deserialize(BytesIO(hex_str_to_bytes(signed_replaceable_tx["hex"])))
+        replacement_tx = tx_from_hex(signed_replaceable_tx["hex"])
         replacement_tx.vout[0].nValue -= int(fee * COIN)  # Doubled fee
         signed_replacement_tx = node.signrawtransactionwithkey(replacement_tx.serialize().hex(), self.privkeys)
-        replacement_tx.deserialize(BytesIO(hex_str_to_bytes(signed_replacement_tx["hex"])))
+        replacement_tx = tx_from_hex(signed_replacement_tx["hex"])
 
         self.log.info("Test that transactions within a package cannot replace each other")
         testres_rbf_conflicting = node.testmempoolaccept([signed_replaceable_tx["hex"], signed_replacement_tx["hex"]])
diff --git a/test/functional/rpc_rawtransaction.py b/test/functional/rpc_rawtransaction.py
index 53ddf24e47..b3d5771534 100755
--- a/test/functional/rpc_rawtransaction.py
+++ b/test/functional/rpc_rawtransaction.py
@@ -14,16 +14,18 @@ Test the following RPCs:
 
 from collections import OrderedDict
 from decimal import Decimal
-from io import BytesIO
 
 from test_framework.blocktools import COINBASE_MATURITY
-from test_framework.messages import CTransaction, ToHex
+from test_framework.messages import (
+    CTransaction,
+    ToHex,
+    tx_from_hex,
+)
 from test_framework.test_framework import BitcoinTestFramework
 from test_framework.util import (
     assert_equal,
     assert_raises_rpc_error,
     find_vout_for_address,
-    hex_str_to_bytes,
 )
 
 
@@ -127,23 +129,22 @@ class RawTransactionsTest(BitcoinTestFramework):
         assert_raises_rpc_error(-3, "Expected type bool", self.nodes[0].createrawtransaction, [], {}, 0, 'foo')
 
         self.log.info('Check that createrawtransaction accepts an array and object as outputs')
-        tx = CTransaction()
         # One output
-        tx.deserialize(BytesIO(hex_str_to_bytes(self.nodes[2].createrawtransaction(inputs=[{'txid': txid, 'vout': 9}], outputs={address: 99}))))
+        tx = tx_from_hex(self.nodes[2].createrawtransaction(inputs=[{'txid': txid, 'vout': 9}], outputs={address: 99}))
         assert_equal(len(tx.vout), 1)
         assert_equal(
             tx.serialize().hex(),
             self.nodes[2].createrawtransaction(inputs=[{'txid': txid, 'vout': 9}], outputs=[{address: 99}]),
         )
         # Two outputs
-        tx.deserialize(BytesIO(hex_str_to_bytes(self.nodes[2].createrawtransaction(inputs=[{'txid': txid, 'vout': 9}], outputs=OrderedDict([(address, 99), (address2, 99)])))))
+        tx = tx_from_hex(self.nodes[2].createrawtransaction(inputs=[{'txid': txid, 'vout': 9}], outputs=OrderedDict([(address, 99), (address2, 99)])))
         assert_equal(len(tx.vout), 2)
         assert_equal(
             tx.serialize().hex(),
             self.nodes[2].createrawtransaction(inputs=[{'txid': txid, 'vout': 9}], outputs=[{address: 99}, {address2: 99}]),
         )
         # Multiple mixed outputs
-        tx.deserialize(BytesIO(hex_str_to_bytes(self.nodes[2].createrawtransaction(inputs=[{'txid': txid, 'vout': 9}], outputs=multidict([(address, 99), (address2, 99), ('data', '99')])))))
+        tx = tx_from_hex(self.nodes[2].createrawtransaction(inputs=[{'txid': txid, 'vout': 9}], outputs=multidict([(address, 99), (address2, 99), ('data', '99')])))
         assert_equal(len(tx.vout), 3)
         assert_equal(
             tx.serialize().hex(),
diff --git a/test/functional/rpc_signrawtransaction.py b/test/functional/rpc_signrawtransaction.py
index 16b0019866..ef5d08e7b9 100755
--- a/test/functional/rpc_signrawtransaction.py
+++ b/test/functional/rpc_signrawtransaction.py
@@ -5,17 +5,44 @@
 """Test transaction signing using the signrawtransaction* RPCs."""
 
 from test_framework.blocktools import COINBASE_MATURITY
-from test_framework.address import check_script, script_to_p2sh, script_to_p2wsh
+from test_framework.address import (
+    check_script,
+    script_to_p2sh,
+    script_to_p2wsh,
+)
 from test_framework.key import ECKey
 from test_framework.test_framework import BitcoinTestFramework
-from test_framework.util import assert_equal, assert_raises_rpc_error, find_vout_for_address, hex_str_to_bytes
-from test_framework.messages import sha256, CTransaction, CTxInWitness
-from test_framework.script import CScript, OP_0, OP_CHECKSIG, OP_CHECKSEQUENCEVERIFY, OP_CHECKLOCKTIMEVERIFY, OP_DROP, OP_TRUE
-from test_framework.script_util import key_to_p2pkh_script, script_to_p2sh_p2wsh_script, script_to_p2wsh_script
+from test_framework.util import (
+    assert_equal,
+    assert_raises_rpc_error,
+    find_vout_for_address,
+    hex_str_to_bytes,
+)
+from test_framework.messages import (
+    CTxInWitness,
+    sha256,
+    tx_from_hex,
+)
+from test_framework.script import (
+    CScript,
+    OP_0,
+    OP_CHECKLOCKTIMEVERIFY,
+    OP_CHECKSIG,
+    OP_CHECKSEQUENCEVERIFY,
+    OP_DROP,
+    OP_TRUE,
+)
+from test_framework.script_util import (
+    key_to_p2pkh_script,
+    script_to_p2sh_p2wsh_script,
+    script_to_p2wsh_script,
+)
 from test_framework.wallet_util import bytes_to_wif
 
-from decimal import Decimal, getcontext
-from io import BytesIO
+from decimal import (
+    Decimal,
+    getcontext,
+)
 
 class SignRawTransactionsTest(BitcoinTestFramework):
     def set_test_params(self):
@@ -265,8 +292,7 @@ class SignRawTransactionsTest(BitcoinTestFramework):
         )
 
         # Set the witness script
-        ctx = CTransaction()
-        ctx.deserialize(BytesIO(hex_str_to_bytes(tx)))
+        ctx = tx_from_hex(tx)
         ctx.wit.vtxinwit.append(CTxInWitness())
         ctx.wit.vtxinwit[0].scriptWitness.stack = [CScript([OP_TRUE]), script]
         tx = ctx.serialize_with_witness().hex()
@@ -301,8 +327,7 @@ class SignRawTransactionsTest(BitcoinTestFramework):
         )
 
         # Set the witness script
-        ctx = CTransaction()
-        ctx.deserialize(BytesIO(hex_str_to_bytes(tx)))
+        ctx = tx_from_hex(tx)
         ctx.wit.vtxinwit.append(CTxInWitness())
         ctx.wit.vtxinwit[0].scriptWitness.stack = [CScript([OP_TRUE]), script]
         tx = ctx.serialize_with_witness().hex()
diff --git a/test/functional/test_framework/blocktools.py b/test/functional/test_framework/blocktools.py
index f35ea6c122..0019b7a426 100644
--- a/test/functional/test_framework/blocktools.py
+++ b/test/functional/test_framework/blocktools.py
@@ -23,12 +23,12 @@ from .messages import (
     CTxIn,
     CTxInWitness,
     CTxOut,
-    FromHex,
     ToHex,
     hash256,
     hex_str_to_bytes,
     ser_uint256,
     sha256,
+    tx_from_hex,
     uint256_from_str,
 )
 from .script import (
@@ -79,7 +79,7 @@ def create_block(hashprev=None, coinbase=None, ntime=None, *, version=None, tmpl
     if txlist:
         for tx in txlist:
             if not hasattr(tx, 'calc_sha256'):
-                tx = FromHex(CTransaction(), tx)
+                tx = tx_from_hex(tx)
             block.vtx.append(tx)
     block.hashMerkleRoot = block.calc_merkle_root()
     block.calc_sha256()
@@ -166,7 +166,7 @@ def create_transaction(node, txid, to_address, *, amount):
         sign for the output that is being spent.
     """
     raw_tx = create_raw_transaction(node, txid, to_address, amount=amount)
-    tx = FromHex(CTransaction(), raw_tx)
+    tx = tx_from_hex(raw_tx)
     return tx
 
 def create_raw_transaction(node, txid, to_address, *, amount):
@@ -248,7 +248,7 @@ def send_to_witness(use_p2wsh, node, utxo, pubkey, encode_p2sh, amount, sign=Tru
         return node.sendrawtransaction(signed["hex"])
     else:
         if (insert_redeem_script):
-            tx = FromHex(CTransaction(), tx_to_witness)
+            tx = tx_from_hex(tx_to_witness)
             tx.vin[0].scriptSig += CScript([hex_str_to_bytes(insert_redeem_script)])
             tx_to_witness = ToHex(tx)
 
diff --git a/test/functional/test_framework/messages.py b/test/functional/test_framework/messages.py
index 3d79174356..43c80fa7a8 100755
--- a/test/functional/test_framework/messages.py
+++ b/test/functional/test_framework/messages.py
@@ -195,10 +195,17 @@ def FromHex(obj, hex_string):
     obj.deserialize(BytesIO(hex_str_to_bytes(hex_string)))
     return obj
 
+
 # Convert a binary-serializable object to hex (eg for submission via RPC)
 def ToHex(obj):
     return obj.serialize().hex()
 
+
+def tx_from_hex(hex_string):
+    """Deserialize from hex string to a transaction object"""
+    return FromHex(CTransaction(), hex_string)
+
+
 # Objects that map to bitcoind objects, which can be serialized/deserialized
 
 
diff --git a/test/functional/test_framework/util.py b/test/functional/test_framework/util.py
index f354ae0075..35dbfbba8d 100644
--- a/test/functional/test_framework/util.py
+++ b/test/functional/test_framework/util.py
@@ -19,7 +19,6 @@ import unittest
 
 from . import coverage
 from .authproxy import AuthServiceProxy, JSONRPCException
-from io import BytesIO
 from typing import Callable, Optional
 
 logger = logging.getLogger("TestFramework.utils")
@@ -528,7 +527,7 @@ def gen_return_txouts():
 def create_lots_of_big_transactions(node, txouts, utxos, num, fee):
     addr = node.getnewaddress()
     txids = []
-    from .messages import CTransaction
+    from .messages import tx_from_hex
     for _ in range(num):
         t = utxos.pop()
         inputs = [{"txid": t["txid"], "vout": t["vout"]}]
@@ -536,8 +535,7 @@ def create_lots_of_big_transactions(node, txouts, utxos, num, fee):
         change = t['amount'] - fee
         outputs[addr] = satoshi_round(change)
         rawtx = node.createrawtransaction(inputs, outputs)
-        tx = CTransaction()
-        tx.deserialize(BytesIO(hex_str_to_bytes(rawtx)))
+        tx = tx_from_hex(rawtx)
         for txout in txouts:
             tx.vout.append(txout)
         newtx = tx.serialize().hex()
diff --git a/test/functional/wallet_bumpfee.py b/test/functional/wallet_bumpfee.py
index b21461ee7b..c04986038d 100755
--- a/test/functional/wallet_bumpfee.py
+++ b/test/functional/wallet_bumpfee.py
@@ -14,17 +14,23 @@ added in the future, they should try to follow the same convention and not
 make assumptions about execution order.
 """
 from decimal import Decimal
-import io
 
-from test_framework.blocktools import COINBASE_MATURITY
-from test_framework.blocktools import add_witness_commitment, create_block, create_coinbase, send_to_witness
-from test_framework.messages import BIP125_SEQUENCE_NUMBER, CTransaction
+from test_framework.blocktools import (
+    COINBASE_MATURITY,
+    add_witness_commitment,
+    create_block,
+    create_coinbase,
+    send_to_witness,
+)
+from test_framework.messages import (
+    BIP125_SEQUENCE_NUMBER,
+    tx_from_hex,
+)
 from test_framework.test_framework import BitcoinTestFramework
 from test_framework.util import (
     assert_equal,
     assert_greater_than,
     assert_raises_rpc_error,
-    hex_str_to_bytes,
 )
 
 WALLET_PASSPHRASE = "test"
@@ -576,9 +582,7 @@ def spend_one_input(node, dest_address, change_size=Decimal("0.00049000")):
 
 
 def submit_block_with_tx(node, tx):
-    ctx = CTransaction()
-    ctx.deserialize(io.BytesIO(hex_str_to_bytes(tx)))
-
+    ctx = tx_from_hex(tx)
     tip = node.getbestblockhash()
     height = node.getblockcount() + 1
     block_time = node.getblockheader(tip)["mediantime"] + 1
diff --git a/test/functional/wallet_groups.py b/test/functional/wallet_groups.py
index f32acb8e15..1b52ddcba5 100755
--- a/test/functional/wallet_groups.py
+++ b/test/functional/wallet_groups.py
@@ -6,7 +6,10 @@
 
 from test_framework.blocktools import COINBASE_MATURITY
 from test_framework.test_framework import BitcoinTestFramework
-from test_framework.messages import CTransaction, FromHex, ToHex
+from test_framework.messages import (
+    ToHex,
+    tx_from_hex,
+)
 from test_framework.util import (
     assert_approx,
     assert_equal,
@@ -154,7 +157,7 @@ class WalletGroupTest(BitcoinTestFramework):
         self.log.info("Fill a wallet with 10,000 outputs corresponding to the same scriptPubKey")
         for _ in range(5):
             raw_tx = self.nodes[0].createrawtransaction([{"txid":"0"*64, "vout":0}], [{addr2[0]: 0.05}])
-            tx = FromHex(CTransaction(), raw_tx)
+            tx = tx_from_hex(raw_tx)
             tx.vin = []
             tx.vout = [tx.vout[0]] * 2000
             funded_tx = self.nodes[0].fundrawtransaction(ToHex(tx))
diff --git a/test/functional/wallet_listtransactions.py b/test/functional/wallet_listtransactions.py
index 71573964de..8b503f5971 100755
--- a/test/functional/wallet_listtransactions.py
+++ b/test/functional/wallet_listtransactions.py
@@ -4,22 +4,17 @@
 # file COPYING or http://www.opensource.org/licenses/mit-license.php.
 """Test the listtransactions API."""
 from decimal import Decimal
-from io import BytesIO
 
-from test_framework.messages import COIN, CTransaction
+from test_framework.messages import (
+    COIN,
+    tx_from_hex,
+)
 from test_framework.test_framework import BitcoinTestFramework
 from test_framework.util import (
     assert_array_result,
     assert_equal,
-    hex_str_to_bytes,
 )
 
-def tx_from_hex(hexstring):
-    tx = CTransaction()
-    f = BytesIO(hex_str_to_bytes(hexstring))
-    tx.deserialize(f)
-    return tx
-
 class ListTransactionsTest(BitcoinTestFramework):
     def set_test_params(self):
         self.num_nodes = 2
diff --git a/test/functional/wallet_txn_clone.py b/test/functional/wallet_txn_clone.py
index 84ff9ad772..76b39201e3 100755
--- a/test/functional/wallet_txn_clone.py
+++ b/test/functional/wallet_txn_clone.py
@@ -4,12 +4,14 @@
 # file COPYING or http://www.opensource.org/licenses/mit-license.php.
 """Test the wallet accounts properly when there are cloned transactions with malleated scriptsigs."""
 
-import io
 from test_framework.test_framework import BitcoinTestFramework
 from test_framework.util import (
     assert_equal,
 )
-from test_framework.messages import CTransaction, COIN
+from test_framework.messages import (
+    COIN,
+    tx_from_hex,
+)
 
 
 class TxnMallTest(BitcoinTestFramework):
@@ -71,8 +73,7 @@ class TxnMallTest(BitcoinTestFramework):
         clone_raw = self.nodes[0].createrawtransaction(clone_inputs, clone_outputs, clone_locktime)
 
         # createrawtransaction randomizes the order of its outputs, so swap them if necessary.
-        clone_tx = CTransaction()
-        clone_tx.deserialize(io.BytesIO(bytes.fromhex(clone_raw)))
+        clone_tx = tx_from_hex(clone_raw)
         if (rawtx1["vout"][0]["value"] == 40 and clone_tx.vout[0].nValue != 40*COIN or rawtx1["vout"][0]["value"] != 40 and clone_tx.vout[0].nValue == 40*COIN):
             (clone_tx.vout[0], clone_tx.vout[1]) = (clone_tx.vout[1], clone_tx.vout[0])