1  #!/usr/bin/env python3
  2  # Copyright (c) 2014-2022 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  """Test the importprunedfunds and removeprunedfunds RPCs."""
  6  from decimal import Decimal
  7  
  8  from test_framework.address import key_to_p2wpkh
  9  from test_framework.blocktools import COINBASE_MATURITY
 10  from test_framework.messages import (
 11      CMerkleBlock,
 12      from_hex,
 13  )
 14  from test_framework.test_framework import BitcoinTestFramework
 15  from test_framework.util import (
 16      assert_equal,
 17      assert_raises_rpc_error,
 18  )
 19  from test_framework.wallet_util import generate_keypair
 20  
 21  
 22  class ImportPrunedFundsTest(BitcoinTestFramework):
 23      def set_test_params(self):
 24          self.setup_clean_chain = True
 25          self.num_nodes = 2
 26  
 27      def skip_test_if_missing_module(self):
 28          self.skip_if_no_wallet()
 29  
 30      def run_test(self):
 31          self.log.info("Mining blocks...")
 32          self.generate(self.nodes[0], COINBASE_MATURITY + 1)
 33  
 34          # address
 35          address1 = self.nodes[0].getnewaddress()
 36          # pubkey
 37          address2 = self.nodes[0].getnewaddress()
 38          # privkey
 39          address3_privkey, address3_pubkey = generate_keypair(wif=True)
 40          address3 = key_to_p2wpkh(address3_pubkey)
 41          self.nodes[0].importprivkey(address3_privkey)
 42  
 43          # Check only one address
 44          address_info = self.nodes[0].getaddressinfo(address1)
 45          assert_equal(address_info['ismine'], True)
 46  
 47          self.sync_all()
 48  
 49          # Node 1 sync test
 50          assert_equal(self.nodes[1].getblockcount(), COINBASE_MATURITY + 1)
 51  
 52          # Address Test - before import
 53          address_info = self.nodes[1].getaddressinfo(address1)
 54          assert_equal(address_info['iswatchonly'], False)
 55          assert_equal(address_info['ismine'], False)
 56  
 57          address_info = self.nodes[1].getaddressinfo(address2)
 58          assert_equal(address_info['iswatchonly'], False)
 59          assert_equal(address_info['ismine'], False)
 60  
 61          address_info = self.nodes[1].getaddressinfo(address3)
 62          assert_equal(address_info['iswatchonly'], False)
 63          assert_equal(address_info['ismine'], False)
 64  
 65          # Send funds to self
 66          txnid1 = self.nodes[0].sendtoaddress(address1, 0.1)
 67          self.generate(self.nodes[0], 1)
 68          rawtxn1 = self.nodes[0].gettransaction(txnid1)['hex']
 69          proof1 = self.nodes[0].gettxoutproof([txnid1])
 70  
 71          txnid2 = self.nodes[0].sendtoaddress(address2, 0.05)
 72          self.generate(self.nodes[0], 1)
 73          rawtxn2 = self.nodes[0].gettransaction(txnid2)['hex']
 74          proof2 = self.nodes[0].gettxoutproof([txnid2])
 75  
 76          txnid3 = self.nodes[0].sendtoaddress(address3, 0.025)
 77          self.generate(self.nodes[0], 1)
 78          rawtxn3 = self.nodes[0].gettransaction(txnid3)['hex']
 79          proof3 = self.nodes[0].gettxoutproof([txnid3])
 80  
 81          self.sync_all()
 82  
 83          # Import with no affiliated address
 84          assert_raises_rpc_error(-5, "No addresses", self.nodes[1].importprunedfunds, rawtxn1, proof1)
 85  
 86          balance1 = self.nodes[1].getbalance()
 87          assert_equal(balance1, Decimal(0))
 88  
 89          # Import with affiliated address with no rescan
 90          self.nodes[1].createwallet('wwatch', disable_private_keys=True)
 91          wwatch = self.nodes[1].get_wallet_rpc('wwatch')
 92          wwatch.importaddress(address=address2, rescan=False)
 93          wwatch.importprunedfunds(rawtransaction=rawtxn2, txoutproof=proof2)
 94          assert [tx for tx in wwatch.listtransactions(include_watchonly=True) if tx['txid'] == txnid2]
 95  
 96          # Import with private key with no rescan
 97          w1 = self.nodes[1].get_wallet_rpc(self.default_wallet_name)
 98          w1.importprivkey(privkey=address3_privkey, rescan=False)
 99          w1.importprunedfunds(rawtxn3, proof3)
100          assert [tx for tx in w1.listtransactions() if tx['txid'] == txnid3]
101          balance3 = w1.getbalance()
102          assert_equal(balance3, Decimal('0.025'))
103  
104          # Addresses Test - after import
105          address_info = w1.getaddressinfo(address1)
106          assert_equal(address_info['iswatchonly'], False)
107          assert_equal(address_info['ismine'], False)
108          address_info = wwatch.getaddressinfo(address2)
109          assert_equal(address_info['iswatchonly'], False)
110          assert_equal(address_info['ismine'], True)
111          address_info = w1.getaddressinfo(address3)
112          assert_equal(address_info['iswatchonly'], False)
113          assert_equal(address_info['ismine'], True)
114  
115          # Remove transactions
116          assert_raises_rpc_error(-4, f'Transaction {txnid1} does not belong to this wallet', w1.removeprunedfunds, txnid1)
117          assert not [tx for tx in w1.listtransactions(include_watchonly=True) if tx['txid'] == txnid1]
118  
119          wwatch.removeprunedfunds(txnid2)
120          assert not [tx for tx in wwatch.listtransactions(include_watchonly=True) if tx['txid'] == txnid2]
121  
122          w1.removeprunedfunds(txnid3)
123          assert not [tx for tx in w1.listtransactions(include_watchonly=True) if tx['txid'] == txnid3]
124  
125          # Check various RPC parameter validation errors
126          assert_raises_rpc_error(-22, "TX decode failed", w1.importprunedfunds, b'invalid tx'.hex(), proof1)
127          assert_raises_rpc_error(-5, "Transaction given doesn't exist in proof", w1.importprunedfunds, rawtxn2, proof1)
128  
129          mb = from_hex(CMerkleBlock(), proof1)
130          mb.header.hashMerkleRoot = 0xdeadbeef  # cause mismatch between merkle root and merkle block
131          assert_raises_rpc_error(-5, "Something wrong with merkleblock", w1.importprunedfunds, rawtxn1, mb.serialize().hex())
132  
133          mb = from_hex(CMerkleBlock(), proof1)
134          mb.header.nTime += 1  # modify arbitrary block header field to change block hash
135          assert_raises_rpc_error(-5, "Block not found in chain", w1.importprunedfunds, rawtxn1, mb.serialize().hex())
136  
137  
138  if __name__ == '__main__':
139      ImportPrunedFundsTest(__file__).main()