rpc_mempool_info.py
1 #!/usr/bin/env python3 2 # Copyright (c) 2014-2022 The Bitcoin Core 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 BitcoinTestFramework 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(BitcoinTestFramework): 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().main()