rpc_getdescriptoractivity.py raw
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_equal(a1['blockhash'], blockhash)
115 assert_equal(a1['amount'], 1.0)
116
117 assert_equal(a2['blockhash'], blockhash)
118 assert_equal(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_equal(confirmed['txid'], txid_1)
163 assert_equal(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 txid_2 in [a['txid'] 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_equal(result['activity'][1]['type'], 'receive')
189 assert_equal(result['activity'][1]['txid'], sent1['txid'])
190 assert_equal(result['activity'][1]['blockhash'], blockhash_1)
191
192 assert_equal(result['activity'][2]['type'], 'spend')
193 assert_equal(result['activity'][2]['spend_txid'], sent2['txid'])
194 assert_equal(result['activity'][2]['spend_vin'], 0)
195 assert_equal(result['activity'][2]['prevout_txid'], sent1['txid'])
196 assert_equal(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_equal(a1['type'], "spend")
225 assert_equal(a1['blockhash'], blockhash)
226 # sPK lacks address.
227 assert_equal(list(a1['prevout_spk'].keys()), ['asm', 'desc', 'hex', 'type'])
228 assert_equal(a1['amount'], no_addr_tx["fee"] + Decimal(no_addr_tx["tx"].vout[0].nValue) / COIN)
229
230 assert_equal(a2['type'], "receive")
231 assert_equal(a2['blockhash'], blockhash)
232 # sPK lacks address.
233 assert_equal(list(a2['output_spk'].keys()), ['asm', 'desc', 'hex', 'type'])
234 assert_equal(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()
244