test: move create_malleated_version() to messages.py for reuse

test/functional/p2p_orphan_handling.py

@@ -11,11 +11,11 @@
 )
 from test_framework.messages import (
     CInv,
-    CTxInWitness,
     DEFAULT_ANCESTOR_LIMIT,
     MSG_TX,
     MSG_WITNESS_TX,
     MSG_WTX,
+    malleate_tx_to_invalid_witness,
     msg_getdata,
     msg_inv,
     msg_notfound,
@@ -138,22 +138,6 @@ def relay_transaction(self, peer, tx):
         peer.wait_for_getdata([wtxid])
         peer.send_and_ping(msg_tx(tx))
 
-    def create_malleated_version(self, tx):
-        """
-        Create a malleated version of the tx where the witness is replaced with garbage data.
-        Returns a CTransaction object.
-        """
-        tx_bad_wit = tx_from_hex(tx["hex"])
-        tx_bad_wit.wit.vtxinwit = [CTxInWitness()]
-        # Add garbage data to witness 0. We cannot simply strip the witness, as the node would
-        # classify it as a transaction in which the witness was missing rather than wrong.
-        tx_bad_wit.wit.vtxinwit[0].scriptWitness.stack = [b'garbage']
-
-        assert_equal(tx["txid"], tx_bad_wit.txid_hex)
-        assert_not_equal(tx["wtxid"], tx_bad_wit.wtxid_hex)
-
-        return tx_bad_wit
-
     @cleanup
     def test_arrival_timing_orphan(self):
         self.log.info("Test missing parents that arrive during delay are not requested")
@@ -449,7 +433,7 @@ def test_same_txid_orphan(self):
         tx_child = self.wallet.create_self_transfer(utxo_to_spend=tx_parent["new_utxo"])
 
         # Create a fake version of the child
-        tx_orphan_bad_wit = self.create_malleated_version(tx_child)
+        tx_orphan_bad_wit = malleate_tx_to_invalid_witness(tx_child)
 
         bad_peer = node.add_p2p_connection(P2PInterface())
         honest_peer = node.add_p2p_connection(P2PInterface())
@@ -496,7 +480,7 @@ def test_same_txid_orphan_of_orphan(self):
         tx_middle = self.wallet.create_self_transfer(utxo_to_spend=tx_grandparent["new_utxo"])
 
         # Create a fake version of the middle tx
-        tx_orphan_bad_wit = self.create_malleated_version(tx_middle)
+        tx_orphan_bad_wit = malleate_tx_to_invalid_witness(tx_middle)
 
         # Create grandchild spending from tx_middle (and spending from tx_orphan_bad_wit since they
         # have the same txid).
@@ -550,7 +534,7 @@ def test_orphan_txid_inv(self):
 
         # Create the real child and fake version
         tx_child = self.wallet.create_self_transfer(utxo_to_spend=tx_parent["new_utxo"])
-        tx_orphan_bad_wit = self.create_malleated_version(tx_child)
+        tx_orphan_bad_wit = malleate_tx_to_invalid_witness(tx_child)
 
         bad_peer = node.add_p2p_connection(PeerTxRelayer())
         # Must not send wtxidrelay because otherwise the inv(TX) will be ignored later

test/functional/test_framework/messages.py

@@ -29,7 +29,10 @@
 import unittest
 
 from test_framework.crypto.siphash import siphash256
-from test_framework.util import assert_equal
+from test_framework.util import (
+    assert_equal,
+    assert_not_equal,
+)
 
 MAX_LOCATOR_SZ = 101
 MAX_BLOCK_WEIGHT = 4000000
@@ -253,6 +256,23 @@ def tx_from_hex(hex_string):
     return from_hex(CTransaction(), hex_string)
 
 
+def malleate_tx_to_invalid_witness(tx):
+    """
+    Create a malleated version of the tx where the witness is replaced with garbage data.
+    Returns a CTransaction object.
+    """
+    tx_bad_wit = tx_from_hex(tx["hex"])
+    tx_bad_wit.wit.vtxinwit = [CTxInWitness()]
+    # Add garbage data to witness 0. We cannot simply strip the witness, as the node would
+    # classify it as a transaction in which the witness was missing rather than wrong.
+    tx_bad_wit.wit.vtxinwit[0].scriptWitness.stack = [b'garbage']
+
+    assert_equal(tx["txid"], tx_bad_wit.txid_hex)
+    assert_not_equal(tx["wtxid"], tx_bad_wit.wtxid_hex)
+
+    return tx_bad_wit
+
+
 # like from_hex, but without the hex part
 def from_binary(cls, stream):
     """deserialize a binary stream (or bytes object) into an object"""