Changeset View
Changeset View
Standalone View
Standalone View
test/functional/interface_rest.py
| #!/usr/bin/env python3 | #!/usr/bin/env python3 | ||||
| # Copyright (c) 2014-2016 The Bitcoin Core developers | # Copyright (c) 2014-2016 The Bitcoin Core developers | ||||
| # Distributed under the MIT software license, see the accompanying | # Distributed under the MIT software license, see the accompanying | ||||
| # file COPYING or http://www.opensource.org/licenses/mit-license.php. | # file COPYING or http://www.opensource.org/licenses/mit-license.php. | ||||
| """Test the REST API.""" | """Test the REST API.""" | ||||
| from codecs import encode | import binascii | ||||
| from decimal import Decimal | from decimal import Decimal | ||||
| from enum import Enum | from enum import Enum | ||||
| import http.client | import http.client | ||||
| from io import BytesIO | from io import BytesIO | ||||
| import json | import json | ||||
| from struct import pack, unpack | from struct import pack, unpack | ||||
| import urllib.parse | import urllib.parse | ||||
| from test_framework.messages import deser_uint256 | |||||
| from test_framework.test_framework import BitcoinTestFramework | from test_framework.test_framework import BitcoinTestFramework | ||||
| from test_framework.util import ( | from test_framework.util import ( | ||||
| assert_equal, | assert_equal, | ||||
| assert_greater_than, | assert_greater_than, | ||||
| assert_greater_than_or_equal, | |||||
| hex_str_to_bytes, | hex_str_to_bytes, | ||||
| ) | ) | ||||
| class ReqType(Enum): | class ReqType(Enum): | ||||
| JSON = 1 | JSON = 1 | ||||
| BIN = 2 | BIN = 2 | ||||
| HEX = 3 | HEX = 3 | ||||
| ▲ Show 20 Lines • Show All 105 Lines • ▼ Show 20 Lines | def run_test(self): | ||||
| "/getutxos/{}-{}/{}-{}".format(*(spending + spent))) | "/getutxos/{}-{}/{}-{}".format(*(spending + spent))) | ||||
| assert_equal(len(json_obj['utxos']), 1) | assert_equal(len(json_obj['utxos']), 1) | ||||
| assert_equal(json_obj['bitmap'], "10") | assert_equal(json_obj['bitmap'], "10") | ||||
| self.log.info( | self.log.info( | ||||
| "Query the TXOs using the /getutxos URI with a binary response") | "Query the TXOs using the /getutxos URI with a binary response") | ||||
| bb_hash = self.nodes[0].getbestblockhash() | |||||
| bin_request = b'\x01\x02' | bin_request = b'\x01\x02' | ||||
| for txid, n in [spending, spent]: | for txid, n in [spending, spent]: | ||||
| bin_request += hex_str_to_bytes(txid) | bin_request += hex_str_to_bytes(txid) | ||||
| bin_request += pack("i", n) | bin_request += pack("i", n) | ||||
| bin_response = self.test_rest_request( | bin_response = self.test_rest_request( | ||||
| "/getutxos", http_method='POST', req_type=ReqType.BIN, body=bin_request, ret_type=RetType.BYTES) | "/getutxos", http_method='POST', req_type=ReqType.BIN, body=bin_request, ret_type=RetType.BYTES) | ||||
| output = BytesIO() | output = BytesIO(bin_response) | ||||
| output.write(bin_response) | chain_height, = unpack("i", output.read(4)) | ||||
| output.seek(0) | response_hash = binascii.hexlify(output.read(32)[::-1]).decode('ascii') | ||||
| chain_height = unpack("i", output.read(4))[0] | |||||
| response_hash = hex(deser_uint256(output))[2:].zfill(64) | |||||
| # Check if getutxo's chaintip during calculation was fine | # Check if getutxo's chaintip during calculation was fine | ||||
| assert_equal(bb_hash, response_hash) | assert_equal(bb_hash, response_hash) | ||||
| # Chain height must be 102 | # Chain height must be 102 | ||||
| assert_equal(chain_height, 102) | assert_equal(chain_height, 102) | ||||
| self.log.info("Test the /getutxos URI with and without /checkmempool") | self.log.info("Test the /getutxos URI with and without /checkmempool") | ||||
| # Create a transaction, check that it's found with /checkmempool, but | # Create a transaction, check that it's found with /checkmempool, but | ||||
| ▲ Show 20 Lines • Show All 51 Lines • ▼ Show 20 Lines | def run_test(self): | ||||
| "/getutxos/checkmempool/{}".format(long_uri), http_method='POST', status=200) | "/getutxos/checkmempool/{}".format(long_uri), http_method='POST', status=200) | ||||
| # Generate block to not affect upcoming tests | # Generate block to not affect upcoming tests | ||||
| self.nodes[0].generate( | self.nodes[0].generate( | ||||
| 1) | 1) | ||||
| self.sync_all() | self.sync_all() | ||||
| self.log.info("Test the /block and /headers URIs") | self.log.info("Test the /block and /headers URIs") | ||||
| bb_hash = self.nodes[0].getbestblockhash() | |||||
| # Check binary format | # Check binary format | ||||
| response = self.test_rest_request( | response = self.test_rest_request( | ||||
| "/block/{}".format(bb_hash), req_type=ReqType.BIN, ret_type=RetType.OBJ) | "/block/{}".format(bb_hash), req_type=ReqType.BIN, ret_type=RetType.OBJ) | ||||
| assert_greater_than(int(response.getheader('content-length')), 80) | assert_greater_than(int(response.getheader('content-length')), 80) | ||||
| response_str = response.read() | response_bytes = response.read() | ||||
| # Compare with block header | # Compare with block header | ||||
| response_header = self.test_rest_request( | response_header = self.test_rest_request( | ||||
| "/headers/1/{}".format(bb_hash), req_type=ReqType.BIN, ret_type=RetType.OBJ) | "/headers/1/{}".format(bb_hash), req_type=ReqType.BIN, ret_type=RetType.OBJ) | ||||
| assert_equal(int(response_header.getheader('content-length')), 80) | assert_equal(int(response_header.getheader('content-length')), 80) | ||||
| response_header_str = response_header.read() | response_header_bytes = response_header.read() | ||||
| assert_equal(response_str[0:80], response_header_str) | assert_equal(response_bytes[0:80], response_header_bytes) | ||||
| # Check block hex format | # Check block hex format | ||||
| response_hex = self.test_rest_request( | response_hex = self.test_rest_request( | ||||
| "/block/{}".format(bb_hash), req_type=ReqType.HEX, ret_type=RetType.OBJ) | "/block/{}".format(bb_hash), req_type=ReqType.HEX, ret_type=RetType.OBJ) | ||||
| assert_greater_than(int(response_hex.getheader('content-length')), 160) | assert_greater_than(int(response_hex.getheader('content-length')), 160) | ||||
| response_hex_str = response_hex.read() | response_hex_bytes = response_hex.read().strip(b'\n') | ||||
| assert_equal( | assert_equal(binascii.hexlify(response_bytes), response_hex_bytes) | ||||
| encode(response_str, "hex_codec")[0:160], response_hex_str[0:160]) | |||||
| # Compare with hex block header | # Compare with hex block header | ||||
| response_header_hex = self.test_rest_request( | response_header_hex = self.test_rest_request( | ||||
| "/headers/1/{}".format(bb_hash), req_type=ReqType.HEX, ret_type=RetType.OBJ) | "/headers/1/{}".format(bb_hash), req_type=ReqType.HEX, ret_type=RetType.OBJ) | ||||
| assert_greater_than( | assert_greater_than( | ||||
| int(response_header_hex.getheader('content-length')), 160) | int(response_header_hex.getheader('content-length')), 160) | ||||
| response_header_hex_str = response_header_hex.read() | response_header_hex_bytes = response_header_hex.read(160) | ||||
| assert_equal(response_hex_str[0:160], response_header_hex_str[0:160]) | assert_equal(binascii.hexlify( | ||||
| assert_equal(encode(response_header_str, "hex_codec")[ | response_bytes[:80]), response_header_hex_bytes) | ||||
| 0:160], response_header_hex_str[0:160]) | |||||
| # Check json format | # Check json format | ||||
| block_json_obj = self.test_rest_request("/block/{}".format(bb_hash)) | block_json_obj = self.test_rest_request("/block/{}".format(bb_hash)) | ||||
| assert_equal(block_json_obj['hash'], bb_hash) | assert_equal(block_json_obj['hash'], bb_hash) | ||||
| # Compare with json block header | # Compare with json block header | ||||
| json_obj = self.test_rest_request("/headers/1/{}".format(bb_hash)) | json_obj = self.test_rest_request("/headers/1/{}".format(bb_hash)) | ||||
| # Ensure that there is one header in the json response | # Ensure that there is one header in the json response | ||||
| Show All 17 Lines | def run_test(self): | ||||
| tx_hash = block_json_obj['tx'][0]['txid'] | tx_hash = block_json_obj['tx'][0]['txid'] | ||||
| json_obj = self.test_rest_request("/tx/{}".format(tx_hash)) | json_obj = self.test_rest_request("/tx/{}".format(tx_hash)) | ||||
| assert_equal(json_obj['txid'], tx_hash) | assert_equal(json_obj['txid'], tx_hash) | ||||
| # Check hex format response | # Check hex format response | ||||
| hex_response = self.test_rest_request( | hex_response = self.test_rest_request( | ||||
| "/tx/{}".format(tx_hash), req_type=ReqType.HEX, ret_type=RetType.OBJ) | "/tx/{}".format(tx_hash), req_type=ReqType.HEX, ret_type=RetType.OBJ) | ||||
| assert_greater_than(int(hex_response.getheader('content-length')), 10) | assert_greater_than_or_equal( | ||||
| int(hex_response.getheader('content-length')), json_obj['size']*2) | |||||
| self.log.info("Test tx inclusion in the /mempool and /block URIs") | self.log.info("Test tx inclusion in the /mempool and /block URIs") | ||||
| # Make 3 tx and mine them on node 1 | # Make 3 tx and mine them on node 1 | ||||
| txs = [] | txs = [] | ||||
| txs.append(self.nodes[0].sendtoaddress(not_related_address, 11)) | txs.append(self.nodes[0].sendtoaddress(not_related_address, 11)) | ||||
| txs.append(self.nodes[0].sendtoaddress(not_related_address, 11)) | txs.append(self.nodes[0].sendtoaddress(not_related_address, 11)) | ||||
| txs.append(self.nodes[0].sendtoaddress(not_related_address, 11)) | txs.append(self.nodes[0].sendtoaddress(not_related_address, 11)) | ||||
| ▲ Show 20 Lines • Show All 42 Lines • Show Last 20 Lines | |||||