interface_rest.py raw
1 #!/usr/bin/env python3
2 # Copyright (c) 2014-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 """Test the REST API."""
6
7 from decimal import Decimal
8 from enum import Enum
9 from io import BytesIO
10 import http.client
11 import json
12 import typing
13 import urllib.parse
14
15
16 from test_framework.messages import (
17 BLOCK_HEADER_SIZE,
18 COIN,
19 deser_block_spent_outputs,
20 )
21 from test_framework.test_framework import BitcoinTestFramework
22 from test_framework.util import (
23 assert_equal,
24 assert_greater_than,
25 assert_greater_than_or_equal,
26 )
27 from test_framework.wallet import (
28 MiniWallet,
29 getnewdestination,
30 )
31
32
33 INVALID_PARAM = "abc"
34 UNKNOWN_PARAM = "0000000000000000000000000000000000000000000000000000000000000000"
35
36
37 class ReqType(Enum):
38 JSON = 1
39 BIN = 2
40 HEX = 3
41
42 class RetType(Enum):
43 OBJ = 1
44 BYTES = 2
45 JSON = 3
46
47 def filter_output_indices_by_value(vouts, value):
48 for vout in vouts:
49 if vout['value'] == value:
50 yield vout['n']
51
52 class RESTTest (BitcoinTestFramework):
53 def set_test_params(self):
54 self.num_nodes = 2
55 self.extra_args = [["-rest", "-blockfilterindex=1"], []]
56 # whitelist peers to speed up tx relay / mempool sync
57 self.noban_tx_relay = True
58
59 def test_rest_request(
60 self,
61 uri: str,
62 http_method: str = 'GET',
63 req_type: ReqType = ReqType.JSON,
64 body: str = '',
65 status: int = 200,
66 ret_type: RetType = RetType.JSON,
67 query_params: typing.Union[dict[str, typing.Any], str, None] = None,
68 ) -> typing.Union[http.client.HTTPResponse, bytes, str, None]:
69 rest_uri = '/rest' + uri
70 if req_type in ReqType:
71 rest_uri += f'.{req_type.name.lower()}'
72 if query_params:
73 if isinstance(query_params, str):
74 rest_uri += f'?{query_params}'
75 else:
76 rest_uri += f'?{urllib.parse.urlencode(query_params)}'
77
78 conn = http.client.HTTPConnection(self.url.hostname, self.url.port)
79 self.log.debug(f'{http_method} {rest_uri} {body}')
80 if http_method == 'GET':
81 conn.request('GET', rest_uri)
82 elif http_method == 'POST':
83 conn.request('POST', rest_uri, body)
84 resp = conn.getresponse()
85
86 assert resp.status == status, f"Expected: {status}, Got: {resp.status} ({resp.reason}) - Response: {str(resp.read())}"
87
88 if ret_type == RetType.OBJ:
89 return resp
90 elif ret_type == RetType.BYTES:
91 return resp.read()
92 elif ret_type == RetType.JSON:
93 return json.loads(resp.read().decode('utf-8'), parse_float=Decimal)
94
95 return None
96
97 def run_test(self):
98 self.url = urllib.parse.urlparse(self.nodes[0].url)
99 self.wallet = MiniWallet(self.nodes[0])
100
101 self.log.info("Broadcast test transaction and sync nodes")
102 txid = self.wallet.send_to(from_node=self.nodes[0], scriptPubKey=getnewdestination()[1], amount=int(0.1 * COIN))["txid"]
103 self.sync_all()
104
105 self.log.info("Test the /tx URI")
106
107 json_obj = self.test_rest_request(f"/tx/{txid}")
108 assert_equal(json_obj['txid'], txid)
109
110 # Check hex format response
111 hex_response = self.test_rest_request(f"/tx/{txid}", req_type=ReqType.HEX, ret_type=RetType.OBJ)
112 assert_greater_than_or_equal(int(hex_response.getheader('content-length')),
113 json_obj['size']*2)
114
115 spent = (json_obj['vin'][0]['txid'], json_obj['vin'][0]['vout']) # get the vin to later check for utxo (should be spent by then)
116 # get n of 0.1 outpoint
117 n, = filter_output_indices_by_value(json_obj['vout'], Decimal('0.1'))
118 spending = (txid, n)
119
120 # Test /tx with an invalid and an unknown txid
121 resp = self.test_rest_request(uri=f"/tx/{INVALID_PARAM}", ret_type=RetType.OBJ, status=400)
122 assert_equal(resp.read().decode('utf-8').rstrip(), f"Invalid hash: {INVALID_PARAM}")
123 resp = self.test_rest_request(uri=f"/tx/{UNKNOWN_PARAM}", ret_type=RetType.OBJ, status=404)
124 assert_equal(resp.read().decode('utf-8').rstrip(), f"{UNKNOWN_PARAM} not found")
125
126 self.log.info("Query an unspent TXO using the /getutxos URI")
127
128 self.generate(self.wallet, 1)
129 bb_hash = self.nodes[0].getbestblockhash()
130
131 # Check chainTip response
132 json_obj = self.test_rest_request(f"/getutxos/{spending[0]}-{spending[1]}")
133 assert_equal(json_obj['chaintipHash'], bb_hash)
134
135 # Make sure there is one utxo
136 assert_equal(len(json_obj['utxos']), 1)
137 assert_equal(json_obj['utxos'][0]['value'], Decimal('0.1'))
138
139 self.log.info("Query a spent TXO using the /getutxos URI")
140
141 json_obj = self.test_rest_request(f"/getutxos/{spent[0]}-{spent[1]}")
142
143 # Check chainTip response
144 assert_equal(json_obj['chaintipHash'], bb_hash)
145
146 # Make sure there is no utxo in the response because this outpoint has been spent
147 assert_equal(len(json_obj['utxos']), 0)
148
149 # Check bitmap
150 assert_equal(json_obj['bitmap'], "0")
151
152 self.log.info("Query two TXOs using the /getutxos URI")
153
154 json_obj = self.test_rest_request(f"/getutxos/{spending[0]}-{spending[1]}/{spent[0]}-{spent[1]}")
155
156 assert_equal(len(json_obj['utxos']), 1)
157 assert_equal(json_obj['bitmap'], "10")
158
159 self.log.info("Query the TXOs using the /getutxos URI with a binary response")
160
161 bin_request = b'\x01\x02'
162 for txid, n in [spending, spent]:
163 bin_request += bytes.fromhex(txid)
164 bin_request += n.to_bytes(4, 'little')
165
166 bin_response = self.test_rest_request("/getutxos", http_method='POST', req_type=ReqType.BIN, body=bin_request, ret_type=RetType.BYTES)
167 chain_height = int.from_bytes(bin_response[0:4], 'little')
168 response_hash = bin_response[4:36][::-1].hex()
169
170 assert_equal(bb_hash, response_hash) # check if getutxo's chaintip during calculation was fine
171 assert_equal(chain_height, 201) # chain height must be 201 (pre-mined chain [200] + generated block [1])
172
173 self.log.info("Test the /getutxos URI with and without /checkmempool")
174 # Create a transaction, check that it's found with /checkmempool, but
175 # not found without. Then confirm the transaction and check that it's
176 # found with or without /checkmempool.
177
178 # do a tx and don't sync
179 txid = self.wallet.send_to(from_node=self.nodes[0], scriptPubKey=getnewdestination()[1], amount=int(0.1 * COIN))["txid"]
180 json_obj = self.test_rest_request(f"/tx/{txid}")
181 # get the spent output to later check for utxo (should be spent by then)
182 spent = (json_obj['vin'][0]['txid'], json_obj['vin'][0]['vout'])
183 # get n of 0.1 outpoint
184 n, = filter_output_indices_by_value(json_obj['vout'], Decimal('0.1'))
185 spending = (txid, n)
186
187 json_obj = self.test_rest_request(f"/getutxos/{spending[0]}-{spending[1]}")
188 assert_equal(len(json_obj['utxos']), 0)
189
190 json_obj = self.test_rest_request(f"/getutxos/checkmempool/{spending[0]}-{spending[1]}")
191 assert_equal(len(json_obj['utxos']), 1)
192
193 json_obj = self.test_rest_request(f"/getutxos/{spent[0]}-{spent[1]}")
194 assert_equal(len(json_obj['utxos']), 1)
195
196 json_obj = self.test_rest_request(f"/getutxos/checkmempool/{spent[0]}-{spent[1]}")
197 assert_equal(len(json_obj['utxos']), 0)
198
199 self.generate(self.nodes[0], 1)
200
201 json_obj = self.test_rest_request(f"/getutxos/{spending[0]}-{spending[1]}")
202 assert_equal(len(json_obj['utxos']), 1)
203
204 json_obj = self.test_rest_request(f"/getutxos/checkmempool/{spending[0]}-{spending[1]}")
205 assert_equal(len(json_obj['utxos']), 1)
206
207 self.log.info("Check some invalid requests")
208 self.test_rest_request("/getutxos", http_method='POST', req_type=ReqType.JSON, body='{"checkmempool', status=400, ret_type=RetType.OBJ)
209 self.test_rest_request("/getutxos", http_method='POST', req_type=ReqType.BIN, body='{"checkmempool', status=400, ret_type=RetType.OBJ)
210 self.test_rest_request("/getutxos/checkmempool", http_method='POST', req_type=ReqType.JSON, status=400, ret_type=RetType.OBJ)
211 self.test_rest_request(f"/getutxos/{spending[0]}_+1", ret_type=RetType.OBJ, status=400)
212 self.test_rest_request(f"/getutxos/{spending[0]}-+1", ret_type=RetType.OBJ, status=400)
213 self.test_rest_request(f"/getutxos/{spending[0]}--1", ret_type=RetType.OBJ, status=400)
214 self.test_rest_request(f"/getutxos/{spending[0]}aa-1234", ret_type=RetType.OBJ, status=400)
215 self.test_rest_request("/getutxos/aa-1234", ret_type=RetType.OBJ, status=400)
216
217 # Test limits
218 long_uri = '/'.join([f"{txid}-{n_}" for n_ in range(20)])
219 self.test_rest_request(f"/getutxos/checkmempool/{long_uri}", http_method='POST', status=400, ret_type=RetType.OBJ)
220
221 long_uri = '/'.join([f'{txid}-{n_}' for n_ in range(15)])
222 self.test_rest_request(f"/getutxos/checkmempool/{long_uri}", http_method='POST', status=200)
223
224 self.generate(self.nodes[0], 1) # generate block to not affect upcoming tests
225
226 self.log.info("Test the /block, /blockhashbyheight, /headers, and /blockfilterheaders URIs")
227 bb_hash = self.nodes[0].getbestblockhash()
228
229 # Check result if block does not exists
230 assert_equal(self.test_rest_request(f"/headers/{UNKNOWN_PARAM}", query_params={"count": 1}), [])
231 self.test_rest_request(f"/block/{UNKNOWN_PARAM}", status=404, ret_type=RetType.OBJ)
232
233 # Check result if block is not in the active chain
234 self.nodes[0].invalidateblock(bb_hash)
235 assert_equal(self.test_rest_request(f'/headers/{bb_hash}', query_params={'count': 1}), [])
236 self.test_rest_request(f'/block/{bb_hash}')
237 self.nodes[0].reconsiderblock(bb_hash)
238
239 # Check binary format
240 response = self.test_rest_request(f"/block/{bb_hash}", req_type=ReqType.BIN, ret_type=RetType.OBJ)
241 assert_greater_than(int(response.getheader('content-length')), BLOCK_HEADER_SIZE)
242 response_bytes = response.read()
243
244 # Compare with block header
245 response_header = self.test_rest_request(f"/headers/{bb_hash}", req_type=ReqType.BIN, ret_type=RetType.OBJ, query_params={"count": 1})
246 assert_equal(int(response_header.getheader('content-length')), BLOCK_HEADER_SIZE)
247 response_header_bytes = response_header.read()
248 assert_equal(response_bytes[:BLOCK_HEADER_SIZE], response_header_bytes)
249
250 # Check block hex format
251 response_hex = self.test_rest_request(f"/block/{bb_hash}", req_type=ReqType.HEX, ret_type=RetType.OBJ)
252 assert_greater_than(int(response_hex.getheader('content-length')), BLOCK_HEADER_SIZE*2)
253 response_hex_bytes = response_hex.read().strip(b'\n')
254 assert_equal(response_bytes.hex().encode(), response_hex_bytes)
255
256 # Compare with hex block header
257 response_header_hex = self.test_rest_request(f"/headers/{bb_hash}", req_type=ReqType.HEX, ret_type=RetType.OBJ, query_params={"count": 1})
258 assert_greater_than(int(response_header_hex.getheader('content-length')), BLOCK_HEADER_SIZE*2)
259 response_header_hex_bytes = response_header_hex.read(BLOCK_HEADER_SIZE*2)
260 assert_equal(response_bytes[:BLOCK_HEADER_SIZE].hex().encode(), response_header_hex_bytes)
261
262 # Check json format
263 block_json_obj = self.test_rest_request(f"/block/{bb_hash}")
264 assert_equal(block_json_obj['hash'], bb_hash)
265 assert_equal(self.test_rest_request(f"/blockhashbyheight/{block_json_obj['height']}")['blockhash'], bb_hash)
266
267 # Check hex/bin format
268 resp_hex = self.test_rest_request(f"/blockhashbyheight/{block_json_obj['height']}", req_type=ReqType.HEX, ret_type=RetType.OBJ)
269 assert_equal(resp_hex.read().decode('utf-8').rstrip(), bb_hash)
270 resp_bytes = self.test_rest_request(f"/blockhashbyheight/{block_json_obj['height']}", req_type=ReqType.BIN, ret_type=RetType.BYTES)
271 blockhash = resp_bytes[::-1].hex()
272 assert_equal(blockhash, bb_hash)
273
274 # Check invalid blockhashbyheight requests
275 resp = self.test_rest_request(f"/blockhashbyheight/{INVALID_PARAM}", ret_type=RetType.OBJ, status=400)
276 assert_equal(resp.read().decode('utf-8').rstrip(), f"Invalid height: {INVALID_PARAM}")
277 resp = self.test_rest_request("/blockhashbyheight/+1", ret_type=RetType.OBJ, status=400)
278 assert_equal(resp.read().decode('utf-8').rstrip(), "Invalid height: +1")
279 resp = self.test_rest_request("/blockhashbyheight/1000000", ret_type=RetType.OBJ, status=404)
280 assert_equal(resp.read().decode('utf-8').rstrip(), "Block height out of range")
281 resp = self.test_rest_request("/blockhashbyheight/-1", ret_type=RetType.OBJ, status=400)
282 assert_equal(resp.read().decode('utf-8').rstrip(), "Invalid height: -1")
283 self.test_rest_request("/blockhashbyheight/", ret_type=RetType.OBJ, status=400)
284
285 # Compare with json block header
286 json_obj = self.test_rest_request(f"/headers/{bb_hash}", query_params={"count": 1})
287 assert_equal(len(json_obj), 1) # ensure that there is one header in the json response
288 assert_equal(json_obj[0]['hash'], bb_hash) # request/response hash should be the same
289
290 # Check tolerance for invalid URI (% symbol at the end of the request)
291 for invalid_uri in [f"/headers/{bb_hash}%", f"/blockfilterheaders/basic/{bb_hash}%"]:
292 resp = self.test_rest_request(invalid_uri, ret_type=RetType.OBJ, status=400)
293 assert_equal(resp.read().decode('utf-8').rstrip(), f"Invalid hash: {bb_hash}%")
294 resp = self.test_rest_request("/mempool/contents.json?%", ret_type=RetType.OBJ, status=200)
295 assert_equal(resp.read().decode('utf-8').rstrip(), "{}")
296
297 # Compare with normal RPC block response
298 rpc_block_json = self.nodes[0].getblock(bb_hash)
299 for key in ['hash', 'confirmations', 'height', 'version', 'merkleroot', 'time', 'nonce', 'bits', 'target', 'difficulty', 'chainwork', 'previousblockhash']:
300 assert_equal(json_obj[0][key], rpc_block_json[key])
301
302 # See if we can get 5 headers in one response
303 self.generate(self.nodes[1], 5)
304 expected_filter = {
305 'basic block filter index': {'synced': True, 'best_block_height': 208},
306 }
307 self.wait_until(lambda: self.nodes[0].getindexinfo() == expected_filter)
308 json_obj = self.test_rest_request(f"/headers/{bb_hash}", query_params={"count": 5})
309 assert_equal(len(json_obj), 5) # now we should have 5 header objects
310 json_obj = self.test_rest_request(f"/blockfilterheaders/basic/{bb_hash}", query_params={"count": 5})
311 first_filter_header = json_obj[0]
312 assert_equal(len(json_obj), 5) # now we should have 5 filter header objects
313 json_obj = self.test_rest_request(f"/blockfilter/basic/{bb_hash}")
314
315 # Compare with normal RPC blockfilter response
316 rpc_blockfilter = self.nodes[0].getblockfilter(bb_hash)
317 assert_equal(first_filter_header, rpc_blockfilter['header'])
318 assert_equal(json_obj['filter'], rpc_blockfilter['filter'])
319
320 # Test blockfilterheaders with an invalid hash and filtertype
321 resp = self.test_rest_request(f"/blockfilterheaders/{INVALID_PARAM}/{bb_hash}", ret_type=RetType.OBJ, status=400)
322 assert_equal(resp.read().decode('utf-8').rstrip(), f"Unknown filtertype {INVALID_PARAM}")
323 resp = self.test_rest_request(f"/blockfilterheaders/basic/{INVALID_PARAM}", ret_type=RetType.OBJ, status=400)
324 assert_equal(resp.read().decode('utf-8').rstrip(), f"Invalid hash: {INVALID_PARAM}")
325
326 # Test number parsing
327 for num in ['5a', '-5', '0', '2001', '99999999999999999999999999999999999']:
328 assert_equal(
329 bytes(f'Header count is invalid or out of acceptable range (1-2000): {num}\r\n', 'ascii'),
330 self.test_rest_request(f"/headers/{bb_hash}", ret_type=RetType.BYTES, status=400, query_params={"count": num}),
331 )
332
333 self.log.info("Test tx inclusion in the /mempool and /block URIs")
334
335 # Make 3 chained txs and mine them on node 1
336 txs = []
337 input_txid = txid
338 for _ in range(3):
339 utxo_to_spend = self.wallet.get_utxo(txid=input_txid)
340 txs.append(self.wallet.send_self_transfer(from_node=self.nodes[0], utxo_to_spend=utxo_to_spend)['txid'])
341 input_txid = txs[-1]
342 self.sync_all()
343
344 # Check that there are exactly 3 transactions in the TX memory pool before generating the block
345 json_obj = self.test_rest_request("/mempool/info")
346 assert_equal(json_obj['size'], 3)
347 # the size of the memory pool should be greater than 3x ~100 bytes
348 assert_greater_than(json_obj['bytes'], 300)
349
350 mempool_info = self.nodes[0].getmempoolinfo()
351 # pop unstable unbroadcastcount before check
352 for obj in [json_obj, mempool_info]:
353 obj.pop("unbroadcastcount")
354 assert_equal(json_obj, mempool_info)
355
356 # Check that there are our submitted transactions in the TX memory pool
357 json_obj = self.test_rest_request("/mempool/contents")
358 raw_mempool_verbose = self.nodes[0].getrawmempool(verbose=True)
359
360 assert_equal(json_obj, raw_mempool_verbose)
361
362 for i, tx in enumerate(txs):
363 assert tx in json_obj
364 assert_equal(json_obj[tx]['spentby'], txs[i + 1:i + 2])
365 assert_equal(json_obj[tx]['depends'], txs[i - 1:i])
366
367 # Check the mempool response for explicit parameters
368 json_obj = self.test_rest_request("/mempool/contents", query_params={"verbose": "true", "mempool_sequence": "false"})
369 assert_equal(json_obj, raw_mempool_verbose)
370
371 # Check the mempool response for not verbose
372 json_obj = self.test_rest_request("/mempool/contents", query_params={"verbose": "false"})
373 raw_mempool = self.nodes[0].getrawmempool(verbose=False)
374
375 assert_equal(json_obj, raw_mempool)
376
377 # Check the mempool response for sequence
378 json_obj = self.test_rest_request("/mempool/contents", query_params={"verbose": "false", "mempool_sequence": "true"})
379 raw_mempool = self.nodes[0].getrawmempool(verbose=False, mempool_sequence=True)
380
381 assert_equal(json_obj, raw_mempool)
382
383 # Check for error response if verbose=true and mempool_sequence=true
384 resp = self.test_rest_request("/mempool/contents", ret_type=RetType.OBJ, status=400, query_params={"verbose": "true", "mempool_sequence": "true"})
385 assert_equal(resp.read().decode('utf-8').strip(), 'Verbose results cannot contain mempool sequence values. (hint: set "verbose=false")')
386
387 # Check for error response if verbose is not "true" or "false"
388 resp = self.test_rest_request("/mempool/contents", ret_type=RetType.OBJ, status=400, query_params={"verbose": "TRUE"})
389 assert_equal(resp.read().decode('utf-8').strip(), 'The "verbose" query parameter must be either "true" or "false".')
390
391 # Check for error response if mempool_sequence is not "true" or "false"
392 resp = self.test_rest_request("/mempool/contents", ret_type=RetType.OBJ, status=400, query_params={"verbose": "false", "mempool_sequence": "TRUE"})
393 assert_equal(resp.read().decode('utf-8').strip(), 'The "mempool_sequence" query parameter must be either "true" or "false".')
394
395 # Now mine the transactions
396 newblockhash = self.generate(self.nodes[1], 1)
397
398 # Check if the 3 tx show up in the new block
399 json_obj = self.test_rest_request(f"/block/{newblockhash[0]}")
400 non_coinbase_txs = {tx['txid'] for tx in json_obj['tx']
401 if 'coinbase' not in tx['vin'][0]}
402 assert_equal(non_coinbase_txs, set(txs))
403
404 # Verify that the non-coinbase tx has "prevout" key set
405 for tx_obj in json_obj["tx"]:
406 for vin in tx_obj["vin"]:
407 if "coinbase" not in vin:
408 assert "prevout" in vin
409 assert_equal(vin["prevout"]["generated"], False)
410 else:
411 assert "prevout" not in vin
412
413 # Check the same but without tx details
414 json_obj = self.test_rest_request(f"/block/notxdetails/{newblockhash[0]}")
415 for tx in txs:
416 assert tx in json_obj['tx']
417
418 self.log.info("Test the /chaininfo URI")
419
420 bb_hash = self.nodes[0].getbestblockhash()
421
422 json_obj = self.test_rest_request("/chaininfo")
423 assert_equal(json_obj['bestblockhash'], bb_hash)
424
425 # Compare with normal RPC getblockchaininfo response
426 blockchain_info = self.nodes[0].getblockchaininfo()
427 assert_equal(blockchain_info, json_obj)
428
429 # Test compatibility of deprecated and newer endpoints
430 self.log.info("Test compatibility of deprecated and newer endpoints")
431 assert_equal(self.test_rest_request(f"/headers/{bb_hash}", query_params={"count": 1}), self.test_rest_request(f"/headers/1/{bb_hash}"))
432 assert_equal(self.test_rest_request(f"/blockfilterheaders/basic/{bb_hash}", query_params={"count": 1}), self.test_rest_request(f"/blockfilterheaders/basic/5/{bb_hash}"))
433
434 self.log.info("Test the /spenttxouts URI")
435
436 block_count = self.nodes[0].getblockcount()
437 for height in range(0, block_count + 1):
438 blockhash = self.nodes[0].getblockhash(height)
439 spent_bin = self.test_rest_request(f"/spenttxouts/{blockhash}", req_type=ReqType.BIN, ret_type=RetType.BYTES)
440 spent_hex = self.test_rest_request(f"/spenttxouts/{blockhash}", req_type=ReqType.HEX, ret_type=RetType.BYTES)
441 spent_json = self.test_rest_request(f"/spenttxouts/{blockhash}", req_type=ReqType.JSON, ret_type=RetType.JSON)
442
443 assert_equal(bytes.fromhex(spent_hex.decode()), spent_bin)
444
445 spent = deser_block_spent_outputs(BytesIO(spent_bin))
446 block = self.nodes[0].getblock(blockhash, 3) # return prevout for each input
447 assert_equal(len(spent), len(block["tx"]))
448 assert_equal(len(spent_json), len(block["tx"]))
449
450 for i, tx in enumerate(block["tx"]):
451 prevouts = [txin["prevout"] for txin in tx["vin"] if "coinbase" not in txin]
452 # compare with `getblock` JSON output (coinbase tx has no prevouts)
453 actual = [(txout.scriptPubKey.hex(), Decimal(txout.nValue) / COIN) for txout in spent[i]]
454 expected = [(p["scriptPubKey"]["hex"], p["value"]) for p in prevouts]
455 assert_equal(expected, actual)
456 # also compare JSON format
457 actual = [(prevout["scriptPubKey"], prevout["value"]) for prevout in spent_json[i]]
458 expected = [(p["scriptPubKey"], p["value"]) for p in prevouts]
459 assert_equal(expected, actual)
460
461 self.log.info("Test the /blockpart URI")
462
463 blockhash = self.nodes[0].getbestblockhash()
464 block_bin = self.test_rest_request(f"/block/{blockhash}", req_type=ReqType.BIN, ret_type=RetType.BYTES)
465 for req_type in (ReqType.BIN, ReqType.HEX):
466 def get_block_part(status: int = 200, **kwargs):
467 resp = self.test_rest_request(f"/blockpart/{blockhash}", status=status,
468 req_type=req_type, ret_type=RetType.BYTES, **kwargs)
469 assert isinstance(resp, bytes)
470 if req_type is ReqType.HEX and status == 200:
471 resp = bytes.fromhex(resp.decode().strip())
472 return resp
473
474 assert_equal(block_bin, get_block_part(query_params={"offset": 0, "size": len(block_bin)}))
475
476 assert len(block_bin) >= 500
477 assert_equal(block_bin[20:320], get_block_part(query_params={"offset": 20, "size": 300}))
478 assert_equal(block_bin[-5:], get_block_part(query_params={"offset": len(block_bin) - 5, "size": 5}))
479
480 get_block_part(status=400, query_params={"offset": 10})
481 get_block_part(status=400, query_params={"size": 100})
482 get_block_part(status=400, query_params={"offset": "x"})
483 get_block_part(status=400, query_params={"size": "y"})
484 get_block_part(status=400, query_params={"offset": "x", "size": "y"})
485 get_block_part(status=400, query_params="%XY")
486 get_block_part(status=400, query_params={"offset": 0, "size": 0})
487 get_block_part(status=400, query_params={"offset": len(block_bin), "size": 0})
488 get_block_part(status=400, query_params={"offset": len(block_bin), "size": 1})
489 get_block_part(status=400, query_params={"offset": len(block_bin) + 1, "size": 1})
490 get_block_part(status=400, query_params={"offset": 0, "size": len(block_bin) + 1})
491
492 res = self.test_rest_request(f"/blockpart/{blockhash}", status=400, req_type=ReqType.BIN, ret_type=RetType.OBJ)
493 assert res.read().decode().startswith("Block part offset missing or invalid")
494
495 res = self.test_rest_request(f"/blockpart/{blockhash}", query_params={"offset":0, "size":1}, status=400, req_type=ReqType.JSON, ret_type=RetType.OBJ)
496 assert res.read().decode().startswith("JSON output is not supported for this request type")
497
498 self.log.info("Missing block data should cause REST API to fail")
499
500 self.test_rest_request(f"/block/{blockhash}", status=200, req_type=ReqType.BIN, ret_type=RetType.OBJ)
501 self.test_rest_request(f"/blockpart/{blockhash}", query_params={"offset": 0, "size": 1}, status=200, req_type=ReqType.BIN, ret_type=RetType.OBJ)
502 blk_files = list(self.nodes[0].blocks_path.glob("blk*.dat"))
503 for blk_file in blk_files:
504 blk_file.rename(blk_file.with_suffix('.bkp'))
505 self.test_rest_request(f"/block/{blockhash}", status=500, req_type=ReqType.BIN, ret_type=RetType.OBJ)
506 self.test_rest_request(f"/blockpart/{blockhash}", query_params={"offset": 0, "size": 1}, status=500, req_type=ReqType.BIN, ret_type=RetType.OBJ)
507 for blk_file in blk_files:
508 blk_file.with_suffix('.bkp').rename(blk_file)
509
510 self.log.info("Test the /deploymentinfo URI")
511
512 deployment_info = self.nodes[0].getdeploymentinfo()
513 assert_equal(deployment_info, self.test_rest_request('/deploymentinfo'))
514
515 previous_bb_hash = self.nodes[0].getblockhash(self.nodes[0].getblockcount() - 1)
516 deployment_info = self.nodes[0].getdeploymentinfo(previous_bb_hash)
517 assert_equal(deployment_info, self.test_rest_request(f"/deploymentinfo/{previous_bb_hash}"))
518
519 non_existing_blockhash = '42759cde25462784395a337460bde75f58e73d3f08bd31fdc3507cbac856a2c4'
520 resp = self.test_rest_request(f'/deploymentinfo/{non_existing_blockhash}', ret_type=RetType.OBJ, status=400)
521 assert_equal(resp.read().decode('utf-8').rstrip(), "Block not found")
522
523 resp = self.test_rest_request(f"/deploymentinfo/{INVALID_PARAM}", ret_type=RetType.OBJ, status=400)
524 assert_equal(resp.read().decode('utf-8').rstrip(), f"Invalid hash: {INVALID_PARAM}")
525
526 if __name__ == '__main__':
527 RESTTest(__file__).main()
528