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 35 def test_no_activity(self, node): 36 _, _, addr_1 = getnewdestination() 37 result = node.getdescriptoractivity([], [f"addr({addr_1})"], True) 38 assert_equal(len(result['activity']), 0) 39 40 def test_activity_in_block(self, node, wallet): 41 _, spk_1, addr_1 = getnewdestination(address_type='bech32m') 42 txid = wallet.send_to(from_node=node, scriptPubKey=spk_1, amount=1 * COIN)['txid'] 43 blockhash = self.generate(node, 1)[0] 44 45 # Test getdescriptoractivity with the specific blockhash 46 result = node.getdescriptoractivity([blockhash], [f"addr({addr_1})"], True) 47 assert_equal(list(result.keys()), ['activity']) 48 [activity] = result['activity'] 49 50 for k, v in { 51 'amount': Decimal('1.00000000'), 52 'blockhash': blockhash, 53 'height': 201, 54 'txid': txid, 55 'type': 'receive', 56 'vout': 1, 57 }.items(): 58 assert_equal(activity[k], v) 59 60 outspk = activity['output_spk'] 61 62 assert_equal(outspk['asm'][:2], '1 ') 63 assert_equal(outspk['desc'].split('(')[0], 'rawtr') 64 assert_equal(outspk['hex'], spk_1.hex()) 65 assert_equal(outspk['address'], addr_1) 66 assert_equal(outspk['type'], 'witness_v1_taproot') 67 68 69 def test_no_mempool_inclusion(self, node, wallet): 70 _, spk_1, addr_1 = getnewdestination() 71 wallet.send_to(from_node=node, scriptPubKey=spk_1, amount=1 * COIN) 72 73 _, spk_2, addr_2 = getnewdestination() 74 wallet.send_to( 75 from_node=node, scriptPubKey=spk_2, amount=1 * COIN) 76 77 # Do not generate a block to keep the transaction in the mempool 78 79 result = node.getdescriptoractivity([], [f"addr({addr_1})", f"addr({addr_2})"], False) 80 81 assert_equal(len(result['activity']), 0) 82 83 def test_multiple_addresses(self, node, wallet): 84 _, spk_1, addr_1 = getnewdestination() 85 _, spk_2, addr_2 = getnewdestination() 86 wallet.send_to(from_node=node, scriptPubKey=spk_1, amount=1 * COIN) 87 wallet.send_to(from_node=node, scriptPubKey=spk_2, amount=2 * COIN) 88 89 blockhash = self.generate(node, 1)[0] 90 91 result = node.getdescriptoractivity([blockhash], [f"addr({addr_1})", f"addr({addr_2})"], True) 92 93 assert_equal(len(result['activity']), 2) 94 95 # Duplicate address specification is fine. 96 assert_equal( 97 result, 98 node.getdescriptoractivity([blockhash], [ 99 f"addr({addr_1})", f"addr({addr_1})", f"addr({addr_2})"], True)) 100 101 # Flipping descriptor order doesn't affect results. 102 result_flipped = node.getdescriptoractivity( 103 [blockhash], [f"addr({addr_2})", f"addr({addr_1})"], True) 104 assert_equal(result, result_flipped) 105 106 [a1] = [a for a in result['activity'] if a['output_spk']['address'] == addr_1] 107 [a2] = [a for a in result['activity'] if a['output_spk']['address'] == addr_2] 108 109 assert a1['blockhash'] == blockhash 110 assert a1['amount'] == 1.0 111 112 assert a2['blockhash'] == blockhash 113 assert a2['amount'] == 2.0 114 115 def test_invalid_blockhash(self, node, wallet): 116 self.generate(node, 20) # Generate to get more fees 117 118 _, spk_1, addr_1 = getnewdestination() 119 wallet.send_to(from_node=node, scriptPubKey=spk_1, amount=1 * COIN) 120 121 invalid_blockhash = "0000000000000000000000000000000000000000000000000000000000000000" 122 123 assert_raises_rpc_error( 124 -5, "Block not found", 125 node.getdescriptoractivity, [invalid_blockhash], [f"addr({addr_1})"], True) 126 127 def test_invalid_descriptor(self, node, wallet): 128 blockhash = self.generate(node, 1)[0] 129 _, _, addr_1 = getnewdestination() 130 131 assert_raises_rpc_error( 132 -5, "is not a valid descriptor", 133 node.getdescriptoractivity, [blockhash], [f"addrx({addr_1})"], True) 134 135 def test_confirmed_and_unconfirmed(self, node, wallet): 136 self.generate(node, 20) # Generate to get more fees 137 138 _, spk_1, addr_1 = getnewdestination() 139 txid_1 = wallet.send_to( 140 from_node=node, scriptPubKey=spk_1, amount=1 * COIN)['txid'] 141 blockhash = self.generate(node, 1)[0] 142 143 _, spk_2, to_addr = getnewdestination() 144 txid_2 = wallet.send_to( 145 from_node=node, scriptPubKey=spk_2, amount=1 * COIN)['txid'] 146 147 result = node.getdescriptoractivity( 148 [blockhash], [f"addr({addr_1})", f"addr({to_addr})"], True) 149 150 activity = result['activity'] 151 assert_equal(len(activity), 2) 152 153 [confirmed] = [a for a in activity if a.get('blockhash') == blockhash] 154 assert confirmed['txid'] == txid_1 155 assert confirmed['height'] == node.getblockchaininfo()['blocks'] 156 157 [unconfirmed] = [a for a in activity if not a.get('blockhash')] 158 assert 'blockhash' not in unconfirmed 159 assert 'height' not in unconfirmed 160 161 assert any(a['txid'] == txid_2 for a in activity if not a.get('blockhash')) 162 163 def test_receive_then_spend(self, node, wallet): 164 """Also important because this tests multiple blockhashes.""" 165 self.generate(node, 20) # Generate to get more fees 166 167 sent1 = wallet.send_self_transfer(from_node=node) 168 utxo = sent1['new_utxo'] 169 blockhash_1 = self.generate(node, 1)[0] 170 171 sent2 = wallet.send_self_transfer(from_node=node, utxo_to_spend=utxo) 172 blockhash_2 = self.generate(node, 1)[0] 173 174 result = node.getdescriptoractivity( 175 [blockhash_1, blockhash_2], [wallet.get_descriptor()], True) 176 177 assert_equal(len(result['activity']), 4) 178 179 assert result['activity'][1]['type'] == 'receive' 180 assert result['activity'][1]['txid'] == sent1['txid'] 181 assert result['activity'][1]['blockhash'] == blockhash_1 182 183 assert result['activity'][2]['type'] == 'spend' 184 assert result['activity'][2]['spend_txid'] == sent2['txid'] 185 assert result['activity'][2]['prevout_txid'] == sent1['txid'] 186 assert result['activity'][2]['blockhash'] == blockhash_2 187 188 # Test that reversing the blockorder yields the same result. 189 assert_equal(result, node.getdescriptoractivity( 190 [blockhash_1, blockhash_2], [wallet.get_descriptor()], True)) 191 192 # Test that duplicating a blockhash yields the same result. 193 assert_equal(result, node.getdescriptoractivity( 194 [blockhash_1, blockhash_2, blockhash_2], [wallet.get_descriptor()], True)) 195 196 def test_no_address(self, node, wallet): 197 raw_wallet = MiniWallet(self.nodes[0], mode=MiniWalletMode.RAW_P2PK) 198 self.generate(raw_wallet, 100) 199 200 no_addr_tx = raw_wallet.send_self_transfer(from_node=node) 201 raw_desc = raw_wallet.get_descriptor() 202 203 blockhash = self.generate(node, 1)[0] 204 205 result = node.getdescriptoractivity([blockhash], [raw_desc], False) 206 207 assert_equal(len(result['activity']), 2) 208 209 a1 = result['activity'][0] 210 a2 = result['activity'][1] 211 212 assert a1['type'] == "spend" 213 assert a1['blockhash'] == blockhash 214 # sPK lacks address. 215 assert_equal(list(a1['prevout_spk'].keys()), ['asm', 'desc', 'hex', 'type']) 216 assert a1['amount'] == no_addr_tx["fee"] + Decimal(no_addr_tx["tx"].vout[0].nValue) / COIN 217 218 assert a2['type'] == "receive" 219 assert a2['blockhash'] == blockhash 220 # sPK lacks address. 221 assert_equal(list(a2['output_spk'].keys()), ['asm', 'desc', 'hex', 'type']) 222 assert a2['amount'] == Decimal(no_addr_tx["tx"].vout[0].nValue) / COIN 223 224 225 if __name__ == '__main__': 226 GetBlocksActivityTest(__file__).main()