wallet_sweepprivkeys.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 sweepprivkeys RPC."""
6
7 from test_framework.key import ECKey
8 from test_framework.test_framework import LimenkaTestFramework
9 from test_framework.descriptors import descsum_create
10 from test_framework.util import assert_equal, assert_fee_amount
11 from test_framework.wallet_util import bytes_to_wif
12
13 class SweepPrivKeysTest(LimenkaTestFramework):
14 def add_options(self, parser):
15 self.add_wallet_options(parser)
16
17 def set_test_params(self):
18 self.num_nodes = 2
19
20 def check_balance(self, delta, txid):
21 node = self.nodes[0]
22 new_balances = node.getbalances()['mine']
23 new_balance = new_balances['trusted'] + new_balances['untrusted_pending']
24 balance_change = new_balance - self.balance
25 actual_fee = delta - balance_change
26 tx_vsize = node.getrawtransaction(txid, True)['vsize']
27 assert_fee_amount(actual_fee, tx_vsize, self.tx_feerate)
28 self.balance = new_balance
29
30 def skip_test_if_missing_module(self):
31 self.skip_if_no_wallet()
32
33 def run_test(self):
34 node = self.nodes[0]
35 miner = self.nodes[1]
36
37 keys = (
38 ('mkckmmfVv89sW1HUjyRuydGhwFmSaYtRvG', '92YkaycAxLPUqbbV78V9nNngKLnyVd9T8uZuZAzQnc26dJSP4fm'),
39 ('mw8s1FS2Vr7GwQF8bnDVUQHQZq5qWqz5kq', '93VijJgAYnVUGXAfxYhbMHVGVwQUEXK1YnPvcCod3x1RLbzUhXe'),
40 )
41
42 # This test is not meant to test fee estimation and we'd like
43 # to be sure all txs are sent at a consistent desired feerate
44 self.tx_feerate = max(self.nodes[0].getnetworkinfo()['relayfee'], self.nodes[0].getwalletinfo()['mintxfee']) * 2
45 node.settxfee(self.tx_feerate)
46
47 self.generate(miner, 120)
48 self.balance = node.getbalance('*', 0)
49
50 txid = node.sendtoaddress(keys[0][0], 10)
51 self.check_balance(-10, txid)
52
53 # Sweep from mempool
54 txid = node.sweepprivkeys({'privkeys': (keys[0][1],), 'label': 'test 1'})
55 assert_equal(node.listtransactions()[-1]['label'], 'test 1')
56 self.check_balance(10, txid)
57
58 txid = node.sendtoaddress(keys[1][0], 5)
59 self.check_balance(-5, txid)
60 self.sync_all()
61 self.generate(miner, 4)
62 assert_equal(self.balance, node.getbalance('*', 1))
63
64 # Sweep from blockchain
65 txid = node.sweepprivkeys({'privkeys': (keys[1][1],), 'label': 'test 2'})
66 assert_equal(node.listtransactions()[-1]['label'], 'test 2')
67 self.check_balance(5, txid)
68
69 # Test sweeping segwit address types (P2WPKH, P2SH-P2WPKH, P2TR)
70 self.log.info("Test sweeping P2WPKH, P2SH-P2WPKH, and P2TR outputs")
71 eckey = ECKey()
72 eckey.generate(compressed=True)
73 wif = bytes_to_wif(eckey.get_bytes(), compressed=True)
74 for desc_fmt, addr_type in (
75 ("wpkh(%s)", "P2WPKH"),
76 ("sh(wpkh(%s))", "P2SH-P2WPKH"),
77 ("tr(%s)", "P2TR"),
78 ):
79 desc = descsum_create(desc_fmt % wif)
80 addr = node.deriveaddresses(desc)[0]
81 txid = node.sendtoaddress(addr, 2)
82 self.check_balance(-2, txid)
83 self.sync_all()
84 self.generate(miner, 1)
85 txid = node.sweepprivkeys({'privkeys': (wif,), 'label': addr_type})
86 assert_equal(node.listtransactions()[-1]['label'], addr_type)
87 self.check_balance(2, txid)
88
89 if __name__ == '__main__':
90 SweepPrivKeysTest(__file__).main()
91