1  #!/usr/bin/env python3
  2  # Copyright (c) 2018-present The Bitcoin Core developers
  3  # Distributed under the MIT software license, see the accompanying
  4  # file COPYING or http://www.opensource.org/licenses/mit-license.php.
  5  
  6  from test_framework.messages import (
  7      tx_from_hex,
  8  )
  9  from test_framework.test_framework import BitcoinTestFramework
 10  from test_framework.util import (
 11      assert_equal,
 12      assert_raises_rpc_error,
 13  )
 14  from test_framework.blocktools import (
 15      TIME_GENESIS_BLOCK,
 16  )
 17  
 18  from decimal import Decimal
 19  
 20  class CreateTxWalletTest(BitcoinTestFramework):
 21      def set_test_params(self):
 22          self.setup_clean_chain = True
 23          self.num_nodes = 1
 24          self.extra_args = [[]]
 25  
 26      def skip_test_if_missing_module(self):
 27          self.skip_if_no_wallet()
 28  
 29      def run_test(self):
 30          self.log.info('Create some old blocks')
 31          self.nodes[0].setmocktime(TIME_GENESIS_BLOCK)
 32          self.generate(self.nodes[0], 200)
 33          self.nodes[0].setmocktime(0)
 34  
 35          self.test_anti_fee_sniping()
 36          self.test_tx_size_too_large()
 37          self.test_create_too_long_mempool_chain()
 38          self.test_version3()
 39  
 40      def test_anti_fee_sniping(self):
 41          self.log.info('Check that we have some (old) blocks and that anti-fee-sniping is disabled')
 42          assert_equal(self.nodes[0].getblockchaininfo()['blocks'], 200)
 43          txid = self.nodes[0].sendtoaddress(self.nodes[0].getnewaddress(), 1)
 44          tx = self.nodes[0].gettransaction(txid=txid, verbose=True)['decoded']
 45          assert_equal(tx['locktime'], 0)
 46  
 47          self.log.info('Check that anti-fee-sniping is enabled when we mine a recent block')
 48          self.generate(self.nodes[0], 1)
 49          txid = self.nodes[0].sendtoaddress(self.nodes[0].getnewaddress(), 1)
 50          tx = self.nodes[0].gettransaction(txid=txid, verbose=True)['decoded']
 51          assert 0 < tx['locktime'] <= 201
 52  
 53      def test_tx_size_too_large(self):
 54          # More than 10kB of outputs, so that we hit -maxtxfee with a high feerate
 55          outputs = {self.nodes[0].getnewaddress(address_type='bech32'): 0.000025 for _ in range(400)}
 56          raw_tx = self.nodes[0].createrawtransaction(inputs=[], outputs=outputs)
 57  
 58          for fee_setting in ['-minrelaytxfee=0.01', '-mintxfee=0.01']:
 59              self.log.info('Check maxtxfee in combination with {}'.format(fee_setting))
 60              self.restart_node(0, extra_args=[fee_setting])
 61              assert_raises_rpc_error(
 62                  -6,
 63                  "Fee exceeds maximum configured by user (e.g. -maxtxfee, maxfeerate)",
 64                  lambda: self.nodes[0].sendmany(dummy="", amounts=outputs),
 65              )
 66              assert_raises_rpc_error(
 67                  -4,
 68                  "Fee exceeds maximum configured by user (e.g. -maxtxfee, maxfeerate)",
 69                  lambda: self.nodes[0].fundrawtransaction(hexstring=raw_tx),
 70              )
 71  
 72          # Hit maxtxfee with explicit fee rate
 73          self.log.info('Check maxtxfee in combination with explicit fee_rate=1000 sat/vB')
 74  
 75          fee_rate_sats_per_vb = Decimal('0.01') * Decimal(1e8) / 1000  # Convert 0.01 BTC/kvB to sat/vB
 76  
 77          assert_raises_rpc_error(
 78              -6,
 79              "Fee exceeds maximum configured by user (e.g. -maxtxfee, maxfeerate)",
 80              lambda: self.nodes[0].sendmany(dummy="", amounts=outputs, fee_rate=fee_rate_sats_per_vb),
 81          )
 82          assert_raises_rpc_error(
 83              -4,
 84              "Fee exceeds maximum configured by user (e.g. -maxtxfee, maxfeerate)",
 85              lambda: self.nodes[0].fundrawtransaction(hexstring=raw_tx, options={'fee_rate': fee_rate_sats_per_vb}),
 86          )
 87  
 88      def test_create_too_long_mempool_chain(self):
 89          self.log.info('Check too-long mempool chain error')
 90          df_wallet = self.nodes[0].get_wallet_rpc(self.default_wallet_name)
 91  
 92          self.nodes[0].createwallet("too_long")
 93          test_wallet = self.nodes[0].get_wallet_rpc("too_long")
 94  
 95          tx_data = df_wallet.send(outputs=[{test_wallet.getnewaddress(): 25}], options={"change_position": 0})
 96          txid = tx_data['txid']
 97          vout = 1
 98  
 99          self.nodes[0].syncwithvalidationinterfacequeue()
100          options = {"change_position": 0, "add_inputs": False}
101          for i in range(1, 25):
102              options['inputs'] = [{'txid': txid, 'vout': vout}]
103              tx_data = test_wallet.send(outputs=[{test_wallet.getnewaddress(): 25 - i}], options=options)
104              txid = tx_data['txid']
105  
106          # Sending one more chained transaction will fail
107          options = {"minconf": 0, "include_unsafe": True, 'add_inputs': True}
108          assert_raises_rpc_error(-4, "Unconfirmed UTXOs are available, but spending them creates a chain of transactions that will be rejected by the mempool",
109                                  test_wallet.send, outputs=[{test_wallet.getnewaddress(): 0.3}], options=options)
110  
111          test_wallet.unloadwallet()
112  
113      def test_version3(self):
114          self.log.info('Check wallet does not create transactions with version=3 yet')
115          wallet_rpc = self.nodes[0].get_wallet_rpc(self.default_wallet_name)
116  
117          self.nodes[0].createwallet("version3")
118          wallet_v3 = self.nodes[0].get_wallet_rpc("version3")
119  
120          tx_data = wallet_rpc.send(outputs=[{wallet_v3.getnewaddress(): 25}], options={"change_position": 0})
121          wallet_tx_data = wallet_rpc.gettransaction(tx_data["txid"])
122          tx_current_version = tx_from_hex(wallet_tx_data["hex"])
123  
124          # While version=3 transactions are standard, the CURRENT_VERSION is 2.
125          # This test can be removed if CURRENT_VERSION is changed, and replaced with tests that the
126          # wallet handles TRUC rules properly.
127          assert_equal(tx_current_version.version, 2)
128          wallet_v3.unloadwallet()
129  
130  
131  if __name__ == '__main__':
132      CreateTxWalletTest(__file__).main()