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
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
|
#!/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 listtransactions API."""
from decimal import Decimal
import os
import shutil
from test_framework.messages import (
COIN,
tx_from_hex,
)
from test_framework.test_framework import BitcoinTestFramework
from test_framework.util import (
assert_array_result,
assert_equal,
assert_raises_rpc_error,
)
class ListTransactionsTest(BitcoinTestFramework):
def set_test_params(self):
self.num_nodes = 3
# This test isn't testing txn relay/timing, so set whitelist on the
# peers for instant txn relay. This speeds up the test run time 2-3x.
self.extra_args = [["-whitelist=noban@127.0.0.1", "-walletrbf=0"]] * self.num_nodes
def skip_test_if_missing_module(self):
self.skip_if_no_wallet()
def run_test(self):
self.log.info("Test simple send from node0 to node1")
txid = self.nodes[0].sendtoaddress(self.nodes[1].getnewaddress(), 0.1)
self.sync_all()
assert_array_result(self.nodes[0].listtransactions(),
{"txid": txid},
{"category": "send", "amount": Decimal("-0.1"), "confirmations": 0, "trusted": True})
assert_array_result(self.nodes[1].listtransactions(),
{"txid": txid},
{"category": "receive", "amount": Decimal("0.1"), "confirmations": 0, "trusted": False})
self.log.info("Test confirmations change after mining a block")
blockhash = self.generate(self.nodes[0], 1)[0]
blockheight = self.nodes[0].getblockheader(blockhash)['height']
assert_array_result(self.nodes[0].listtransactions(),
{"txid": txid},
{"category": "send", "amount": Decimal("-0.1"), "confirmations": 1, "blockhash": blockhash, "blockheight": blockheight})
assert_array_result(self.nodes[1].listtransactions(),
{"txid": txid},
{"category": "receive", "amount": Decimal("0.1"), "confirmations": 1, "blockhash": blockhash, "blockheight": blockheight})
self.log.info("Test send-to-self on node0")
txid = self.nodes[0].sendtoaddress(self.nodes[0].getnewaddress(), 0.2)
assert_array_result(self.nodes[0].listtransactions(),
{"txid": txid, "category": "send"},
{"amount": Decimal("-0.2")})
assert_array_result(self.nodes[0].listtransactions(),
{"txid": txid, "category": "receive"},
{"amount": Decimal("0.2")})
self.log.info("Test sendmany from node1: twice to self, twice to node0")
send_to = {self.nodes[0].getnewaddress(): 0.11,
self.nodes[1].getnewaddress(): 0.22,
self.nodes[0].getnewaddress(): 0.33,
self.nodes[1].getnewaddress(): 0.44}
txid = self.nodes[1].sendmany("", send_to)
self.sync_all()
assert_array_result(self.nodes[1].listtransactions(),
{"category": "send", "amount": Decimal("-0.11")},
{"txid": txid})
assert_array_result(self.nodes[0].listtransactions(),
{"category": "receive", "amount": Decimal("0.11")},
{"txid": txid})
assert_array_result(self.nodes[1].listtransactions(),
{"category": "send", "amount": Decimal("-0.22")},
{"txid": txid})
assert_array_result(self.nodes[1].listtransactions(),
{"category": "receive", "amount": Decimal("0.22")},
{"txid": txid})
assert_array_result(self.nodes[1].listtransactions(),
{"category": "send", "amount": Decimal("-0.33")},
{"txid": txid})
assert_array_result(self.nodes[0].listtransactions(),
{"category": "receive", "amount": Decimal("0.33")},
{"txid": txid})
assert_array_result(self.nodes[1].listtransactions(),
{"category": "send", "amount": Decimal("-0.44")},
{"txid": txid})
assert_array_result(self.nodes[1].listtransactions(),
{"category": "receive", "amount": Decimal("0.44")},
{"txid": txid})
if not self.options.descriptors:
# include_watchonly is a legacy wallet feature, so don't test it for descriptor wallets
self.log.info("Test 'include_watchonly' feature (legacy wallet)")
pubkey = self.nodes[1].getaddressinfo(self.nodes[1].getnewaddress())['pubkey']
multisig = self.nodes[1].createmultisig(1, [pubkey])
self.nodes[0].importaddress(multisig["redeemScript"], "watchonly", False, True)
txid = self.nodes[1].sendtoaddress(multisig["address"], 0.1)
self.generate(self.nodes[1], 1)
assert_equal(len(self.nodes[0].listtransactions(label="watchonly", include_watchonly=True)), 1)
assert_equal(len(self.nodes[0].listtransactions(dummy="watchonly", include_watchonly=True)), 1)
assert len(self.nodes[0].listtransactions(label="watchonly", count=100, include_watchonly=False)) == 0
assert_array_result(self.nodes[0].listtransactions(label="watchonly", count=100, include_watchonly=True),
{"category": "receive", "amount": Decimal("0.1")},
{"txid": txid, "label": "watchonly"})
self.run_rbf_opt_in_test()
self.run_externally_generated_address_test()
self.run_invalid_parameters_test()
self.test_op_return()
def run_rbf_opt_in_test(self):
"""Test the opt-in-rbf flag for sent and received transactions."""
def is_opt_in(node, txid):
"""Check whether a transaction signals opt-in RBF itself."""
rawtx = node.getrawtransaction(txid, 1)
for x in rawtx["vin"]:
if x["sequence"] < 0xfffffffe:
return True
return False
def get_unconfirmed_utxo_entry(node, txid_to_match):
"""Find an unconfirmed output matching a certain txid."""
utxo = node.listunspent(0, 0)
for i in utxo:
if i["txid"] == txid_to_match:
return i
return None
self.log.info("Test txs w/o opt-in RBF (bip125-replaceable=no)")
# Chain a few transactions that don't opt in.
txid_1 = self.nodes[0].sendtoaddress(self.nodes[1].getnewaddress(), 1)
assert not is_opt_in(self.nodes[0], txid_1)
assert_array_result(self.nodes[0].listtransactions(), {"txid": txid_1}, {"bip125-replaceable": "no"})
self.sync_mempools()
assert_array_result(self.nodes[1].listtransactions(), {"txid": txid_1}, {"bip125-replaceable": "no"})
# Tx2 will build off tx1, still not opting in to RBF.
utxo_to_use = get_unconfirmed_utxo_entry(self.nodes[0], txid_1)
assert_equal(utxo_to_use["safe"], True)
utxo_to_use = get_unconfirmed_utxo_entry(self.nodes[1], txid_1)
assert_equal(utxo_to_use["safe"], False)
# Create tx2 using createrawtransaction
inputs = [{"txid": utxo_to_use["txid"], "vout": utxo_to_use["vout"]}]
outputs = {self.nodes[0].getnewaddress(): 0.999}
tx2 = self.nodes[1].createrawtransaction(inputs=inputs, outputs=outputs, replaceable=False)
tx2_signed = self.nodes[1].signrawtransactionwithwallet(tx2)["hex"]
txid_2 = self.nodes[1].sendrawtransaction(tx2_signed)
# ...and check the result
assert not is_opt_in(self.nodes[1], txid_2)
assert_array_result(self.nodes[1].listtransactions(), {"txid": txid_2}, {"bip125-replaceable": "no"})
self.sync_mempools()
assert_array_result(self.nodes[0].listtransactions(), {"txid": txid_2}, {"bip125-replaceable": "no"})
self.log.info("Test txs with opt-in RBF (bip125-replaceable=yes)")
# Tx3 will opt-in to RBF
utxo_to_use = get_unconfirmed_utxo_entry(self.nodes[0], txid_2)
inputs = [{"txid": txid_2, "vout": utxo_to_use["vout"]}]
outputs = {self.nodes[1].getnewaddress(): 0.998}
tx3 = self.nodes[0].createrawtransaction(inputs, outputs)
tx3_modified = tx_from_hex(tx3)
tx3_modified.vin[0].nSequence = 0
tx3 = tx3_modified.serialize().hex()
tx3_signed = self.nodes[0].signrawtransactionwithwallet(tx3)['hex']
txid_3 = self.nodes[0].sendrawtransaction(tx3_signed)
assert is_opt_in(self.nodes[0], txid_3)
assert_array_result(self.nodes[0].listtransactions(), {"txid": txid_3}, {"bip125-replaceable": "yes"})
self.sync_mempools()
assert_array_result(self.nodes[1].listtransactions(), {"txid": txid_3}, {"bip125-replaceable": "yes"})
# Tx4 will chain off tx3. Doesn't signal itself, but depends on one
# that does.
utxo_to_use = get_unconfirmed_utxo_entry(self.nodes[1], txid_3)
inputs = [{"txid": txid_3, "vout": utxo_to_use["vout"]}]
outputs = {self.nodes[0].getnewaddress(): 0.997}
tx4 = self.nodes[1].createrawtransaction(inputs=inputs, outputs=outputs, replaceable=False)
tx4_signed = self.nodes[1].signrawtransactionwithwallet(tx4)["hex"]
txid_4 = self.nodes[1].sendrawtransaction(tx4_signed)
assert not is_opt_in(self.nodes[1], txid_4)
assert_array_result(self.nodes[1].listtransactions(), {"txid": txid_4}, {"bip125-replaceable": "yes"})
self.sync_mempools()
assert_array_result(self.nodes[0].listtransactions(), {"txid": txid_4}, {"bip125-replaceable": "yes"})
self.log.info("Test tx with unknown RBF state (bip125-replaceable=unknown)")
# Replace tx3, and check that tx4 becomes unknown
tx3_b = tx3_modified
tx3_b.vout[0].nValue -= int(Decimal("0.004") * COIN) # bump the fee
tx3_b = tx3_b.serialize().hex()
tx3_b_signed = self.nodes[0].signrawtransactionwithwallet(tx3_b)['hex']
txid_3b = self.nodes[0].sendrawtransaction(tx3_b_signed, 0)
assert is_opt_in(self.nodes[0], txid_3b)
assert_array_result(self.nodes[0].listtransactions(), {"txid": txid_4}, {"bip125-replaceable": "unknown"})
self.sync_mempools()
assert_array_result(self.nodes[1].listtransactions(), {"txid": txid_4}, {"bip125-replaceable": "unknown"})
self.log.info("Test bip125-replaceable status with gettransaction RPC")
for n in self.nodes[0:2]:
assert_equal(n.gettransaction(txid_1)["bip125-replaceable"], "no")
assert_equal(n.gettransaction(txid_2)["bip125-replaceable"], "no")
assert_equal(n.gettransaction(txid_3)["bip125-replaceable"], "yes")
assert_equal(n.gettransaction(txid_3b)["bip125-replaceable"], "yes")
assert_equal(n.gettransaction(txid_4)["bip125-replaceable"], "unknown")
self.log.info("Test bip125-replaceable status with listsinceblock")
for n in self.nodes[0:2]:
txs = {tx['txid']: tx['bip125-replaceable'] for tx in n.listsinceblock()['transactions']}
assert_equal(txs[txid_1], "no")
assert_equal(txs[txid_2], "no")
assert_equal(txs[txid_3], "yes")
assert_equal(txs[txid_3b], "yes")
assert_equal(txs[txid_4], "unknown")
self.log.info("Test mined transactions are no longer bip125-replaceable")
self.generate(self.nodes[0], 1)
assert txid_3b not in self.nodes[0].getrawmempool()
assert_equal(self.nodes[0].gettransaction(txid_3b)["bip125-replaceable"], "no")
assert_equal(self.nodes[0].gettransaction(txid_4)["bip125-replaceable"], "unknown")
def run_externally_generated_address_test(self):
"""Test behavior when receiving address is not in the address book."""
self.log.info("Setup the same wallet on two nodes")
# refill keypool otherwise the second node wouldn't recognize addresses generated on the first nodes
self.nodes[0].keypoolrefill(1000)
self.stop_nodes()
wallet0 = os.path.join(self.nodes[0].datadir, self.chain, self.default_wallet_name, "wallet.dat")
wallet2 = os.path.join(self.nodes[2].datadir, self.chain, self.default_wallet_name, "wallet.dat")
shutil.copyfile(wallet0, wallet2)
self.start_nodes()
# reconnect nodes
self.connect_nodes(0, 1)
self.connect_nodes(1, 2)
self.connect_nodes(2, 0)
addr1 = self.nodes[0].getnewaddress("pizza1", 'legacy')
addr2 = self.nodes[0].getnewaddress("pizza2", 'p2sh-segwit')
addr3 = self.nodes[0].getnewaddress("pizza3", 'bech32')
self.log.info("Send to externally generated addresses")
# send to an address beyond the next to be generated to test the keypool gap
self.nodes[1].sendtoaddress(addr3, "0.001")
self.generate(self.nodes[1], 1)
# send to an address that is already marked as used due to the keypool gap mechanics
self.nodes[1].sendtoaddress(addr2, "0.001")
self.generate(self.nodes[1], 1)
# send to self transaction
self.nodes[0].sendtoaddress(addr1, "0.001")
self.generate(self.nodes[0], 1)
self.log.info("Verify listtransactions is the same regardless of where the address was generated")
transactions0 = self.nodes[0].listtransactions()
transactions2 = self.nodes[2].listtransactions()
# normalize results: remove fields that normally could differ and sort
def normalize_list(txs):
for tx in txs:
tx.pop('label', None)
tx.pop('time', None)
tx.pop('timereceived', None)
txs.sort(key=lambda x: x['txid'])
normalize_list(transactions0)
normalize_list(transactions2)
assert_equal(transactions0, transactions2)
self.log.info("Verify labels are persistent on the node that generated the addresses")
assert_equal(['pizza1'], self.nodes[0].getaddressinfo(addr1)['labels'])
assert_equal(['pizza2'], self.nodes[0].getaddressinfo(addr2)['labels'])
assert_equal(['pizza3'], self.nodes[0].getaddressinfo(addr3)['labels'])
def run_invalid_parameters_test(self):
self.log.info("Test listtransactions RPC parameter validity")
assert_raises_rpc_error(-8, 'Label argument must be a valid label name or "*".', self.nodes[0].listtransactions, label="")
self.nodes[0].listtransactions(label="*")
assert_raises_rpc_error(-8, "Negative count", self.nodes[0].listtransactions, count=-1)
assert_raises_rpc_error(-8, "Negative from", self.nodes[0].listtransactions, skip=-1)
def test_op_return(self):
"""Test if OP_RETURN outputs will be displayed correctly."""
raw_tx = self.nodes[0].createrawtransaction([], [{'data': 'aa'}])
funded_tx = self.nodes[0].fundrawtransaction(raw_tx)
signed_tx = self.nodes[0].signrawtransactionwithwallet(funded_tx['hex'])
tx_id = self.nodes[0].sendrawtransaction(signed_tx['hex'])
op_ret_tx = [tx for tx in self.nodes[0].listtransactions() if tx['txid'] == tx_id][0]
assert 'address' not in op_ret_tx
if __name__ == '__main__':
ListTransactionsTest().main()
|
