rpc_getdescriptoractivity.py
1 #!/usr/bin/env python3 2 # Copyright (c) 2024-present 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 6 from decimal import Decimal 7 8 from test_framework.test_framework import BitcoinTestFramework 9 from test_framework.util import assert_equal, assert_raises_rpc_error 10 from test_framework.messages import COIN 11 from test_framework.wallet import MiniWallet, MiniWalletMode, getnewdestination 12 13 14 class GetBlocksActivityTest(BitcoinTestFramework): 15 def set_test_params(self): 16 self.num_nodes = 1 17 self.setup_clean_chain = True 18 19 def run_test(self): 20 node = self.nodes[0] 21 wallet = MiniWallet(node) 22 node.setmocktime(node.getblockheader(node.getbestblockhash())['time']) 23 self.generate(wallet, 200) 24 25 self.test_no_activity(node) 26 self.test_activity_in_block(node, wallet) 27 self.test_no_mempool_inclusion(node, wallet) 28 self.test_multiple_addresses(node, wallet) 29 self.test_invalid_blockhash(node, wallet) 30 self.test_invalid_descriptor(node, wallet) 31 self.test_confirmed_and_unconfirmed(node, wallet) 32 self.test_receive_then_spend(node, wallet) 33 self.test_no_address(node, wallet) 34 self.test_required_args(node) 35 36 def test_no_activity(self, node): 37 self.log.info("Test that no activity is found for an unused address") 38 _, _, addr_1 = getnewdestination() 39 result = node.getdescriptoractivity([], [f"addr({addr_1})"], True) 40 assert_equal(len(result['activity']), 0) 41 42 def test_activity_in_block(self, node, wallet): 43 self.log.info("Test that receive activity is correctly reported in a mined block") 44 _, spk_1, addr_1 = getnewdestination(address_type='bech32m') 45 txid = wallet.send_to(from_node=node, scriptPubKey=spk_1, amount=1 * COIN)['txid'] 46 blockhash = self.generate(node, 1)[0] 47 48 # Test getdescriptoractivity with the specific blockhash 49 result = node.getdescriptoractivity([blockhash], [f"addr({addr_1})"], True) 50 assert_equal(list(result.keys()), ['activity']) 51 [activity] = result['activity'] 52 53 for k, v in { 54 'amount': Decimal('1.00000000'), 55 'blockhash': blockhash, 56 'height': 201, 57 'txid': txid, 58 'type': 'receive', 59 'vout': 1, 60 }.items(): 61 assert_equal(activity[k], v) 62 63 outspk = activity['output_spk'] 64 65 assert_equal(outspk['asm'][:2], '1 ') 66 assert_equal(outspk['desc'].split('(')[0], 'rawtr') 67 assert_equal(outspk['hex'], spk_1.hex()) 68 assert_equal(outspk['address'], addr_1) 69 assert_equal(outspk['type'], 'witness_v1_taproot') 70 71 72 def test_no_mempool_inclusion(self, node, wallet): 73 self.log.info("Test that mempool transactions are not included when include_mempool argument is False") 74 _, spk_1, addr_1 = getnewdestination() 75 wallet.send_to(from_node=node, scriptPubKey=spk_1, amount=1 * COIN) 76 77 _, spk_2, addr_2 = getnewdestination() 78 wallet.send_to( 79 from_node=node, scriptPubKey=spk_2, amount=1 * COIN) 80 81 # Do not generate a block to keep the transaction in the mempool 82 83 result = node.getdescriptoractivity([], [f"addr({addr_1})", f"addr({addr_2})"], False) 84 85 assert_equal(len(result['activity']), 0) 86 87 def test_multiple_addresses(self, node, wallet): 88 self.log.info("Test querying multiple addresses returns all activity correctly") 89 _, spk_1, addr_1 = getnewdestination() 90 _, spk_2, addr_2 = getnewdestination() 91 wallet.send_to(from_node=node, scriptPubKey=spk_1, amount=1 * COIN) 92 wallet.send_to(from_node=node, scriptPubKey=spk_2, amount=2 * COIN) 93 94 blockhash = self.generate(node, 1)[0] 95 96 result = node.getdescriptoractivity([blockhash], [f"addr({addr_1})", f"addr({addr_2})"], True) 97 98 assert_equal(len(result['activity']), 2) 99 100 # Duplicate address specification is fine. 101 assert_equal( 102 result, 103 node.getdescriptoractivity([blockhash], [ 104 f"addr({addr_1})", f"addr({addr_1})", f"addr({addr_2})"], True)) 105 106 # Flipping descriptor order doesn't affect results. 107 result_flipped = node.getdescriptoractivity( 108 [blockhash], [f"addr({addr_2})", f"addr({addr_1})"], True) 109 assert_equal(result, result_flipped) 110 111 [a1] = [a for a in result['activity'] if a['output_spk']['address'] == addr_1] 112 [a2] = [a for a in result['activity'] if a['output_spk']['address'] == addr_2] 113 114 assert a1['blockhash'] == blockhash 115 assert a1['amount'] == 1.0 116 117 assert a2['blockhash'] == blockhash 118 assert a2['amount'] == 2.0 119 120 def test_invalid_blockhash(self, node, wallet): 121 self.log.info("Test that passing an invalid blockhash raises appropriate RPC error") 122 self.generate(node, 20) # Generate to get more fees 123 124 _, spk_1, addr_1 = getnewdestination() 125 wallet.send_to(from_node=node, scriptPubKey=spk_1, amount=1 * COIN) 126 127 invalid_blockhash = "0000000000000000000000000000000000000000000000000000000000000000" 128 129 assert_raises_rpc_error( 130 -5, "Block not found", 131 node.getdescriptoractivity, [invalid_blockhash], [f"addr({addr_1})"], True) 132 133 def test_invalid_descriptor(self, node, wallet): 134 self.log.info("Test that an invalid descriptor raises the correct RPC error") 135 blockhash = self.generate(node, 1)[0] 136 _, _, addr_1 = getnewdestination() 137 138 assert_raises_rpc_error( 139 -5, "is not a valid descriptor", 140 node.getdescriptoractivity, [blockhash], [f"addrx({addr_1})"], True) 141 142 def test_confirmed_and_unconfirmed(self, node, wallet): 143 self.log.info("Test that both confirmed and unconfirmed transactions are reported correctly") 144 self.generate(node, 20) # Generate to get more fees 145 146 _, spk_1, addr_1 = getnewdestination() 147 txid_1 = wallet.send_to( 148 from_node=node, scriptPubKey=spk_1, amount=1 * COIN)['txid'] 149 blockhash = self.generate(node, 1)[0] 150 151 _, spk_2, to_addr = getnewdestination() 152 txid_2 = wallet.send_to( 153 from_node=node, scriptPubKey=spk_2, amount=1 * COIN)['txid'] 154 155 result = node.getdescriptoractivity( 156 [blockhash], [f"addr({addr_1})", f"addr({to_addr})"], True) 157 158 activity = result['activity'] 159 assert_equal(len(activity), 2) 160 161 [confirmed] = [a for a in activity if a.get('blockhash') == blockhash] 162 assert confirmed['txid'] == txid_1 163 assert confirmed['height'] == node.getblockchaininfo()['blocks'] 164 165 [unconfirmed] = [a for a in activity if not a.get('blockhash')] 166 assert 'blockhash' not in unconfirmed 167 assert 'height' not in unconfirmed 168 169 assert any(a['txid'] == txid_2 for a in activity if not a.get('blockhash')) 170 171 def test_receive_then_spend(self, node, wallet): 172 """Also important because this tests multiple blockhashes.""" 173 self.log.info("Test receive and spend activities across different blocks are reported consistently") 174 self.generate(node, 20) # Generate to get more fees 175 176 sent1 = wallet.send_self_transfer(from_node=node) 177 utxo = sent1['new_utxo'] 178 blockhash_1 = self.generate(node, 1)[0] 179 180 sent2 = wallet.send_self_transfer(from_node=node, utxo_to_spend=utxo) 181 blockhash_2 = self.generate(node, 1)[0] 182 183 result = node.getdescriptoractivity( 184 [blockhash_1, blockhash_2], [wallet.get_descriptor()], True) 185 186 assert_equal(len(result['activity']), 4) 187 188 assert result['activity'][1]['type'] == 'receive' 189 assert result['activity'][1]['txid'] == sent1['txid'] 190 assert result['activity'][1]['blockhash'] == blockhash_1 191 192 assert result['activity'][2]['type'] == 'spend' 193 assert result['activity'][2]['spend_txid'] == sent2['txid'] 194 assert result['activity'][2]['spend_vin'] == 0 195 assert result['activity'][2]['prevout_txid'] == sent1['txid'] 196 assert result['activity'][2]['blockhash'] == blockhash_2 197 198 # Test that reversing the blockorder yields the same result. 199 assert_equal(result, node.getdescriptoractivity( 200 [blockhash_1, blockhash_2], [wallet.get_descriptor()], True)) 201 202 self.log.info("Test that duplicated blockhash request does not report duplicated results") 203 # Test that duplicating a blockhash yields the same result. 204 assert_equal(result, node.getdescriptoractivity( 205 [blockhash_1, blockhash_2, blockhash_2], [wallet.get_descriptor()], True)) 206 207 def test_no_address(self, node, wallet): 208 self.log.info("Test that activity is still reported for scripts without an associated address") 209 raw_wallet = MiniWallet(self.nodes[0], mode=MiniWalletMode.RAW_P2PK) 210 self.generate(raw_wallet, 100) 211 212 no_addr_tx = raw_wallet.send_self_transfer(from_node=node) 213 raw_desc = raw_wallet.get_descriptor() 214 215 blockhash = self.generate(node, 1)[0] 216 217 result = node.getdescriptoractivity([blockhash], [raw_desc], False) 218 219 assert_equal(len(result['activity']), 2) 220 221 a1 = result['activity'][0] 222 a2 = result['activity'][1] 223 224 assert a1['type'] == "spend" 225 assert a1['blockhash'] == blockhash 226 # sPK lacks address. 227 assert_equal(list(a1['prevout_spk'].keys()), ['asm', 'desc', 'hex', 'type']) 228 assert a1['amount'] == no_addr_tx["fee"] + Decimal(no_addr_tx["tx"].vout[0].nValue) / COIN 229 230 assert a2['type'] == "receive" 231 assert a2['blockhash'] == blockhash 232 # sPK lacks address. 233 assert_equal(list(a2['output_spk'].keys()), ['asm', 'desc', 'hex', 'type']) 234 assert a2['amount'] == Decimal(no_addr_tx["tx"].vout[0].nValue) / COIN 235 236 def test_required_args(self, node): 237 self.log.info("Test that required arguments must be passed") 238 assert_raises_rpc_error(-1, "getdescriptoractivity", node.getdescriptoractivity) 239 assert_raises_rpc_error(-1, "getdescriptoractivity", node.getdescriptoractivity, []) 240 241 242 if __name__ == '__main__': 243 GetBlocksActivityTest(__file__).main()