1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
|
#!/usr/bin/env python3
# Copyright (c) 2014-2021 The Bitcoin Core developers
# Distributed under the MIT software license, see the accompanying
# file COPYING or http://www.opensource.org/licenses/mit-license.php.
"""Test the listreceivedbyaddress, listreceivedbylabel, getreceivedybaddress, and getreceivedbylabel RPCs."""
from decimal import Decimal
from test_framework.blocktools import COINBASE_MATURITY
from test_framework.test_framework import BitcoinTestFramework
from test_framework.util import (
assert_array_result,
assert_equal,
assert_raises_rpc_error,
)
from test_framework.wallet_util import test_address
class ReceivedByTest(BitcoinTestFramework):
def set_test_params(self):
self.num_nodes = 2
def skip_test_if_missing_module(self):
self.skip_if_no_wallet()
self.skip_if_no_cli()
def run_test(self):
# save the number of coinbase reward addresses so far
num_cb_reward_addresses = len(self.nodes[1].listreceivedbyaddress(minconf=0, include_empty=True, include_watchonly=True))
self.log.info("listreceivedbyaddress Test")
# Send from node 0 to 1
addr = self.nodes[1].getnewaddress()
txid = self.nodes[0].sendtoaddress(addr, 0.1)
self.sync_all()
# Check not listed in listreceivedbyaddress because has 0 confirmations
assert_array_result(self.nodes[1].listreceivedbyaddress(),
{"address": addr},
{},
True)
# Bury Tx under 10 block so it will be returned by listreceivedbyaddress
self.generate(self.nodes[1], 10)
assert_array_result(self.nodes[1].listreceivedbyaddress(),
{"address": addr},
{"address": addr, "label": "", "amount": Decimal("0.1"), "confirmations": 10, "txids": [txid, ]})
# With min confidence < 10
assert_array_result(self.nodes[1].listreceivedbyaddress(5),
{"address": addr},
{"address": addr, "label": "", "amount": Decimal("0.1"), "confirmations": 10, "txids": [txid, ]})
# With min confidence > 10, should not find Tx
assert_array_result(self.nodes[1].listreceivedbyaddress(11), {"address": addr}, {}, True)
# Empty Tx
empty_addr = self.nodes[1].getnewaddress()
assert_array_result(self.nodes[1].listreceivedbyaddress(0, True),
{"address": empty_addr},
{"address": empty_addr, "label": "", "amount": 0, "confirmations": 0, "txids": []})
# No returned addy should be a change addr
for node in self.nodes:
for addr_obj in node.listreceivedbyaddress():
assert_equal(node.getaddressinfo(addr_obj["address"])["ischange"], False)
# Test Address filtering
# Only on addr
expected = {"address": addr, "label": "", "amount": Decimal("0.1"), "confirmations": 10, "txids": [txid, ]}
res = self.nodes[1].listreceivedbyaddress(minconf=0, include_empty=True, include_watchonly=True, address_filter=addr)
assert_array_result(res, {"address": addr}, expected)
assert_equal(len(res), 1)
# Test for regression on CLI calls with address string (#14173)
cli_res = self.nodes[1].cli.listreceivedbyaddress(0, True, True, addr)
assert_array_result(cli_res, {"address": addr}, expected)
assert_equal(len(cli_res), 1)
# Error on invalid address
assert_raises_rpc_error(-4, "address_filter parameter was invalid", self.nodes[1].listreceivedbyaddress, minconf=0, include_empty=True, include_watchonly=True, address_filter="bamboozling")
# Another address receive money
res = self.nodes[1].listreceivedbyaddress(0, True, True)
assert_equal(len(res), 2 + num_cb_reward_addresses) # Right now 2 entries
other_addr = self.nodes[1].getnewaddress()
txid2 = self.nodes[0].sendtoaddress(other_addr, 0.1)
self.generate(self.nodes[0], 1)
# Same test as above should still pass
expected = {"address": addr, "label": "", "amount": Decimal("0.1"), "confirmations": 11, "txids": [txid, ]}
res = self.nodes[1].listreceivedbyaddress(0, True, True, addr)
assert_array_result(res, {"address": addr}, expected)
assert_equal(len(res), 1)
# Same test as above but with other_addr should still pass
expected = {"address": other_addr, "label": "", "amount": Decimal("0.1"), "confirmations": 1, "txids": [txid2, ]}
res = self.nodes[1].listreceivedbyaddress(0, True, True, other_addr)
assert_array_result(res, {"address": other_addr}, expected)
assert_equal(len(res), 1)
# Should be two entries though without filter
res = self.nodes[1].listreceivedbyaddress(0, True, True)
assert_equal(len(res), 3 + num_cb_reward_addresses) # Became 3 entries
# Not on random addr
other_addr = self.nodes[0].getnewaddress() # note on node[0]! just a random addr
res = self.nodes[1].listreceivedbyaddress(0, True, True, other_addr)
assert_equal(len(res), 0)
self.log.info("getreceivedbyaddress Test")
# Send from node 0 to 1
addr = self.nodes[1].getnewaddress()
txid = self.nodes[0].sendtoaddress(addr, 0.1)
self.sync_all()
# Check balance is 0 because of 0 confirmations
balance = self.nodes[1].getreceivedbyaddress(addr)
assert_equal(balance, Decimal("0.0"))
# Check balance is 0.1
balance = self.nodes[1].getreceivedbyaddress(addr, 0)
assert_equal(balance, Decimal("0.1"))
# Bury Tx under 10 block so it will be returned by the default getreceivedbyaddress
self.generate(self.nodes[1], 10)
balance = self.nodes[1].getreceivedbyaddress(addr)
assert_equal(balance, Decimal("0.1"))
# Trying to getreceivedby for an address the wallet doesn't own should return an error
assert_raises_rpc_error(-4, "Address not found in wallet", self.nodes[0].getreceivedbyaddress, addr)
self.log.info("listreceivedbylabel + getreceivedbylabel Test")
# set pre-state
label = ''
address = self.nodes[1].getnewaddress()
test_address(self.nodes[1], address, labels=[label])
received_by_label_json = [r for r in self.nodes[1].listreceivedbylabel() if r["label"] == label][0]
balance_by_label = self.nodes[1].getreceivedbylabel(label)
txid = self.nodes[0].sendtoaddress(addr, 0.1)
self.sync_all()
# getreceivedbylabel returns an error if the wallet doesn't own the label
assert_raises_rpc_error(-4, "Label not found in wallet", self.nodes[0].getreceivedbylabel, "dummy")
# listreceivedbylabel should return received_by_label_json because of 0 confirmations
assert_array_result(self.nodes[1].listreceivedbylabel(),
{"label": label},
received_by_label_json)
# getreceivedbyaddress should return same balance because of 0 confirmations
balance = self.nodes[1].getreceivedbylabel(label)
assert_equal(balance, balance_by_label)
self.generate(self.nodes[1], 10)
# listreceivedbylabel should return updated received list
assert_array_result(self.nodes[1].listreceivedbylabel(),
{"label": label},
{"label": received_by_label_json["label"], "amount": (received_by_label_json["amount"] + Decimal("0.1"))})
# getreceivedbylabel should return updated receive total
balance = self.nodes[1].getreceivedbylabel(label)
assert_equal(balance, balance_by_label + Decimal("0.1"))
# Create a new label named "mynewlabel" that has a 0 balance
address = self.nodes[1].getnewaddress()
self.nodes[1].setlabel(address, "mynewlabel")
received_by_label_json = [r for r in self.nodes[1].listreceivedbylabel(0, True) if r["label"] == "mynewlabel"][0]
# Test includeempty of listreceivedbylabel
assert_equal(received_by_label_json["amount"], Decimal("0.0"))
# Test getreceivedbylabel for 0 amount labels
balance = self.nodes[1].getreceivedbylabel("mynewlabel")
assert_equal(balance, Decimal("0.0"))
self.log.info("Tests for including coinbase outputs")
# Generate block reward to address with label
label = "label"
address = self.nodes[0].getnewaddress(label)
reward = Decimal("25")
self.generatetoaddress(self.nodes[0], 1, address)
hash = self.nodes[0].getbestblockhash()
self.log.info("getreceivedbyaddress returns nothing with defaults")
balance = self.nodes[0].getreceivedbyaddress(address)
assert_equal(balance, 0)
self.log.info("getreceivedbyaddress returns block reward when including immature coinbase")
balance = self.nodes[0].getreceivedbyaddress(address=address, include_immature_coinbase=True)
assert_equal(balance, reward)
self.log.info("getreceivedbylabel returns nothing with defaults")
balance = self.nodes[0].getreceivedbylabel("label")
assert_equal(balance, 0)
self.log.info("getreceivedbylabel returns block reward when including immature coinbase")
balance = self.nodes[0].getreceivedbylabel(label="label", include_immature_coinbase=True)
assert_equal(balance, reward)
self.log.info("listreceivedbyaddress does not include address with defaults")
assert_array_result(self.nodes[0].listreceivedbyaddress(),
{"address": address},
{}, True)
self.log.info("listreceivedbyaddress includes address when including immature coinbase")
assert_array_result(self.nodes[0].listreceivedbyaddress(minconf=1, include_immature_coinbase=True),
{"address": address},
{"address": address, "amount": reward})
self.log.info("listreceivedbylabel does not include label with defaults")
assert_array_result(self.nodes[0].listreceivedbylabel(),
{"label": label},
{}, True)
self.log.info("listreceivedbylabel includes label when including immature coinbase")
assert_array_result(self.nodes[0].listreceivedbylabel(minconf=1, include_immature_coinbase=True),
{"label": label},
{"label": label, "amount": reward})
self.log.info("Generate 100 more blocks")
self.generate(self.nodes[0], COINBASE_MATURITY)
self.log.info("getreceivedbyaddress returns reward with defaults")
balance = self.nodes[0].getreceivedbyaddress(address)
assert_equal(balance, reward)
self.log.info("getreceivedbylabel returns reward with defaults")
balance = self.nodes[0].getreceivedbylabel("label")
assert_equal(balance, reward)
self.log.info("listreceivedbyaddress includes address with defaults")
assert_array_result(self.nodes[0].listreceivedbyaddress(),
{"address": address},
{"address": address, "amount": reward})
self.log.info("listreceivedbylabel includes label with defaults")
assert_array_result(self.nodes[0].listreceivedbylabel(),
{"label": label},
{"label": label, "amount": reward})
self.log.info("Invalidate block that paid to address")
self.nodes[0].invalidateblock(hash)
self.log.info("getreceivedbyaddress does not include invalidated block when minconf is 0 when including immature coinbase")
balance = self.nodes[0].getreceivedbyaddress(address=address, minconf=0, include_immature_coinbase=True)
assert_equal(balance, 0)
self.log.info("getreceivedbylabel does not include invalidated block when minconf is 0 when including immature coinbase")
balance = self.nodes[0].getreceivedbylabel(label="label", minconf=0, include_immature_coinbase=True)
assert_equal(balance, 0)
self.log.info("listreceivedbyaddress does not include invalidated block when minconf is 0 when including immature coinbase")
assert_array_result(self.nodes[0].listreceivedbyaddress(minconf=0, include_immature_coinbase=True),
{"address": address},
{}, True)
self.log.info("listreceivedbylabel does not include invalidated block when minconf is 0 when including immature coinbase")
assert_array_result(self.nodes[0].listreceivedbylabel(minconf=0, include_immature_coinbase=True),
{"label": label},
{}, True)
if __name__ == '__main__':
ReceivedByTest().main()
|
