wallet_importprunedfunds.py raw
1 #!/usr/bin/env python3
2 # Copyright (c) 2014-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 """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_not_equal,
18 assert_raises_rpc_error,
19 wallet_importprivkey,
20 )
21 from test_framework.wallet_util import generate_keypair
22
23
24 class ImportPrunedFundsTest(BitcoinTestFramework):
25 def set_test_params(self):
26 self.setup_clean_chain = True
27 self.num_nodes = 2
28
29 def skip_test_if_missing_module(self):
30 self.skip_if_no_wallet()
31
32 def run_test(self):
33 self.log.info("Mining blocks...")
34 self.generate(self.nodes[0], COINBASE_MATURITY + 1)
35
36 # address
37 address1 = self.nodes[0].getnewaddress()
38 # pubkey
39 address2 = self.nodes[0].getnewaddress()
40 # privkey
41 address3_privkey, address3_pubkey = generate_keypair(wif=True)
42 address3 = key_to_p2wpkh(address3_pubkey)
43 wallet_importprivkey(self.nodes[0], address3_privkey, "now")
44
45 # Check only one address
46 address_info = self.nodes[0].getaddressinfo(address1)
47 assert_equal(address_info['ismine'], True)
48
49 self.sync_all()
50
51 # Node 1 sync test
52 assert_equal(self.nodes[1].getblockcount(), COINBASE_MATURITY + 1)
53
54 # Address Test - before import
55 address_info = self.nodes[1].getaddressinfo(address1)
56 assert_equal(address_info['ismine'], False)
57
58 address_info = self.nodes[1].getaddressinfo(address2)
59 assert_equal(address_info['ismine'], False)
60
61 address_info = self.nodes[1].getaddressinfo(address3)
62 assert_equal(address_info['ismine'], False)
63
64 # Send funds to self
65 txnid1 = self.nodes[0].sendtoaddress(address1, 0.1)
66 self.generate(self.nodes[0], 1)
67 rawtxn1 = self.nodes[0].gettransaction(txnid1)['hex']
68 proof1 = self.nodes[0].gettxoutproof([txnid1])
69
70 txnid2 = self.nodes[0].sendtoaddress(address2, 0.05)
71 self.generate(self.nodes[0], 1)
72 rawtxn2 = self.nodes[0].gettransaction(txnid2)['hex']
73 proof2 = self.nodes[0].gettxoutproof([txnid2])
74
75 txnid3 = self.nodes[0].sendtoaddress(address3, 0.025)
76 self.generate(self.nodes[0], 1)
77 rawtxn3 = self.nodes[0].gettransaction(txnid3)['hex']
78 proof3 = self.nodes[0].gettxoutproof([txnid3])
79
80 self.sync_all()
81
82 # Import with no affiliated address
83 assert_raises_rpc_error(-5, "No addresses", self.nodes[1].importprunedfunds, rawtxn1, proof1)
84
85 balance1 = self.nodes[1].getbalance()
86 assert_equal(balance1, Decimal(0))
87
88 # Import with affiliated address with no rescan
89 self.nodes[1].createwallet('wwatch', disable_private_keys=True)
90 wwatch = self.nodes[1].get_wallet_rpc('wwatch')
91 wwatch.importdescriptors([{"desc": self.nodes[0].getaddressinfo(address2)["desc"], "timestamp": "now"}])
92 wwatch.importprunedfunds(rawtransaction=rawtxn2, txoutproof=proof2)
93 assert txnid2 in [tx['txid'] for tx in wwatch.listtransactions()]
94
95 # Import with private key with no rescan
96 w1 = self.nodes[1].get_wallet_rpc(self.default_wallet_name)
97 wallet_importprivkey(w1, address3_privkey, "now")
98 w1.importprunedfunds(rawtxn3, proof3)
99 assert txnid3 in [tx['txid'] for tx in w1.listtransactions()]
100 balance3 = w1.getbalance()
101 assert_equal(balance3, Decimal('0.025'))
102
103 # Addresses Test - after import
104 address_info = w1.getaddressinfo(address1)
105 assert_equal(address_info['ismine'], False)
106 address_info = wwatch.getaddressinfo(address2)
107 assert_equal(address_info['ismine'], True)
108 address_info = w1.getaddressinfo(address3)
109 assert_equal(address_info['ismine'], True)
110
111 # Remove transactions
112 assert_raises_rpc_error(-4, f'Transaction {txnid1} does not belong to this wallet', w1.removeprunedfunds, txnid1)
113 assert txnid1 not in [tx['txid'] for tx in w1.listtransactions()]
114
115 wwatch.removeprunedfunds(txnid2)
116 assert txnid2 not in [tx['txid'] for tx in wwatch.listtransactions()]
117
118 w1.removeprunedfunds(txnid3)
119 assert txnid3 not in [tx['txid'] for tx in w1.listtransactions()]
120
121 # Check various RPC parameter validation errors
122 assert_raises_rpc_error(-22, "TX decode failed", w1.importprunedfunds, b'invalid tx'.hex(), proof1)
123 assert_raises_rpc_error(-5, "Transaction given doesn't exist in proof", w1.importprunedfunds, rawtxn2, proof1)
124
125 mb = from_hex(CMerkleBlock(), proof1)
126 mb.header.hashMerkleRoot = 0xdeadbeef # cause mismatch between merkle root and merkle block
127 assert_raises_rpc_error(-5, "Something wrong with merkleblock", w1.importprunedfunds, rawtxn1, mb.serialize().hex())
128
129 mb = from_hex(CMerkleBlock(), proof1)
130 mb.header.nTime += 1 # modify arbitrary block header field to change block hash
131 assert_raises_rpc_error(-5, "Block not found in chain", w1.importprunedfunds, rawtxn1, mb.serialize().hex())
132
133 self.log.info("Test removeprunedfunds with conflicting transactions")
134 node = self.nodes[0]
135
136 # Create a transaction
137 utxo = node.listunspent()[0]
138 addr = node.getnewaddress()
139 tx1_id = node.send(outputs=[{addr: 1}], inputs=[utxo])["txid"]
140 tx1_fee = node.gettransaction(tx1_id)["fee"]
141
142 # Create a conflicting tx with a larger fee (tx1_fee is negative)
143 output_value = utxo["amount"] + tx1_fee - Decimal("0.00001")
144 raw_tx2 = node.createrawtransaction(inputs=[utxo], outputs=[{addr: output_value}])
145 signed_tx2 = node.signrawtransactionwithwallet(raw_tx2)
146 tx2_id = node.sendrawtransaction(signed_tx2["hex"])
147 assert_not_equal(tx2_id, tx1_id)
148
149 # Both txs should be in the wallet, tx2 replaced tx1 in mempool
150 assert tx1_id in [tx["txid"] for tx in node.listtransactions()]
151 assert tx2_id in [tx["txid"] for tx in node.listtransactions()]
152
153 # Remove the replaced tx from wallet
154 node.removeprunedfunds(tx1_id)
155
156 # The UTXO should still be considered spent (by tx2)
157 available_utxos = [u["txid"] for u in node.listunspent(minconf=0)]
158 assert utxo["txid"] not in available_utxos, "UTXO should still be spent by conflicting tx"
159
160
161 if __name__ == '__main__':
162 ImportPrunedFundsTest(__file__).main()
163