rpc_mempool_info.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 RPCs that retrieve information from the mempool."""
6
7 from test_framework.test_framework import LimenkaTestFramework
8 from test_framework.util import (
9 assert_equal,
10 assert_raises_rpc_error,
11 )
12 from test_framework.wallet import MiniWallet
13
14
15 class RPCMempoolInfoTest(LimenkaTestFramework):
16 def set_test_params(self):
17 self.num_nodes = 1
18
19 def run_test(self):
20 self.wallet = MiniWallet(self.nodes[0])
21 confirmed_utxo = self.wallet.get_utxo()
22
23 # Create a tree of unconfirmed transactions in the mempool:
24 # txA
25 # / \
26 # / \
27 # / \
28 # / \
29 # / \
30 # txB txC
31 # / \ / \
32 # / \ / \
33 # txD txE txF txG
34 # \ /
35 # \ /
36 # txH
37
38 def create_tx(**kwargs):
39 return self.wallet.send_self_transfer_multi(
40 from_node=self.nodes[0],
41 **kwargs,
42 )
43
44 txA = create_tx(utxos_to_spend=[confirmed_utxo], num_outputs=2)
45 txB = create_tx(utxos_to_spend=[txA["new_utxos"][0]], num_outputs=2)
46 txC = create_tx(utxos_to_spend=[txA["new_utxos"][1]], num_outputs=2)
47 txD = create_tx(utxos_to_spend=[txB["new_utxos"][0]], num_outputs=1)
48 txE = create_tx(utxos_to_spend=[txB["new_utxos"][1]], num_outputs=1)
49 txF = create_tx(utxos_to_spend=[txC["new_utxos"][0]], num_outputs=2)
50 txG = create_tx(utxos_to_spend=[txC["new_utxos"][1]], num_outputs=1)
51 txH = create_tx(utxos_to_spend=[txE["new_utxos"][0],txF["new_utxos"][0]], num_outputs=1)
52 txidA, txidB, txidC, txidD, txidE, txidF, txidG, txidH = [
53 tx["txid"] for tx in [txA, txB, txC, txD, txE, txF, txG, txH]
54 ]
55
56 mempool = self.nodes[0].getrawmempool()
57 assert_equal(len(mempool), 8)
58 for txid in [txidA, txidB, txidC, txidD, txidE, txidF, txidG, txidH]:
59 assert_equal(txid in mempool, True)
60
61 self.log.info("Find transactions spending outputs")
62 result = self.nodes[0].gettxspendingprevout([ {'txid' : confirmed_utxo['txid'], 'vout' : 0}, {'txid' : txidA, 'vout' : 1} ])
63 assert_equal(result, [ {'txid' : confirmed_utxo['txid'], 'vout' : 0, 'spendingtxid' : txidA}, {'txid' : txidA, 'vout' : 1, 'spendingtxid' : txidC} ])
64
65 self.log.info("Find transaction spending multiple outputs")
66 result = self.nodes[0].gettxspendingprevout([ {'txid' : txidE, 'vout' : 0}, {'txid' : txidF, 'vout' : 0} ])
67 assert_equal(result, [ {'txid' : txidE, 'vout' : 0, 'spendingtxid' : txidH}, {'txid' : txidF, 'vout' : 0, 'spendingtxid' : txidH} ])
68
69 self.log.info("Find no transaction when output is unspent")
70 result = self.nodes[0].gettxspendingprevout([ {'txid' : txidH, 'vout' : 0} ])
71 assert_equal(result, [ {'txid' : txidH, 'vout' : 0} ])
72 result = self.nodes[0].gettxspendingprevout([ {'txid' : txidA, 'vout' : 5} ])
73 assert_equal(result, [ {'txid' : txidA, 'vout' : 5} ])
74
75 self.log.info("Mixed spent and unspent outputs")
76 result = self.nodes[0].gettxspendingprevout([ {'txid' : txidB, 'vout' : 0}, {'txid' : txidG, 'vout' : 3} ])
77 assert_equal(result, [ {'txid' : txidB, 'vout' : 0, 'spendingtxid' : txidD}, {'txid' : txidG, 'vout' : 3} ])
78
79 self.log.info("Unknown input fields")
80 assert_raises_rpc_error(-3, "Unexpected key unknown", self.nodes[0].gettxspendingprevout, [{'txid' : txidC, 'vout' : 1, 'unknown' : 42}])
81
82 self.log.info("Invalid vout provided")
83 assert_raises_rpc_error(-8, "Invalid parameter, vout cannot be negative", self.nodes[0].gettxspendingprevout, [{'txid' : txidA, 'vout' : -1}])
84
85 self.log.info("Invalid txid provided")
86 assert_raises_rpc_error(-3, "JSON value of type number for field txid is not of expected type string", self.nodes[0].gettxspendingprevout, [{'txid' : 42, 'vout' : 0}])
87
88 self.log.info("Missing outputs")
89 assert_raises_rpc_error(-8, "Invalid parameter, outputs are missing", self.nodes[0].gettxspendingprevout, [])
90
91 self.log.info("Missing vout")
92 assert_raises_rpc_error(-3, "Missing vout", self.nodes[0].gettxspendingprevout, [{'txid' : txidA}])
93
94 self.log.info("Missing txid")
95 assert_raises_rpc_error(-3, "Missing txid", self.nodes[0].gettxspendingprevout, [{'vout' : 3}])
96
97
98 if __name__ == '__main__':
99 RPCMempoolInfoTest(__file__).main()
100