test: use MiniWallet for feature_csv_activation.py

This test can now be run even with the Bitcoin Core wallet disabled.

Author: theStack

test/functional/feature_csv_activation.py

@@ -37,13 +37,13 @@
 bip112tx_special - test negative argument to OP_CSV
 bip112tx_emptystack - test empty stack (= no argument) OP_CSV
 """
-from decimal import Decimal
 from itertools import product
-from io import BytesIO
 import time
 
-from test_framework.blocktools import create_coinbase, create_block, create_transaction
-from test_framework.messages import ToHex, CTransaction
+from test_framework.blocktools import (
+    create_block,
+    create_coinbase,
+)
 from test_framework.p2p import P2PDataStore
 from test_framework.script import (
     CScript,
@@ -53,9 +53,9 @@
 from test_framework.test_framework import BitcoinTestFramework
 from test_framework.util import (
     assert_equal,
-    hex_str_to_bytes,
     softfork_active,
 )
+from test_framework.wallet import MiniWallet
 
 TESTING_TX_COUNT = 83  # Number of testing transactions: 1 BIP113 tx, 16 BIP68 txs, 66 BIP112 txs (see comments above)
 COINBASE_BLOCK_COUNT = TESTING_TX_COUNT  # Number of coinbase blocks we need to generate as inputs for our txs
@@ -83,65 +83,60 @@ def relative_locktime(sdf, srhb, stf, srlb):
 def all_rlt_txs(txs):
     return [tx['tx'] for tx in txs]
 
-def sign_transaction(node, unsignedtx):
-    rawtx = ToHex(unsignedtx)
-    signresult = node.signrawtransactionwithwallet(rawtx)
-    tx = CTransaction()
-    f = BytesIO(hex_str_to_bytes(signresult['hex']))
-    tx.deserialize(f)
+def create_self_transfer_from_utxo(node, input_tx):
+    utxo = miniwallet.get_utxo(txid=input_tx.rehash(), mark_as_spent=False)
+    tx = miniwallet.create_self_transfer(from_node=node, utxo_to_spend=utxo)['tx']
     return tx
 
-def create_bip112special(node, input, txversion, address):
-    tx = create_transaction(node, input, address, amount=Decimal("49.98"))
+def create_bip112special(node, input, txversion):
+    tx = create_self_transfer_from_utxo(node, input)
     tx.nVersion = txversion
-    signtx = sign_transaction(node, tx)
-    signtx.vin[0].scriptSig = CScript([-1, OP_CHECKSEQUENCEVERIFY, OP_DROP] + list(CScript(signtx.vin[0].scriptSig)))
-    return signtx
+    tx.vin[0].scriptSig = CScript([-1, OP_CHECKSEQUENCEVERIFY, OP_DROP] + list(CScript(tx.vin[0].scriptSig)))
+    return tx
 
-def create_bip112emptystack(node, input, txversion, address):
-    tx = create_transaction(node, input, address, amount=Decimal("49.98"))
+def create_bip112emptystack(node, input, txversion):
+    tx = create_self_transfer_from_utxo(node, input)
     tx.nVersion = txversion
-    signtx = sign_transaction(node, tx)
-    signtx.vin[0].scriptSig = CScript([OP_CHECKSEQUENCEVERIFY] + list(CScript(signtx.vin[0].scriptSig)))
-    return signtx
+    tx.vin[0].scriptSig = CScript([OP_CHECKSEQUENCEVERIFY] + list(CScript(tx.vin[0].scriptSig)))
+    return tx
 
-def send_generic_input_tx(node, coinbases, address):
-    return node.sendrawtransaction(ToHex(sign_transaction(node, create_transaction(node, node.getblock(coinbases.pop())['tx'][0], address, amount=Decimal("49.99")))))
+def send_generic_input_tx(node, coinbases):
+    input_txid = node.getblock(coinbases.pop(), 2)['tx'][0]['txid']
+    utxo_to_spend = miniwallet.get_utxo(txid=input_txid)
+    return miniwallet.send_self_transfer(from_node=node, utxo_to_spend=utxo_to_spend)['tx']
 
-def create_bip68txs(node, bip68inputs, txversion, address, locktime_delta=0):
+def create_bip68txs(node, bip68inputs, txversion, locktime_delta=0):
     """Returns a list of bip68 transactions with different bits set."""
     txs = []
     assert len(bip68inputs) >= 16
     for i, (sdf, srhb, stf, srlb) in enumerate(product(*[[True, False]] * 4)):
         locktime = relative_locktime(sdf, srhb, stf, srlb)
-        tx = create_transaction(node, bip68inputs[i], address, amount=Decimal("49.98"))
+        tx = create_self_transfer_from_utxo(node, bip68inputs[i])
         tx.nVersion = txversion
         tx.vin[0].nSequence = locktime + locktime_delta
-        tx = sign_transaction(node, tx)
         tx.rehash()
         txs.append({'tx': tx, 'sdf': sdf, 'stf': stf})
 
     return txs
 
-def create_bip112txs(node, bip112inputs, varyOP_CSV, txversion, address, locktime_delta=0):
+def create_bip112txs(node, bip112inputs, varyOP_CSV, txversion, locktime_delta=0):
     """Returns a list of bip68 transactions with different bits set."""
     txs = []
     assert len(bip112inputs) >= 16
     for i, (sdf, srhb, stf, srlb) in enumerate(product(*[[True, False]] * 4)):
         locktime = relative_locktime(sdf, srhb, stf, srlb)
-        tx = create_transaction(node, bip112inputs[i], address, amount=Decimal("49.98"))
+        tx = create_self_transfer_from_utxo(node, bip112inputs[i])
         if (varyOP_CSV):  # if varying OP_CSV, nSequence is fixed
             tx.vin[0].nSequence = BASE_RELATIVE_LOCKTIME + locktime_delta
         else:  # vary nSequence instead, OP_CSV is fixed
             tx.vin[0].nSequence = locktime + locktime_delta
         tx.nVersion = txversion
-        signtx = sign_transaction(node, tx)
         if (varyOP_CSV):
-            signtx.vin[0].scriptSig = CScript([locktime, OP_CHECKSEQUENCEVERIFY, OP_DROP] + list(CScript(signtx.vin[0].scriptSig)))
+            tx.vin[0].scriptSig = CScript([locktime, OP_CHECKSEQUENCEVERIFY, OP_DROP] + list(CScript(tx.vin[0].scriptSig)))
         else:
-            signtx.vin[0].scriptSig = CScript([BASE_RELATIVE_LOCKTIME, OP_CHECKSEQUENCEVERIFY, OP_DROP] + list(CScript(signtx.vin[0].scriptSig)))
+            tx.vin[0].scriptSig = CScript([BASE_RELATIVE_LOCKTIME, OP_CHECKSEQUENCEVERIFY, OP_DROP] + list(CScript(tx.vin[0].scriptSig)))
         tx.rehash()
-        txs.append({'tx': signtx, 'sdf': sdf, 'stf': stf})
+        txs.append({'tx': tx, 'sdf': sdf, 'stf': stf})
     return txs
 
 class BIP68_112_113Test(BitcoinTestFramework):
@@ -150,14 +145,11 @@ def set_test_params(self):
         self.setup_clean_chain = True
         self.extra_args = [[
             '[email protected]',
-            '-addresstype=legacy',
+            '-acceptnonstdtxn=1',
             '-par=1',  # Use only one script thread to get the exact reject reason for testing
         ]]
         self.supports_cli = False
 
-    def skip_test_if_missing_module(self):
-        self.skip_if_no_wallet()
-
     def generate_blocks(self, number):
         test_blocks = []
         for _ in range(number):
@@ -185,16 +177,18 @@ def send_blocks(self, blocks, success=True, reject_reason=None):
 
     def run_test(self):
         self.helper_peer = self.nodes[0].add_p2p_connection(P2PDataStore())
+        # TODO: store as class member to get rid of global variable
+        global miniwallet
+        miniwallet = MiniWallet(self.nodes[0], raw_script=True)
 
         self.log.info("Generate blocks in the past for coinbase outputs.")
         long_past_time = int(time.time()) - 600 * 1000  # enough to build up to 1000 blocks 10 minutes apart without worrying about getting into the future
         self.nodes[0].setmocktime(long_past_time - 100)  # enough so that the generated blocks will still all be before long_past_time
-        self.coinbase_blocks = self.nodes[0].generate(COINBASE_BLOCK_COUNT)  # blocks generated for inputs
+        self.coinbase_blocks = miniwallet.generate(COINBASE_BLOCK_COUNT)  # blocks generated for inputs
         self.nodes[0].setmocktime(0)  # set time back to present so yielded blocks aren't in the future as we advance last_block_time
         self.tipheight = COINBASE_BLOCK_COUNT  # height of the next block to build
         self.last_block_time = long_past_time
         self.tip = int(self.nodes[0].getbestblockhash(), 16)
-        self.nodeaddress = self.nodes[0].getnewaddress()
 
         # Activation height is hardcoded
         # We advance to block height five below BIP112 activation for the following tests
@@ -209,31 +203,31 @@ def run_test(self):
         # 16 normal inputs
         bip68inputs = []
         for _ in range(16):
-            bip68inputs.append(send_generic_input_tx(self.nodes[0], self.coinbase_blocks, self.nodeaddress))
+            bip68inputs.append(send_generic_input_tx(self.nodes[0], self.coinbase_blocks))
 
         # 2 sets of 16 inputs with 10 OP_CSV OP_DROP (actually will be prepended to spending scriptSig)
         bip112basicinputs = []
         for _ in range(2):
             inputs = []
             for _ in range(16):
-                inputs.append(send_generic_input_tx(self.nodes[0], self.coinbase_blocks, self.nodeaddress))
+                inputs.append(send_generic_input_tx(self.nodes[0], self.coinbase_blocks))
             bip112basicinputs.append(inputs)
 
         # 2 sets of 16 varied inputs with (relative_lock_time) OP_CSV OP_DROP (actually will be prepended to spending scriptSig)
         bip112diverseinputs = []
         for _ in range(2):
             inputs = []
             for _ in range(16):
-                inputs.append(send_generic_input_tx(self.nodes[0], self.coinbase_blocks, self.nodeaddress))
+                inputs.append(send_generic_input_tx(self.nodes[0], self.coinbase_blocks))
             bip112diverseinputs.append(inputs)
 
         # 1 special input with -1 OP_CSV OP_DROP (actually will be prepended to spending scriptSig)
-        bip112specialinput = send_generic_input_tx(self.nodes[0], self.coinbase_blocks, self.nodeaddress)
+        bip112specialinput = send_generic_input_tx(self.nodes[0], self.coinbase_blocks)
         # 1 special input with (empty stack) OP_CSV (actually will be prepended to spending scriptSig)
-        bip112emptystackinput = send_generic_input_tx(self.nodes[0],self.coinbase_blocks, self.nodeaddress)
+        bip112emptystackinput = send_generic_input_tx(self.nodes[0], self.coinbase_blocks)
 
         # 1 normal input
-        bip113input = send_generic_input_tx(self.nodes[0], self.coinbase_blocks, self.nodeaddress)
+        bip113input = send_generic_input_tx(self.nodes[0], self.coinbase_blocks)
 
         self.nodes[0].setmocktime(self.last_block_time + 600)
         inputblockhash = self.nodes[0].generate(1)[0]  # 1 block generated for inputs to be in chain at height 431
@@ -253,36 +247,36 @@ def run_test(self):
 
         # Test both version 1 and version 2 transactions for all tests
         # BIP113 test transaction will be modified before each use to put in appropriate block time
-        bip113tx_v1 = create_transaction(self.nodes[0], bip113input, self.nodeaddress, amount=Decimal("49.98"))
+        bip113tx_v1 = create_self_transfer_from_utxo(self.nodes[0], bip113input)
         bip113tx_v1.vin[0].nSequence = 0xFFFFFFFE
         bip113tx_v1.nVersion = 1
-        bip113tx_v2 = create_transaction(self.nodes[0], bip113input, self.nodeaddress, amount=Decimal("49.98"))
+        bip113tx_v2 = create_self_transfer_from_utxo(self.nodes[0], bip113input)
         bip113tx_v2.vin[0].nSequence = 0xFFFFFFFE
         bip113tx_v2.nVersion = 2
 
         # For BIP68 test all 16 relative sequence locktimes
-        bip68txs_v1 = create_bip68txs(self.nodes[0], bip68inputs, 1, self.nodeaddress)
-        bip68txs_v2 = create_bip68txs(self.nodes[0], bip68inputs, 2, self.nodeaddress)
+        bip68txs_v1 = create_bip68txs(self.nodes[0], bip68inputs, 1)
+        bip68txs_v2 = create_bip68txs(self.nodes[0], bip68inputs, 2)
 
         # For BIP112 test:
         # 16 relative sequence locktimes of 10 against 10 OP_CSV OP_DROP inputs
-        bip112txs_vary_nSequence_v1 = create_bip112txs(self.nodes[0], bip112basicinputs[0], False, 1, self.nodeaddress)
-        bip112txs_vary_nSequence_v2 = create_bip112txs(self.nodes[0], bip112basicinputs[0], False, 2, self.nodeaddress)
+        bip112txs_vary_nSequence_v1 = create_bip112txs(self.nodes[0], bip112basicinputs[0], False, 1)
+        bip112txs_vary_nSequence_v2 = create_bip112txs(self.nodes[0], bip112basicinputs[0], False, 2)
         # 16 relative sequence locktimes of 9 against 10 OP_CSV OP_DROP inputs
-        bip112txs_vary_nSequence_9_v1 = create_bip112txs(self.nodes[0], bip112basicinputs[1], False, 1, self.nodeaddress, -1)
-        bip112txs_vary_nSequence_9_v2 = create_bip112txs(self.nodes[0], bip112basicinputs[1], False, 2, self.nodeaddress, -1)
+        bip112txs_vary_nSequence_9_v1 = create_bip112txs(self.nodes[0], bip112basicinputs[1], False, 1, -1)
+        bip112txs_vary_nSequence_9_v2 = create_bip112txs(self.nodes[0], bip112basicinputs[1], False, 2, -1)
         # sequence lock time of 10 against 16 (relative_lock_time) OP_CSV OP_DROP inputs
-        bip112txs_vary_OP_CSV_v1 = create_bip112txs(self.nodes[0], bip112diverseinputs[0], True, 1, self.nodeaddress)
-        bip112txs_vary_OP_CSV_v2 = create_bip112txs(self.nodes[0], bip112diverseinputs[0], True, 2, self.nodeaddress)
+        bip112txs_vary_OP_CSV_v1 = create_bip112txs(self.nodes[0], bip112diverseinputs[0], True, 1)
+        bip112txs_vary_OP_CSV_v2 = create_bip112txs(self.nodes[0], bip112diverseinputs[0], True, 2)
         # sequence lock time of 9 against 16 (relative_lock_time) OP_CSV OP_DROP inputs
-        bip112txs_vary_OP_CSV_9_v1 = create_bip112txs(self.nodes[0], bip112diverseinputs[1], True, 1, self.nodeaddress, -1)
-        bip112txs_vary_OP_CSV_9_v2 = create_bip112txs(self.nodes[0], bip112diverseinputs[1], True, 2, self.nodeaddress, -1)
+        bip112txs_vary_OP_CSV_9_v1 = create_bip112txs(self.nodes[0], bip112diverseinputs[1], True, 1, -1)
+        bip112txs_vary_OP_CSV_9_v2 = create_bip112txs(self.nodes[0], bip112diverseinputs[1], True, 2, -1)
         # -1 OP_CSV OP_DROP input
-        bip112tx_special_v1 = create_bip112special(self.nodes[0], bip112specialinput, 1, self.nodeaddress)
-        bip112tx_special_v2 = create_bip112special(self.nodes[0], bip112specialinput, 2, self.nodeaddress)
+        bip112tx_special_v1 = create_bip112special(self.nodes[0], bip112specialinput, 1)
+        bip112tx_special_v2 = create_bip112special(self.nodes[0], bip112specialinput, 2)
         # (empty stack) OP_CSV input
-        bip112tx_emptystack_v1 = create_bip112emptystack(self.nodes[0], bip112emptystackinput, 1, self.nodeaddress)
-        bip112tx_emptystack_v2 = create_bip112emptystack(self.nodes[0], bip112emptystackinput, 2, self.nodeaddress)
+        bip112tx_emptystack_v1 = create_bip112emptystack(self.nodes[0], bip112emptystackinput, 1)
+        bip112tx_emptystack_v2 = create_bip112emptystack(self.nodes[0], bip112emptystackinput, 2)
 
         self.log.info("TESTING")
 
@@ -292,8 +286,8 @@ def run_test(self):
         success_txs = []
         # BIP113 tx, -1 CSV tx and empty stack CSV tx should succeed
         bip113tx_v1.nLockTime = self.last_block_time - 600 * 5  # = MTP of prior block (not <) but < time put on current block
-        bip113signed1 = sign_transaction(self.nodes[0], bip113tx_v1)
-        success_txs.append(bip113signed1)
+        bip113tx_v1.rehash()
+        success_txs.append(bip113tx_v1)
         success_txs.append(bip112tx_special_v1)
         success_txs.append(bip112tx_emptystack_v1)
         # add BIP 68 txs
@@ -312,8 +306,8 @@ def run_test(self):
         success_txs = []
         # BIP113 tx, -1 CSV tx and empty stack CSV tx should succeed
         bip113tx_v2.nLockTime = self.last_block_time - 600 * 5  # = MTP of prior block (not <) but < time put on current block
-        bip113signed2 = sign_transaction(self.nodes[0], bip113tx_v2)
-        success_txs.append(bip113signed2)
+        bip113tx_v2.rehash()
+        success_txs.append(bip113tx_v2)
         success_txs.append(bip112tx_special_v2)
         success_txs.append(bip112tx_emptystack_v2)
         # add BIP 68 txs
@@ -338,17 +332,18 @@ def run_test(self):
         self.log.info("BIP 113 tests")
         # BIP 113 tests should now fail regardless of version number if nLockTime isn't satisfied by new rules
         bip113tx_v1.nLockTime = self.last_block_time - 600 * 5  # = MTP of prior block (not <) but < time put on current block
-        bip113signed1 = sign_transaction(self.nodes[0], bip113tx_v1)
+        bip113tx_v1.rehash()
         bip113tx_v2.nLockTime = self.last_block_time - 600 * 5  # = MTP of prior block (not <) but < time put on current block
-        bip113signed2 = sign_transaction(self.nodes[0], bip113tx_v2)
-        for bip113tx in [bip113signed1, bip113signed2]:
+        bip113tx_v2.rehash()
+        for bip113tx in [bip113tx_v1, bip113tx_v2]:
             self.send_blocks([self.create_test_block([bip113tx])], success=False, reject_reason='bad-txns-nonfinal')
+
         # BIP 113 tests should now pass if the locktime is < MTP
         bip113tx_v1.nLockTime = self.last_block_time - 600 * 5 - 1  # < MTP of prior block
-        bip113signed1 = sign_transaction(self.nodes[0], bip113tx_v1)
+        bip113tx_v1.rehash()
         bip113tx_v2.nLockTime = self.last_block_time - 600 * 5 - 1  # < MTP of prior block
-        bip113signed2 = sign_transaction(self.nodes[0], bip113tx_v2)
-        for bip113tx in [bip113signed1, bip113signed2]:
+        bip113tx_v2.rehash()
+        for bip113tx in [bip113tx_v1, bip113tx_v2]:
             self.send_blocks([self.create_test_block([bip113tx])])
             self.nodes[0].invalidateblock(self.nodes[0].getbestblockhash())
 
@@ -471,8 +466,8 @@ def run_test(self):
         time_txs = []
         for tx in [tx['tx'] for tx in bip112txs_vary_OP_CSV_v2 if not tx['sdf'] and tx['stf']]:
             tx.vin[0].nSequence = BASE_RELATIVE_LOCKTIME | SEQ_TYPE_FLAG
-            signtx = sign_transaction(self.nodes[0], tx)
-            time_txs.append(signtx)
+            tx.rehash()
+            time_txs.append(tx)
 
         self.send_blocks([self.create_test_block(time_txs)])
         self.nodes[0].invalidateblock(self.nodes[0].getbestblockhash())

test/functional/test_framework/wallet.py

@@ -61,7 +61,7 @@ def generate(self, num_blocks):
     def get_address(self):
         return self._address
 
-    def get_utxo(self, *, txid=''):
+    def get_utxo(self, *, txid='', mark_as_spent=True):
         """
         Returns a utxo and marks it as spent (pops it from the internal list)
 
@@ -74,7 +74,10 @@ def get_utxo(self, *, txid=''):
         if txid:
             utxo = next(filter(lambda utxo: txid == utxo['txid'], self._utxos))
             index = self._utxos.index(utxo)
-        return self._utxos.pop(index)
+        if mark_as_spent:
+            return self._utxos.pop(index)
+        else:
+            return self._utxos[index]
 
     def send_self_transfer(self, *, fee_rate=Decimal("0.003"), from_node, utxo_to_spend=None):
         """Create and send a tx with the specified fee_rate. Fee may be exact or at most one satoshi higher than needed."""