wallet_listtransactions.py raw
1 #!/usr/bin/env python3
2 # Copyright (c) 2014-present 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 listtransactions API."""
6
7 from decimal import Decimal
8 import time
9 import os
10 import shutil
11
12 from test_framework.blocktools import MAX_FUTURE_BLOCK_TIME
13 from test_framework.messages import (
14 COIN,
15 tx_from_hex,
16 )
17 from test_framework.test_framework import LimenkaTestFramework
18 from test_framework.util import (
19 assert_array_result,
20 assert_equal,
21 assert_raises_rpc_error,
22 find_vout_for_address,
23 )
24 from test_framework.wallet_util import get_generate_key
25
26
27 class ListTransactionsTest(LimenkaTestFramework):
28 def add_options(self, parser):
29 self.add_wallet_options(parser)
30
31 def set_test_params(self):
32 self.num_nodes = 3
33 # whitelist peers to speed up tx relay / mempool sync
34 self.noban_tx_relay = True
35 self.extra_args = [["-walletrbf=0"]] * self.num_nodes
36
37 def skip_test_if_missing_module(self):
38 self.skip_if_no_wallet()
39
40 def run_test(self):
41 self.log.info("Test simple send from node0 to node1")
42 txid = self.nodes[0].sendtoaddress(self.nodes[1].getnewaddress(), 0.1)
43 self.sync_all()
44 assert_array_result(self.nodes[0].listtransactions(),
45 {"txid": txid},
46 {"category": "send", "amount": Decimal("-0.1"), "confirmations": 0, "trusted": True})
47 assert_array_result(self.nodes[1].listtransactions(),
48 {"txid": txid},
49 {"category": "receive", "amount": Decimal("0.1"), "confirmations": 0, "trusted": False})
50 self.log.info("Test confirmations change after mining a block")
51 blockhash = self.generate(self.nodes[0], 1)[0]
52 blockheight = self.nodes[0].getblockheader(blockhash)['height']
53 assert_array_result(self.nodes[0].listtransactions(),
54 {"txid": txid},
55 {"category": "send", "amount": Decimal("-0.1"), "confirmations": 1, "blockhash": blockhash, "blockheight": blockheight})
56 assert_array_result(self.nodes[1].listtransactions(),
57 {"txid": txid},
58 {"category": "receive", "amount": Decimal("0.1"), "confirmations": 1, "blockhash": blockhash, "blockheight": blockheight})
59
60 self.log.info("Test send-to-self on node0")
61 txid = self.nodes[0].sendtoaddress(self.nodes[0].getnewaddress(), 0.2)
62 assert_array_result(self.nodes[0].listtransactions(),
63 {"txid": txid, "category": "send"},
64 {"amount": Decimal("-0.2")})
65 assert_array_result(self.nodes[0].listtransactions(),
66 {"txid": txid, "category": "receive"},
67 {"amount": Decimal("0.2")})
68
69 self.log.info("Test sendmany from node1: twice to self, twice to node0")
70 send_to = {self.nodes[0].getnewaddress(): 0.11,
71 self.nodes[1].getnewaddress(): 0.22,
72 self.nodes[0].getnewaddress(): 0.33,
73 self.nodes[1].getnewaddress(): 0.44}
74 txid = self.nodes[1].sendmany("", send_to)
75 self.sync_all()
76 assert_array_result(self.nodes[1].listtransactions(),
77 {"category": "send", "amount": Decimal("-0.11")},
78 {"txid": txid})
79 assert_array_result(self.nodes[0].listtransactions(),
80 {"category": "receive", "amount": Decimal("0.11")},
81 {"txid": txid})
82 assert_array_result(self.nodes[1].listtransactions(),
83 {"category": "send", "amount": Decimal("-0.22")},
84 {"txid": txid})
85 assert_array_result(self.nodes[1].listtransactions(),
86 {"category": "receive", "amount": Decimal("0.22")},
87 {"txid": txid})
88 assert_array_result(self.nodes[1].listtransactions(),
89 {"category": "send", "amount": Decimal("-0.33")},
90 {"txid": txid})
91 assert_array_result(self.nodes[0].listtransactions(),
92 {"category": "receive", "amount": Decimal("0.33")},
93 {"txid": txid})
94 assert_array_result(self.nodes[1].listtransactions(),
95 {"category": "send", "amount": Decimal("-0.44")},
96 {"txid": txid})
97 assert_array_result(self.nodes[1].listtransactions(),
98 {"category": "receive", "amount": Decimal("0.44")},
99 {"txid": txid})
100
101 if not self.options.descriptors:
102 # include_watchonly is a legacy wallet feature, so don't test it for descriptor wallets
103 self.log.info("Test 'include_watchonly' feature (legacy wallet)")
104 pubkey = self.nodes[1].getaddressinfo(self.nodes[1].getnewaddress())['pubkey']
105 multisig = self.nodes[1].createmultisig(1, [pubkey])
106 self.nodes[0].importaddress(multisig["redeemScript"], "watchonly", False, True)
107 txid = self.nodes[1].sendtoaddress(multisig["address"], 0.1)
108 self.generate(self.nodes[1], 1)
109 assert_equal(len(self.nodes[0].listtransactions(label="watchonly", include_watchonly=True)), 1)
110 assert len(self.nodes[0].listtransactions(label="watchonly", count=100, include_watchonly=False)) == 0
111 assert_array_result(self.nodes[0].listtransactions(label="watchonly", count=100, include_watchonly=True),
112 {"category": "receive", "amount": Decimal("0.1")},
113 {"txid": txid, "label": "watchonly"})
114
115 self.run_rbf_opt_in_test()
116 self.run_externally_generated_address_test()
117 self.run_coinjoin_test()
118 self.run_invalid_parameters_test()
119 self.test_op_return()
120 self.test_listtransactions_display_in_mempool()
121 self.test_gettransaction_display_in_mempool()
122
123 self.test_from_me_status_change()
124
125 def run_rbf_opt_in_test(self):
126 """Test the opt-in-rbf flag for sent and received transactions."""
127
128 def is_opt_in(node, txid):
129 """Check whether a transaction signals opt-in RBF itself."""
130 rawtx = node.getrawtransaction(txid, 1)
131 for x in rawtx["vin"]:
132 if x["sequence"] < 0xfffffffe:
133 return True
134 return False
135
136 def get_unconfirmed_utxo_entry(node, txid_to_match):
137 """Find an unconfirmed output matching a certain txid."""
138 utxo = node.listunspent(0, 0)
139 for i in utxo:
140 if i["txid"] == txid_to_match:
141 return i
142 return None
143
144 self.log.info("Test txs w/o opt-in RBF (bip125-replaceable=no)")
145 # Chain a few transactions that don't opt in.
146 txid_1 = self.nodes[0].sendtoaddress(self.nodes[1].getnewaddress(), 1)
147 assert not is_opt_in(self.nodes[0], txid_1)
148 assert_array_result(self.nodes[0].listtransactions(), {"txid": txid_1}, {"bip125-replaceable": "no"})
149 self.sync_mempools()
150 assert_array_result(self.nodes[1].listtransactions(), {"txid": txid_1}, {"bip125-replaceable": "no"})
151
152 # Tx2 will build off tx1, still not opting in to RBF.
153 utxo_to_use = get_unconfirmed_utxo_entry(self.nodes[0], txid_1)
154 assert_equal(utxo_to_use["safe"], True)
155 utxo_to_use = get_unconfirmed_utxo_entry(self.nodes[1], txid_1)
156 assert_equal(utxo_to_use["safe"], False)
157
158 # Create tx2 using createrawtransaction
159 inputs = [{"txid": utxo_to_use["txid"], "vout": utxo_to_use["vout"]}]
160 outputs = {self.nodes[0].getnewaddress(): 0.999}
161 tx2 = self.nodes[1].createrawtransaction(inputs=inputs, outputs=outputs, replaceable=False)
162 tx2_signed = self.nodes[1].signrawtransactionwithwallet(tx2)["hex"]
163 txid_2 = self.nodes[1].sendrawtransaction(tx2_signed)
164
165 # ...and check the result
166 assert not is_opt_in(self.nodes[1], txid_2)
167 assert_array_result(self.nodes[1].listtransactions(), {"txid": txid_2}, {"bip125-replaceable": "no"})
168 self.sync_mempools()
169 assert_array_result(self.nodes[0].listtransactions(), {"txid": txid_2}, {"bip125-replaceable": "no"})
170
171 self.log.info("Test txs with opt-in RBF (bip125-replaceable=yes)")
172 # Tx3 will opt-in to RBF
173 utxo_to_use = get_unconfirmed_utxo_entry(self.nodes[0], txid_2)
174 inputs = [{"txid": txid_2, "vout": utxo_to_use["vout"]}]
175 outputs = {self.nodes[1].getnewaddress(): 0.998}
176 tx3 = self.nodes[0].createrawtransaction(inputs, outputs)
177 tx3_modified = tx_from_hex(tx3)
178 tx3_modified.vin[0].nSequence = 0
179 tx3 = tx3_modified.serialize().hex()
180 tx3_signed = self.nodes[0].signrawtransactionwithwallet(tx3)['hex']
181 txid_3 = self.nodes[0].sendrawtransaction(tx3_signed)
182
183 assert is_opt_in(self.nodes[0], txid_3)
184 assert_array_result(self.nodes[0].listtransactions(), {"txid": txid_3}, {"bip125-replaceable": "yes"})
185 self.sync_mempools()
186 assert_array_result(self.nodes[1].listtransactions(), {"txid": txid_3}, {"bip125-replaceable": "yes"})
187
188 # Tx4 will chain off tx3. Doesn't signal itself, but depends on one
189 # that does.
190 utxo_to_use = get_unconfirmed_utxo_entry(self.nodes[1], txid_3)
191 inputs = [{"txid": txid_3, "vout": utxo_to_use["vout"]}]
192 outputs = {self.nodes[0].getnewaddress(): 0.997}
193 tx4 = self.nodes[1].createrawtransaction(inputs=inputs, outputs=outputs, replaceable=False)
194 tx4_signed = self.nodes[1].signrawtransactionwithwallet(tx4)["hex"]
195 txid_4 = self.nodes[1].sendrawtransaction(tx4_signed)
196
197 assert not is_opt_in(self.nodes[1], txid_4)
198 assert_array_result(self.nodes[1].listtransactions(), {"txid": txid_4}, {"bip125-replaceable": "yes"})
199 self.sync_mempools()
200 assert_array_result(self.nodes[0].listtransactions(), {"txid": txid_4}, {"bip125-replaceable": "yes"})
201
202 self.log.info("Test tx with unknown RBF state (bip125-replaceable=unknown)")
203 # Replace tx3, and check that tx4 becomes unknown
204 tx3_b = tx3_modified
205 tx3_b.vout[0].nValue -= int(Decimal("0.004") * COIN) # bump the fee
206 tx3_b = tx3_b.serialize().hex()
207 tx3_b_signed = self.nodes[0].signrawtransactionwithwallet(tx3_b)['hex']
208 txid_3b = self.nodes[0].sendrawtransaction(tx3_b_signed, 0)
209 assert is_opt_in(self.nodes[0], txid_3b)
210
211 assert_array_result(self.nodes[0].listtransactions(), {"txid": txid_4}, {"bip125-replaceable": "unknown"})
212 self.sync_mempools()
213 assert_array_result(self.nodes[1].listtransactions(), {"txid": txid_4}, {"bip125-replaceable": "unknown"})
214
215 self.log.info("Test bip125-replaceable status with gettransaction RPC")
216 for n in self.nodes[0:2]:
217 assert_equal(n.gettransaction(txid_1)["bip125-replaceable"], "no")
218 assert_equal(n.gettransaction(txid_2)["bip125-replaceable"], "no")
219 assert_equal(n.gettransaction(txid_3)["bip125-replaceable"], "yes")
220 assert_equal(n.gettransaction(txid_3b)["bip125-replaceable"], "yes")
221 assert_equal(n.gettransaction(txid_4)["bip125-replaceable"], "unknown")
222
223 self.log.info("Test bip125-replaceable status with listsinceblock")
224 for n in self.nodes[0:2]:
225 txs = {tx['txid']: tx['bip125-replaceable'] for tx in n.listsinceblock()['transactions']}
226 assert_equal(txs[txid_1], "no")
227 assert_equal(txs[txid_2], "no")
228 assert_equal(txs[txid_3], "yes")
229 assert_equal(txs[txid_3b], "yes")
230 assert_equal(txs[txid_4], "unknown")
231
232 self.log.info("Test mined transactions are no longer bip125-replaceable")
233 self.generate(self.nodes[0], 1)
234 assert txid_3b not in self.nodes[0].getrawmempool()
235 assert_equal(self.nodes[0].gettransaction(txid_3b)["bip125-replaceable"], "no")
236 assert_equal(self.nodes[0].gettransaction(txid_4)["bip125-replaceable"], "unknown")
237
238 def run_externally_generated_address_test(self):
239 """Test behavior when receiving address is not in the address book."""
240
241 self.log.info("Setup the same wallet on two nodes")
242 # refill keypool otherwise the second node wouldn't recognize addresses generated on the first nodes
243 self.nodes[0].keypoolrefill(1000)
244 self.stop_nodes()
245 wallet0 = os.path.join(self.nodes[0].chain_path, self.default_wallet_name, "wallet.dat")
246 wallet2 = os.path.join(self.nodes[2].chain_path, self.default_wallet_name, "wallet.dat")
247 shutil.copyfile(wallet0, wallet2)
248 self.start_nodes()
249 # reconnect nodes
250 self.connect_nodes(0, 1)
251 self.connect_nodes(1, 2)
252 self.connect_nodes(2, 0)
253
254 addr1 = self.nodes[0].getnewaddress("pizza1", 'legacy')
255 addr2 = self.nodes[0].getnewaddress("pizza2", 'p2sh-segwit')
256 addr3 = self.nodes[0].getnewaddress("pizza3", 'bech32')
257
258 self.log.info("Send to externally generated addresses")
259 # send to an address beyond the next to be generated to test the keypool gap
260 self.nodes[1].sendtoaddress(addr3, "0.001")
261 self.generate(self.nodes[1], 1)
262
263 # send to an address that is already marked as used due to the keypool gap mechanics
264 self.nodes[1].sendtoaddress(addr2, "0.001")
265 self.generate(self.nodes[1], 1)
266
267 # send to self transaction
268 self.nodes[0].sendtoaddress(addr1, "0.001")
269 self.generate(self.nodes[0], 1)
270
271 self.log.info("Verify listtransactions is the same regardless of where the address was generated")
272 transactions0 = self.nodes[0].listtransactions()
273 transactions2 = self.nodes[2].listtransactions()
274
275 # normalize results: remove fields that normally could differ and sort
276 def normalize_list(txs):
277 for tx in txs:
278 tx.pop('label', None)
279 tx.pop('time', None)
280 tx.pop('timereceived', None)
281 txs.sort(key=lambda x: x['txid'])
282
283 normalize_list(transactions0)
284 normalize_list(transactions2)
285 assert_equal(transactions0, transactions2)
286
287 self.log.info("Verify labels are persistent on the node that generated the addresses")
288 assert_equal(['pizza1'], self.nodes[0].getaddressinfo(addr1)['labels'])
289 assert_equal(['pizza2'], self.nodes[0].getaddressinfo(addr2)['labels'])
290 assert_equal(['pizza3'], self.nodes[0].getaddressinfo(addr3)['labels'])
291
292 def run_coinjoin_test(self):
293 self.log.info('Check "coin-join" transaction')
294 input_0 = next(i for i in self.nodes[0].listunspent(query_options={"minimumAmount": 0.2}, include_unsafe=False))
295 input_1 = next(i for i in self.nodes[1].listunspent(query_options={"minimumAmount": 0.2}, include_unsafe=False))
296 raw_hex = self.nodes[0].createrawtransaction(
297 inputs=[
298 {
299 "txid": input_0["txid"],
300 "vout": input_0["vout"],
301 },
302 {
303 "txid": input_1["txid"],
304 "vout": input_1["vout"],
305 },
306 ],
307 outputs={
308 self.nodes[0].getnewaddress(): 0.123,
309 self.nodes[1].getnewaddress(): 0.123,
310 },
311 )
312 raw_hex = self.nodes[0].signrawtransactionwithwallet(raw_hex)["hex"]
313 raw_hex = self.nodes[1].signrawtransactionwithwallet(raw_hex)["hex"]
314 txid_join = self.nodes[0].sendrawtransaction(hexstring=raw_hex, maxfeerate=0)
315 fee_join = self.nodes[0].getmempoolentry(txid_join)["fees"]["base"]
316 # Fee should be correct: assert_equal(fee_join, self.nodes[0].gettransaction(txid_join)['fee'])
317 # But it is not, see for example https://github.com/limenka/limenka/issues/14136:
318 assert fee_join != self.nodes[0].gettransaction(txid_join)["fee"]
319
320 def run_invalid_parameters_test(self):
321 self.log.info("Test listtransactions RPC parameter validity")
322 assert_raises_rpc_error(-8, 'Label argument must be a valid label name or "*".', self.nodes[0].listtransactions, label="")
323 self.nodes[0].listtransactions(label="*")
324 assert_raises_rpc_error(-8, "Negative count", self.nodes[0].listtransactions, count=-1)
325 assert_raises_rpc_error(-8, "Negative from", self.nodes[0].listtransactions, skip=-1)
326
327 def test_op_return(self):
328 """Test if OP_RETURN outputs will be displayed correctly."""
329 raw_tx = self.nodes[0].createrawtransaction([], [{'data': 'aa'}])
330 funded_tx = self.nodes[0].fundrawtransaction(raw_tx)
331 signed_tx = self.nodes[0].signrawtransactionwithwallet(funded_tx['hex'])
332 tx_id = self.nodes[0].sendrawtransaction(signed_tx['hex'])
333
334 op_ret_tx = [tx for tx in self.nodes[0].listtransactions() if tx['txid'] == tx_id][0]
335
336 assert 'address' not in op_ret_tx
337
338 def test_from_me_status_change(self):
339 self.log.info("Test gettransaction after changing a transaction's 'from me' status")
340 self.nodes[0].createwallet("fromme")
341 default_wallet = self.nodes[0].get_wallet_rpc(self.default_wallet_name)
342 wallet = self.nodes[0].get_wallet_rpc("fromme")
343
344 # The 'fee' field of gettransaction is only added when the transaction is 'from me'
345 # Run twice, once for a transaction in the mempool, again when it confirms
346 for confirm in [False, True]:
347 key = get_generate_key()
348 default_wallet.importprivkey(key.privkey)
349
350 send_res = default_wallet.send(outputs=[{key.p2wpkh_addr: 1}, {wallet.getnewaddress(): 1}])
351 assert_equal(send_res["complete"], True)
352 vout = find_vout_for_address(self.nodes[0], send_res["txid"], key.p2wpkh_addr)
353 utxos = [{"txid": send_res["txid"], "vout": vout}]
354 self.generate(self.nodes[0], 1, sync_fun=self.no_op)
355
356 # Send to the test wallet, ensuring that one input is for the descriptor we will import,
357 # and that there are other inputs belonging to only the sending wallet
358 send_res = default_wallet.send(outputs=[{wallet.getnewaddress(): 1.5}], inputs=utxos, add_inputs=True)
359 assert_equal(send_res["complete"], True)
360 txid = send_res["txid"]
361 self.nodes[0].syncwithvalidationinterfacequeue()
362 tx_info = wallet.gettransaction(txid)
363 assert "fee" not in tx_info
364 assert_equal(any(detail["category"] == "send" for detail in tx_info["details"]), False)
365
366 if confirm:
367 self.generate(self.nodes[0], 1, sync_fun=self.no_op)
368 # Mock time forward and generate blocks so that the import does not rescan the transaction
369 self.nodes[0].setmocktime(int(time.time()) + MAX_FUTURE_BLOCK_TIME + 1)
370 self.generate(self.nodes[0], 10, sync_fun=self.no_op)
371
372 wallet.importprivkey(key.privkey)
373 # TODO: We should check that the fee matches, but since the transaction spends inputs
374 # not known to the wallet, it is incorrectly calculating the fee.
375 # assert_equal(wallet.gettransaction(txid)["fee"], fee)
376 tx_info = wallet.gettransaction(txid)
377 assert "fee" in tx_info
378 assert_equal(any(detail["category"] == "send" for detail in tx_info["details"]), True)
379
380 def create_and_send_transaction(self, utxo, address, amt, feeRate):
381 psbtx = self.nodes[0].walletcreatefundedpsbt([{"txid": utxo['txid'], "vout": utxo['vout']}],
382 {address: amt},
383 0,
384 {"replaceable":True, "feeRate":feeRate})['psbt']
385 signed_tx = self.nodes[0].walletprocesspsbt(psbtx)['psbt']
386 final_tx = self.nodes[0].finalizepsbt(signed_tx)['hex']
387 return self.nodes[0].sendrawtransaction(final_tx)
388
389 def test_listtransactions_display_in_mempool(self):
390 self.log.info('Testing that listtransactions correctly displays whether a transaction is in the mempool')
391 utxo = self.nodes[0].listunspent(query_options={'minimumAmount': 0.15})[0]
392 address = self.nodes[0].getnewaddress()
393
394 tx1_id = self.create_and_send_transaction(utxo, address, 0.1, 0.001)
395
396 new_txs = self.nodes[0].listtransactions(count=2)
397 for tx in new_txs:
398 assert_equal(tx['txid'], tx1_id)
399 assert_equal(tx['in_mempool'], True)
400
401 tx2_id = self.create_and_send_transaction(utxo, address, 0.1, 0.002)
402
403 new_txs = self.nodes[0].listtransactions(count=4)
404 for i in range(2):
405 assert_equal(new_txs[i]['txid'], tx1_id)
406 assert_equal(new_txs[i]['in_mempool'], False)
407
408 for i in range(2, 4):
409 assert_equal(new_txs[i]['txid'], tx2_id)
410 assert_equal(new_txs[i]['in_mempool'], True)
411
412 def test_gettransaction_display_in_mempool(self):
413 self.log.info('Testing that gettransaction correctly displays whether a transaction is in the mempool')
414 utxo = self.nodes[0].listunspent(query_options={'minimumAmount': 0.15})[0]
415 address = self.nodes[0].getnewaddress()
416
417 tx1_id = self.create_and_send_transaction(utxo, address, 0.1, 0.001)
418
419 tx1 = self.nodes[0].gettransaction(tx1_id)
420 assert_equal(tx1['txid'], tx1_id)
421 assert_equal(tx1['in_mempool'], True)
422
423 tx2_id = self.create_and_send_transaction(utxo, address, 0.1, 0.002)
424 tx1 = self.nodes[0].gettransaction(tx1_id)
425 tx2 = self.nodes[0].gettransaction(tx2_id)
426 assert_equal(tx1['txid'], tx1_id)
427 assert_equal(tx1['in_mempool'], False)
428 assert_equal(tx2['txid'], tx2_id)
429 assert_equal(tx2['in_mempool'], True)
430
431
432 if __name__ == '__main__':
433 ListTransactionsTest(__file__).main()
434