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