wallet_labels.py raw
1 #!/usr/bin/env python3
2 # Copyright (c) 2016-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 label RPCs.
6
7 RPCs tested are:
8 - getaddressesbylabel
9 - listaddressgroupings
10 - setlabel
11 """
12 from collections import defaultdict
13
14 from test_framework.blocktools import COINBASE_MATURITY
15 from test_framework.test_framework import LimenkaTestFramework
16 from test_framework.util import assert_equal, assert_raises_rpc_error
17 from test_framework.wallet_util import test_address
18
19
20 class WalletLabelsTest(LimenkaTestFramework):
21 def add_options(self, parser):
22 self.add_wallet_options(parser)
23
24 def set_test_params(self):
25 self.setup_clean_chain = True
26 self.num_nodes = 2
27
28 def skip_test_if_missing_module(self):
29 self.skip_if_no_wallet()
30
31 def invalid_label_name_test(self):
32 node = self.nodes[0]
33 address = node.getnewaddress()
34 pubkey = node.getaddressinfo(address)['pubkey']
35 rpc_calls = [
36 [node.getnewaddress],
37 [node.setlabel, address],
38 [node.getaddressesbylabel],
39 [node.importpubkey, pubkey],
40 [node.addmultisigaddress, 1, [pubkey]],
41 [node.getreceivedbylabel],
42 [node.listsinceblock, node.getblockhash(0), 1, False, True, False],
43 ]
44 if self.options.descriptors:
45 response = node.importdescriptors([{
46 'desc': f'pkh({pubkey})',
47 'label': '*',
48 'timestamp': 'now',
49 }])
50 else:
51 rpc_calls.extend([
52 [node.importprivkey, node.dumpprivkey(address)],
53 [node.importaddress, address],
54 ])
55
56 response = node.importmulti([{
57 'scriptPubKey': {'address': address},
58 'label': '*',
59 'timestamp': 'now',
60 }])
61
62 assert_equal(response[0]['success'], False)
63 assert_equal(response[0]['error']['code'], -11)
64 assert_equal(response[0]['error']['message'], "Invalid label name")
65
66 for rpc_call in rpc_calls:
67 assert_raises_rpc_error(-11, "Invalid label name", *rpc_call, "*")
68
69 def test_sort_multisig(self, node):
70 node.importprivkey("cSJUMwramrFYHKPfY77FH94bv4Q5rwUCyfD6zX3kLro4ZcWsXFEM")
71 node.importprivkey("cSpQbSsdKRmxaSWJ3TckCFTrksXNPbh8tfeZESGNQekkVxMbQ77H")
72 node.importprivkey("cRNbfcJgnvk2QJEVbMsxzoprotm1cy3kVA2HoyjSs3ss5NY5mQqr")
73
74 addresses = [
75 "muRmfCwue81ZT9oc3NaepefPscUHtP5kyC",
76 "n12RzKwqWPPA4cWGzkiebiM7Gu6NXUnDW8",
77 "n2yWMtx8jVbo8wv9BK2eN1LdbaakgKL3Mt",
78 ]
79
80 sorted_default = node.addmultisigaddress(2, addresses, None, 'legacy')
81 sorted_false = node.addmultisigaddress(2, addresses, {"sort": False}, 'legacy')
82 sorted_true = node.addmultisigaddress(2, addresses, {"sort": True}, 'legacy')
83
84 assert_equal(sorted_default, sorted_false)
85 assert_equal("2N6dne8yzh13wsRJxCcMgCYNeN9fxKWNHt8", sorted_default['address'])
86 assert_equal("2MsJ2YhGewgDPGEQk4vahGs4wRikJXpRRtU", sorted_true['address'])
87
88 sorted_default = node.addmultisigaddress(2, addresses, {'address_type': 'legacy'})
89 sorted_false = node.addmultisigaddress(2, addresses, {'address_type': 'legacy', "sort": False})
90 sorted_true = node.addmultisigaddress(2, addresses, {'address_type': 'legacy', "sort": True})
91
92 assert_equal(sorted_default, sorted_false)
93 assert_equal("2N6dne8yzh13wsRJxCcMgCYNeN9fxKWNHt8", sorted_default['address'])
94 assert_equal("2MsJ2YhGewgDPGEQk4vahGs4wRikJXpRRtU", sorted_true['address'])
95
96 assert_raises_rpc_error(-1, "address_type provided in both options and 4th parameter", node.addmultisigaddress, 2, addresses, {"address_type": 'legacy'}, 'bech32')
97
98 def test_sort_multisig_with_uncompressed_hash160(self, node):
99 node.importpubkey("02632b12f4ac5b1d1b72b2a3b508c19172de44f6f46bcee50ba33f3f9291e47ed0")
100 node.importpubkey("04dd4fe618a8ad14732f8172fe7c9c5e76dd18c2cc501ef7f86e0f4e285ca8b8b32d93df2f4323ebb02640fa6b975b2e63ab3c9d6979bc291193841332442cc6ad")
101 address = "2MxvEpFdXeEDbnz8MbRwS23kDZC8tzQ9NjK"
102
103 addresses = [
104 "msDoRfEfZQFaQNfAEWyqf69H99yntZoBbG",
105 "myrfasv56W7579LpepuRy7KFhVhaWsJYS8",
106 ]
107 default = self.nodes[0].addmultisigaddress(2, addresses, {'address_type': 'legacy'})
108 assert_equal(address, default['address'])
109
110 unsorted = self.nodes[0].addmultisigaddress(2, addresses, {'address_type': 'legacy', "sort": False})
111 assert_equal(address, unsorted['address'])
112
113 assert_raises_rpc_error(-1, "Compressed key required for BIP67: myrfasv56W7579LpepuRy7KFhVhaWsJYS8", node.addmultisigaddress, 2, addresses, {"sort": True})
114
115 def run_test(self):
116 # Check that there's no UTXO on the node
117 node = self.nodes[0]
118 assert_equal(len(node.listunspent()), 0)
119
120 self.log.info("Checking listlabels' invalid parameters")
121 assert_raises_rpc_error(-8, "Invalid 'purpose' argument, must be a known purpose string, typically 'send', or 'receive'.", node.listlabels, "notavalidpurpose")
122 assert_raises_rpc_error(-8, "Invalid 'purpose' argument, must be a known purpose string, typically 'send', or 'receive'.", node.listlabels, "unknown")
123
124 # Note each time we call generate, all generated coins go into
125 # the same address, so we call twice to get two addresses w/50 each
126 self.generatetoaddress(node, nblocks=1, address=node.getnewaddress(label='coinbase'))
127 self.generatetoaddress(node, nblocks=COINBASE_MATURITY + 1, address=node.getnewaddress(label='coinbase'))
128 assert_equal(node.getbalance(), 100)
129
130 # there should be 2 address groups
131 # each with 1 address with a balance of 50 Limenkas
132 address_groups = node.listaddressgroupings()
133 assert_equal(len(address_groups), 2)
134 # the addresses aren't linked now, but will be after we send to the
135 # common address
136 linked_addresses = set()
137 for address_group in address_groups:
138 assert_equal(len(address_group), 1)
139 assert_equal(len(address_group[0]), 3)
140 assert_equal(address_group[0][1], 50)
141 assert_equal(address_group[0][2], 'coinbase')
142 linked_addresses.add(address_group[0][0])
143
144 # send 50 from each address to a third address not in this wallet
145 common_address = "msf4WtN1YQKXvNtvdFYt9JBnUD2FB41kjr"
146 node.sendmany(
147 amounts={common_address: 100},
148 subtractfeefrom=[common_address],
149 minconf=1,
150 )
151 # there should be 1 address group, with the previously
152 # unlinked addresses now linked (they both have 0 balance)
153 address_groups = node.listaddressgroupings()
154 assert_equal(len(address_groups), 1)
155 assert_equal(len(address_groups[0]), 2)
156 assert_equal(set([a[0] for a in address_groups[0]]), linked_addresses)
157 assert_equal([a[1] for a in address_groups[0]], [0, 0])
158
159 self.generate(node, 1)
160
161 # we want to reset so that the "" label has what's expected.
162 # otherwise we're off by exactly the fee amount as that's mined
163 # and matures in the next 100 blocks
164 amount_to_send = 1.0
165
166 # Create labels and make sure subsequent label API calls
167 # recognize the label/address associations.
168 labels = [Label(name) for name in ("a", "b", "c", "d", "e")]
169 for label in labels:
170 address = node.getnewaddress(label.name)
171 label.add_receive_address(address)
172 label.verify(node)
173
174 # Check listlabels when passing 'purpose'
175 node2_addr = self.nodes[1].getnewaddress()
176 node.setlabel(node2_addr, "node2_addr")
177 assert_equal(node.listlabels(purpose="send"), ["node2_addr"])
178 assert_equal(node.listlabels(purpose="receive"), sorted(['coinbase'] + [label.name for label in labels]))
179
180 # Check all labels are returned by listlabels.
181 assert_equal(node.listlabels(), sorted(['coinbase'] + [label.name for label in labels] + ["node2_addr"]))
182
183 # Send a transaction to each label.
184 for label in labels:
185 node.sendtoaddress(label.addresses[0], amount_to_send)
186 label.verify(node)
187
188 # Check the amounts received.
189 self.generate(node, 1)
190 for label in labels:
191 assert_equal(
192 node.getreceivedbyaddress(label.addresses[0]), amount_to_send)
193 assert_equal(node.getreceivedbylabel(label.name), amount_to_send)
194
195 for i, label in enumerate(labels):
196 to_label = labels[(i + 1) % len(labels)]
197 node.sendtoaddress(to_label.addresses[0], amount_to_send)
198 self.generate(node, 1)
199 for label in labels:
200 address = node.getnewaddress(label.name)
201 label.add_receive_address(address)
202 label.verify(node)
203 assert_equal(node.getreceivedbylabel(label.name), 2)
204 label.verify(node)
205 self.generate(node, COINBASE_MATURITY + 1)
206
207 # Check that setlabel can assign a label to a new unused address.
208 for label in labels:
209 address = node.getnewaddress()
210 node.setlabel(address, label.name)
211 label.add_address(address)
212 label.verify(node)
213 assert_raises_rpc_error(-11, "No addresses with label", node.getaddressesbylabel, "")
214
215 # Check that addmultisigaddress can assign labels.
216 if not self.options.descriptors:
217 for label in labels:
218 addresses = []
219 for _ in range(10):
220 addresses.append(node.getnewaddress())
221 multisig_address = node.addmultisigaddress(5, addresses, label.name)['address']
222 label.add_address(multisig_address)
223 label.purpose[multisig_address] = "send"
224 label.verify(node)
225 self.generate(node, COINBASE_MATURITY + 1)
226
227 # Check that setlabel can change the label of an address from a
228 # different label.
229 change_label(node, labels[0].addresses[0], labels[0], labels[1])
230
231 # Check that setlabel can set the label of an address already
232 # in the label. This is a no-op.
233 change_label(node, labels[2].addresses[0], labels[2], labels[2])
234
235 self.invalid_label_name_test()
236
237 if not self.options.descriptors:
238 self.test_sort_multisig(node)
239 self.test_sort_multisig_with_uncompressed_hash160(node)
240
241 if self.options.descriptors:
242 # This is a descriptor wallet test because of segwit v1+ addresses
243 self.log.info('Check watchonly labels')
244 node.createwallet(wallet_name='watch_only', disable_private_keys=True)
245 wallet_watch_only = node.get_wallet_rpc('watch_only')
246 BECH32_VALID = {
247 '✔️_VER15_PROG40': 'bcrt10qqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqxkg7fn',
248 '✔️_VER16_PROG03': 'bcrt1sqqqqq8uhdgr',
249 '✔️_VER16_PROB02': 'bcrt1sqqqq4wstyw',
250 }
251 BECH32_INVALID = {
252 '❌_VER15_PROG41': 'bcrt1sqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqajlxj8',
253 '❌_VER16_PROB01': 'bcrt1sqq5r4036',
254 }
255 for l in BECH32_VALID:
256 ad = BECH32_VALID[l]
257 wallet_watch_only.importaddress(label=l, rescan=False, address=ad)
258 self.generatetoaddress(node, 1, ad)
259 assert_equal(wallet_watch_only.getaddressesbylabel(label=l), {ad: {'purpose': 'receive'}})
260 assert_equal(wallet_watch_only.getreceivedbylabel(label=l), 0)
261 for l in BECH32_INVALID:
262 ad = BECH32_INVALID[l]
263 assert_raises_rpc_error(
264 -5,
265 "Invalid Limenka address or script",
266 lambda: wallet_watch_only.importaddress(label=l, rescan=False, address=ad),
267 )
268
269
270 class Label:
271 def __init__(self, name):
272 # Label name
273 self.name = name
274 # Current receiving address associated with this label.
275 self.receive_address = None
276 # List of all addresses assigned with this label
277 self.addresses = []
278 # Map of address to address purpose
279 self.purpose = defaultdict(lambda: "receive")
280
281 def add_address(self, address):
282 assert_equal(address not in self.addresses, True)
283 self.addresses.append(address)
284
285 def add_receive_address(self, address):
286 self.add_address(address)
287
288 def verify(self, node):
289 if self.receive_address is not None:
290 assert self.receive_address in self.addresses
291 for address in self.addresses:
292 test_address(node, address, labels=[self.name])
293 assert self.name in node.listlabels()
294 assert_equal(
295 node.getaddressesbylabel(self.name),
296 {address: {"purpose": self.purpose[address]} for address in self.addresses})
297
298 def change_label(node, address, old_label, new_label):
299 assert_equal(address in old_label.addresses, True)
300 node.setlabel(address, new_label.name)
301
302 old_label.addresses.remove(address)
303 new_label.add_address(address)
304
305 old_label.verify(node)
306 new_label.verify(node)
307
308 if __name__ == '__main__':
309 WalletLabelsTest(__file__).main()
310