#!/usr/bin/env python3 # Copyright (c) 2024-present The Bitcoin Core developers # Distributed under the MIT software license, see the accompanying # file COPYING or http://www.opensource.org/licenses/mit-license.php. from decimal import Decimal from test_framework.test_framework import BitcoinTestFramework from test_framework.util import assert_equal, assert_raises_rpc_error from test_framework.messages import COIN from test_framework.wallet import MiniWallet, MiniWalletMode, getnewdestination class GetBlocksActivityTest(BitcoinTestFramework): def set_test_params(self): self.num_nodes = 1 self.setup_clean_chain = True def run_test(self): node = self.nodes[0] wallet = MiniWallet(node) node.setmocktime(node.getblockheader(node.getbestblockhash())['time']) self.generate(wallet, 200) self.test_no_activity(node) self.test_activity_in_block(node, wallet) self.test_no_mempool_inclusion(node, wallet) self.test_multiple_addresses(node, wallet) self.test_invalid_blockhash(node, wallet) self.test_invalid_descriptor(node, wallet) self.test_confirmed_and_unconfirmed(node, wallet) self.test_receive_then_spend(node, wallet) self.test_no_address(node, wallet) def test_no_activity(self, node): _, _, addr_1 = getnewdestination() result = node.getdescriptoractivity([], [f"addr({addr_1})"], True) assert_equal(len(result['activity']), 0) def test_activity_in_block(self, node, wallet): _, spk_1, addr_1 = getnewdestination(address_type='bech32m') txid = wallet.send_to(from_node=node, scriptPubKey=spk_1, amount=1 * COIN)['txid'] blockhash = self.generate(node, 1)[0] # Test getdescriptoractivity with the specific blockhash result = node.getdescriptoractivity([blockhash], [f"addr({addr_1})"], True) assert_equal(list(result.keys()), ['activity']) [activity] = result['activity'] for k, v in { 'amount': Decimal('1.00000000'), 'blockhash': blockhash, 'height': 201, 'txid': txid, 'type': 'receive', 'vout': 1, }.items(): assert_equal(activity[k], v) outspk = activity['output_spk'] assert_equal(outspk['asm'][:2], '1 ') assert_equal(outspk['desc'].split('(')[0], 'rawtr') assert_equal(outspk['hex'], spk_1.hex()) assert_equal(outspk['address'], addr_1) assert_equal(outspk['type'], 'witness_v1_taproot') def test_no_mempool_inclusion(self, node, wallet): _, spk_1, addr_1 = getnewdestination() wallet.send_to(from_node=node, scriptPubKey=spk_1, amount=1 * COIN) _, spk_2, addr_2 = getnewdestination() wallet.send_to( from_node=node, scriptPubKey=spk_2, amount=1 * COIN) # Do not generate a block to keep the transaction in the mempool result = node.getdescriptoractivity([], [f"addr({addr_1})", f"addr({addr_2})"], False) assert_equal(len(result['activity']), 0) def test_multiple_addresses(self, node, wallet): _, spk_1, addr_1 = getnewdestination() _, spk_2, addr_2 = getnewdestination() wallet.send_to(from_node=node, scriptPubKey=spk_1, amount=1 * COIN) wallet.send_to(from_node=node, scriptPubKey=spk_2, amount=2 * COIN) blockhash = self.generate(node, 1)[0] result = node.getdescriptoractivity([blockhash], [f"addr({addr_1})", f"addr({addr_2})"], True) assert_equal(len(result['activity']), 2) # Duplicate address specification is fine. assert_equal( result, node.getdescriptoractivity([blockhash], [ f"addr({addr_1})", f"addr({addr_1})", f"addr({addr_2})"], True)) # Flipping descriptor order doesn't affect results. result_flipped = node.getdescriptoractivity( [blockhash], [f"addr({addr_2})", f"addr({addr_1})"], True) assert_equal(result, result_flipped) [a1] = [a for a in result['activity'] if a['output_spk']['address'] == addr_1] [a2] = [a for a in result['activity'] if a['output_spk']['address'] == addr_2] assert a1['blockhash'] == blockhash assert a1['amount'] == 1.0 assert a2['blockhash'] == blockhash assert a2['amount'] == 2.0 def test_invalid_blockhash(self, node, wallet): self.generate(node, 20) # Generate to get more fees _, spk_1, addr_1 = getnewdestination() wallet.send_to(from_node=node, scriptPubKey=spk_1, amount=1 * COIN) invalid_blockhash = "0000000000000000000000000000000000000000000000000000000000000000" assert_raises_rpc_error( -5, "Block not found", node.getdescriptoractivity, [invalid_blockhash], [f"addr({addr_1})"], True) def test_invalid_descriptor(self, node, wallet): blockhash = self.generate(node, 1)[0] _, _, addr_1 = getnewdestination() assert_raises_rpc_error( -5, "is not a valid descriptor", node.getdescriptoractivity, [blockhash], [f"addrx({addr_1})"], True) def test_confirmed_and_unconfirmed(self, node, wallet): self.generate(node, 20) # Generate to get more fees _, spk_1, addr_1 = getnewdestination() txid_1 = wallet.send_to( from_node=node, scriptPubKey=spk_1, amount=1 * COIN)['txid'] blockhash = self.generate(node, 1)[0] _, spk_2, to_addr = getnewdestination() txid_2 = wallet.send_to( from_node=node, scriptPubKey=spk_2, amount=1 * COIN)['txid'] result = node.getdescriptoractivity( [blockhash], [f"addr({addr_1})", f"addr({to_addr})"], True) activity = result['activity'] assert_equal(len(activity), 2) [confirmed] = [a for a in activity if a.get('blockhash') == blockhash] assert confirmed['txid'] == txid_1 assert confirmed['height'] == node.getblockchaininfo()['blocks'] [unconfirmed] = [a for a in activity if not a.get('blockhash')] assert 'blockhash' not in unconfirmed assert 'height' not in unconfirmed assert any(a['txid'] == txid_2 for a in activity if not a.get('blockhash')) def test_receive_then_spend(self, node, wallet): """Also important because this tests multiple blockhashes.""" self.generate(node, 20) # Generate to get more fees sent1 = wallet.send_self_transfer(from_node=node) utxo = sent1['new_utxo'] blockhash_1 = self.generate(node, 1)[0] sent2 = wallet.send_self_transfer(from_node=node, utxo_to_spend=utxo) blockhash_2 = self.generate(node, 1)[0] result = node.getdescriptoractivity( [blockhash_1, blockhash_2], [wallet.get_descriptor()], True) assert_equal(len(result['activity']), 4) assert result['activity'][1]['type'] == 'receive' assert result['activity'][1]['txid'] == sent1['txid'] assert result['activity'][1]['blockhash'] == blockhash_1 assert result['activity'][2]['type'] == 'spend' assert result['activity'][2]['spend_txid'] == sent2['txid'] assert result['activity'][2]['prevout_txid'] == sent1['txid'] assert result['activity'][2]['blockhash'] == blockhash_2 # Test that reversing the blockorder yields the same result. assert_equal(result, node.getdescriptoractivity( [blockhash_1, blockhash_2], [wallet.get_descriptor()], True)) # Test that duplicating a blockhash yields the same result. assert_equal(result, node.getdescriptoractivity( [blockhash_1, blockhash_2, blockhash_2], [wallet.get_descriptor()], True)) def test_no_address(self, node, wallet): raw_wallet = MiniWallet(self.nodes[0], mode=MiniWalletMode.RAW_P2PK) self.generate(raw_wallet, 100) no_addr_tx = raw_wallet.send_self_transfer(from_node=node) raw_desc = raw_wallet.get_descriptor() blockhash = self.generate(node, 1)[0] result = node.getdescriptoractivity([blockhash], [raw_desc], False) assert_equal(len(result['activity']), 2) a1 = result['activity'][0] a2 = result['activity'][1] assert a1['type'] == "spend" assert a1['blockhash'] == blockhash # sPK lacks address. assert_equal(list(a1['prevout_spk'].keys()), ['asm', 'desc', 'hex', 'type']) assert a1['amount'] == no_addr_tx["fee"] + Decimal(no_addr_tx["tx"].vout[0].nValue) / COIN assert a2['type'] == "receive" assert a2['blockhash'] == blockhash # sPK lacks address. assert_equal(list(a2['output_spk'].keys()), ['asm', 'desc', 'hex', 'type']) assert a2['amount'] == Decimal(no_addr_tx["tx"].vout[0].nValue) / COIN if __name__ == '__main__': GetBlocksActivityTest(__file__).main()