BTCPrivate
/
BTCP-Rebase
mirror of https://github.com/BTCPrivate/BTCP-Rebase.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

[tests] refactor interface_rest.py to avoid code repetition 

Also refactor txout index parsing and formatting.
This commit is contained in:
John Newbery 2018-03-22 16:37:09 -04:00 committed by Roman Zeyde
parent 7a3181a767
commit ad00fbed3c
No known key found for this signature in database
GPG Key ID: 87CAE5FA46917CBB
1 changed files with 91 additions and 124 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

215
test/functional/interface_rest.py
View File

@ -4,6 +4,7 @@
# 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 decimal import Decimal from decimal import Decimal
from enum import Enum
from io import BytesIO from io import BytesIO
import json import json
from codecs import encode from codecs import encode
@ -21,25 +22,20 @@ from test_framework.util import (
hex_str_to_bytes, hex_str_to_bytes,
) )
def http_get_call(host, port, path, response_object=0): class ReqType(Enum):
"""Make a simple HTTP GET request.""" JSON = 1
conn = http.client.HTTPConnection(host, port) BIN = 2
conn.request('GET', path) HEX = 3
if response_object: class RetType(Enum):
return conn.getresponse() OBJ = 1
BYTES = 2
JSON = 3
return conn.getresponse().read().decode('utf-8') def filter_output_indices_by_value(vouts, value):
for vout in vouts:
def http_post_call(host, port, path, requestdata='', response_object=0): if vout['value'] == value:
"""Make a simple HTTP POST request with a request body.""" yield vout['n']
conn = http.client.HTTPConnection(host, port)
conn.request('POST', path, requestdata)
if response_object:
return conn.getresponse()
return conn.getresponse().read()
class RESTTest (BitcoinTestFramework): class RESTTest (BitcoinTestFramework):
def set_test_params(self): def set_test_params(self):
@ -50,10 +46,35 @@ class RESTTest (BitcoinTestFramework):
def setup_network(self, split=False): def setup_network(self, split=False):
super().setup_network() super().setup_network()
connect_nodes_bi(self.nodes, 0, 2) connect_nodes_bi(self.nodes, 0, 2)
self.url = urllib.parse.urlparse(self.nodes[0].url)
def test_rest_request(self, uri, http_method='GET', req_type=ReqType.JSON, body='', status=200, ret_type=RetType.JSON):
rest_uri = '/rest' + uri
if req_type == ReqType.JSON:
rest_uri += '.json'
elif req_type == ReqType.BIN:
rest_uri += '.bin'
elif req_type == ReqType.HEX:
rest_uri += '.hex'
conn = http.client.HTTPConnection(self.url.hostname, self.url.port)
self.log.debug('%s %s %s', http_method, rest_uri, body)
if http_method == 'GET':
conn.request('GET', rest_uri)
elif http_method == 'POST':
conn.request('POST', rest_uri, body)
resp = conn.getresponse()
assert_equal(resp.status, status)
if ret_type == RetType.OBJ:
return resp
elif ret_type == RetType.BYTES:
return resp.read()
elif ret_type == RetType.JSON:
return json.loads(resp.read().decode('utf-8'), parse_float=Decimal)
def run_test(self): def run_test(self):
url = urllib.parse.urlparse(self.nodes[0].url)
self.log.info("Mine blocks and send Bitcoin to node 1") self.log.info("Mine blocks and send Bitcoin to node 1")
self.nodes[0].generate(1) self.nodes[0].generate(1)
@ -73,39 +94,31 @@ class RESTTest (BitcoinTestFramework):
self.log.info("Load the transaction using the /tx URI") self.log.info("Load the transaction using the /tx URI")
json_request = "/rest/tx/{}.json".format(txid) json_obj = self.test_rest_request("/tx/{}".format(txid))
json_string = http_get_call(url.hostname, url.port, json_request) spent = (json_obj['vin'][0]['txid'], json_obj['vin'][0]['vout']) # get the vin to later check for utxo (should be spent by then)
json_obj = json.loads(json_string)
vintx = json_obj['vin'][0]['txid'] # get the vin to later check for utxo (should be spent by then)
# get n of 0.1 outpoint # get n of 0.1 outpoint
n = 0 n, = filter_output_indices_by_value(json_obj['vout'], Decimal('0.1'))
for vout in json_obj['vout']: spending = (txid, n)
if vout['value'] == 0.1:
n = vout['n']
self.log.info("Query an unspent TXO using the /getutxos URI") self.log.info("Query an unspent TXO using the /getutxos URI")
json_request = "/rest/getutxos/{}-{}.json".format(txid, str(n)) json_obj = self.test_rest_request("/getutxos/{}-{}".format(*spending))
json_string = http_get_call(url.hostname, url.port, json_request)
json_obj = json.loads(json_string)
# Check chainTip response # Check chainTip response
assert_equal(json_obj['chaintipHash'], bb_hash) assert_equal(json_obj['chaintipHash'], bb_hash)
# Make sure there is one utxo # Make sure there is one utxo
assert_equal(len(json_obj['utxos']), 1) assert_equal(len(json_obj['utxos']), 1)
assert_equal(json_obj['utxos'][0]['value'], 0.1) assert_equal(json_obj['utxos'][0]['value'], Decimal('0.1'))
self.log.info("Query a spent TXO using the /getutxos URI") self.log.info("Query a spent TXO using the /getutxos URI")
json_request = "/rest/getutxos/{}-0.json".format(vintx) json_obj = self.test_rest_request("/getutxos/{}-{}".format(*spent))
json_string = http_get_call(url.hostname, url.port, json_request)
json_obj = json.loads(json_string)
# Check chainTip response # Check chainTip response
assert_equal(json_obj['chaintipHash'], bb_hash) assert_equal(json_obj['chaintipHash'], bb_hash)
# Make sure there is no utxo in the response because this oupoint has been spent # Make sure there is no utxo in the response because this outpoint has been spent
assert_equal(len(json_obj['utxos']), 0) assert_equal(len(json_obj['utxos']), 0)
# Check bitmap # Check bitmap
@ -113,9 +126,8 @@ class RESTTest (BitcoinTestFramework):
self.log.info("Query two TXOs using the /getutxos URI") self.log.info("Query two TXOs using the /getutxos URI")
json_request = "/rest/getutxos/{}-{}/{}-0.json".format(txid, str(n), vintx) json_obj = self.test_rest_request("/getutxos/{}-{}/{}-{}".format(*(spending + spent)))
json_string = http_get_call(url.hostname, url.port, json_request)
json_obj = json.loads(json_string)
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")
@ -124,12 +136,11 @@ class RESTTest (BitcoinTestFramework):
bb_hash = self.nodes[0].getbestblockhash() bb_hash = self.nodes[0].getbestblockhash()
bin_request = b'\x01\x02' bin_request = b'\x01\x02'
bin_request += hex_str_to_bytes(txid) for txid, n in [spending, spent]:
bin_request += pack("i", n) bin_request += hex_str_to_bytes(txid)
bin_request += hex_str_to_bytes(vintx) bin_request += pack("i", n)
bin_request += pack("i", 0)
bin_response = http_post_call(url.hostname, url.port, '/rest/getutxos.bin', bin_request) bin_response = self.test_rest_request("/getutxos", http_method='POST', req_type=ReqType.BIN, body=bin_request, ret_type=RetType.BYTES)
output = BytesIO() output = BytesIO()
output.write(bin_response) output.write(bin_response)
output.seek(0) output.seek(0)
@ -146,71 +157,45 @@ class RESTTest (BitcoinTestFramework):
# do a tx and don't sync # do a tx and don't sync
txid = self.nodes[0].sendtoaddress(self.nodes[1].getnewaddress(), 0.1) txid = self.nodes[0].sendtoaddress(self.nodes[1].getnewaddress(), 0.1)
json_request = "/rest/tx/{}.json".format(txid) json_obj = self.test_rest_request("/tx/{}".format(txid))
json_string = http_get_call(url.hostname, url.port, json_request)
json_obj = json.loads(json_string)
# get the spent output to later check for utxo (should be spent by then) # get the spent output to later check for utxo (should be spent by then)
spent = '{}-{}'.format(json_obj['vin'][0]['txid'], json_obj['vin'][0]['vout']) spent = (json_obj['vin'][0]['txid'], json_obj['vin'][0]['vout'])
# get n of 0.1 outpoint # get n of 0.1 outpoint
n = 0 n, = filter_output_indices_by_value(json_obj['vout'], Decimal('0.1'))
for vout in json_obj['vout']: spending = (txid, n)
if vout['value'] == 0.1:
n = vout['n']
spending = '{}-{}'.format(txid, n)
json_request = '/rest/getutxos/{}.json'.format(spending) json_obj = self.test_rest_request("/getutxos/{}-{}".format(*spending))
json_string = http_get_call(url.hostname, url.port, json_request)
json_obj = json.loads(json_string)
assert_equal(len(json_obj['utxos']), 0) assert_equal(len(json_obj['utxos']), 0)
json_request = '/rest/getutxos/checkmempool/{}.json'.format(spending) json_obj = self.test_rest_request("/getutxos/checkmempool/{}-{}".format(*spending))
json_string = http_get_call(url.hostname, url.port, json_request)
json_obj = json.loads(json_string)
assert_equal(len(json_obj['utxos']), 1) assert_equal(len(json_obj['utxos']), 1)
json_request = '/rest/getutxos/{}.json'.format(spent) json_obj = self.test_rest_request("/getutxos/{}-{}".format(*spent))
json_string = http_get_call(url.hostname, url.port, json_request)
json_obj = json.loads(json_string)
assert_equal(len(json_obj['utxos']), 1) assert_equal(len(json_obj['utxos']), 1)
json_request = '/rest/getutxos/checkmempool/{}.json'.format(spent) json_obj = self.test_rest_request("/getutxos/checkmempool/{}-{}".format(*spent))
json_string = http_get_call(url.hostname, url.port, json_request)
json_obj = json.loads(json_string)
assert_equal(len(json_obj['utxos']), 0) assert_equal(len(json_obj['utxos']), 0)
self.nodes[0].generate(1) self.nodes[0].generate(1)
self.sync_all() self.sync_all()
json_request = '/rest/getutxos/{}.json'.format(spending) json_obj = self.test_rest_request("/getutxos/{}-{}".format(*spending))
json_string = http_get_call(url.hostname, url.port, json_request)
json_obj = json.loads(json_string)
assert_equal(len(json_obj['utxos']), 1) assert_equal(len(json_obj['utxos']), 1)
json_request = '/rest/getutxos/checkmempool/{}.json'.format(spending) json_obj = self.test_rest_request("/getutxos/checkmempool/{}-{}".format(*spending))
json_string = http_get_call(url.hostname, url.port, json_request)
json_obj = json.loads(json_string)
assert_equal(len(json_obj['utxos']), 1) assert_equal(len(json_obj['utxos']), 1)
# Do some invalid requests # Do some invalid requests
json_request = '{"checkmempool' self.test_rest_request("/getutxos", http_method='POST', req_type=ReqType.JSON, body='{"checkmempool', status=400, ret_type=RetType.OBJ)
response = http_post_call(url.hostname, url.port, '/rest/getutxos.json', json_request, True) self.test_rest_request("/getutxos", http_method='POST', req_type=ReqType.BIN, body='{"checkmempool', status=400, ret_type=RetType.OBJ)
assert_equal(response.status, 400) # must be a 400 because we send an invalid json request self.test_rest_request("/getutxos/checkmempool", http_method='POST', req_type=ReqType.JSON, status=400, ret_type=RetType.OBJ)
json_request = '{"checkmempool'
response = http_post_call(url.hostname, url.port, '/rest/getutxos.bin', json_request, True)
assert_equal(response.status, 400) # must be a 400 because we send an invalid bin request
response = http_post_call(url.hostname, url.port, '/rest/getutxos/checkmempool.json', '', True)
assert_equal(response.status, 400) # must be a 400 because we send an invalid bin request
# Test limits # Test limits
json_request = '/rest/getutxos/checkmempool/' + '/'.join(["{}-{}".format(txid, n) for n in range(20)]) + '.json' long_uri = '/'.join(["{}-{}".format(txid, n) for n in range(20)])
response = http_post_call(url.hostname, url.port, json_request, '', True) self.test_rest_request("/getutxos/checkmempool/{}".format(long_uri), http_method='POST', status=400, ret_type=RetType.OBJ)
assert_equal(response.status, 400) # must be a 400 because we exceeding the limits
json_request = '/rest/getutxos/checkmempool/' + '/'.join(['{}-{}'.format(txid, n) for n in range(15)]) + '.json' long_uri = '/'.join(['{}-{}'.format(txid, n) for n in range(15)])
response = http_post_call(url.hostname, url.port, json_request, '', True) self.test_rest_request("/getutxos/checkmempool/{}".format(long_uri), http_method='POST', status=200)
assert_equal(response.status, 200) # must be a 200 because we are within the limits
self.nodes[0].generate(1) # generate block to not affect upcoming tests self.nodes[0].generate(1) # generate block to not affect upcoming tests
self.sync_all() self.sync_all()
@ -218,43 +203,35 @@ class RESTTest (BitcoinTestFramework):
self.log.info("Test the /block and /headers URIs") self.log.info("Test the /block and /headers URIs")
# Check binary format # Check binary format
response = http_get_call(url.hostname, url.port, '/rest/block/{}.bin'.format(bb_hash), True) response = self.test_rest_request("/block/{}".format(bb_hash), req_type=ReqType.BIN, ret_type=RetType.OBJ)
assert_equal(response.status, 200)
assert_greater_than(int(response.getheader('content-length')), 80) assert_greater_than(int(response.getheader('content-length')), 80)
response_str = response.read() response_str = response.read()
# Compare with block header # Compare with block header
response_header = http_get_call(url.hostname, url.port, '/rest/headers/1/{}.bin'.format(bb_hash), True) response_header = self.test_rest_request("/headers/1/{}".format(bb_hash), req_type=ReqType.BIN, ret_type=RetType.OBJ)
assert_equal(response_header.status, 200)
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_str = response_header.read()
assert_equal(response_str[0:80], response_header_str) assert_equal(response_str[0:80], response_header_str)
# Check block hex format # Check block hex format
response_hex = http_get_call(url.hostname, url.port, '/rest/block/{}.hex'.format(bb_hash), True) response_hex = self.test_rest_request("/block/{}".format(bb_hash), req_type=ReqType.HEX, ret_type=RetType.OBJ)
assert_equal(response_hex.status, 200)
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_str = response_hex.read()
assert_equal(encode(response_str, "hex_codec")[0:160], response_hex_str[0:160]) assert_equal(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 = http_get_call(url.hostname, url.port, '/rest/headers/1/{}.hex'.format(bb_hash), True) response_header_hex = self.test_rest_request("/headers/1/{}".format(bb_hash), req_type=ReqType.HEX, ret_type=RetType.OBJ)
assert_equal(response_header_hex.status, 200)
assert_greater_than(int(response_header_hex.getheader('content-length')), 160) assert_greater_than(int(response_header_hex.getheader('content-length')), 160)
response_header_hex_str = response_header_hex.read() response_header_hex_str = response_header_hex.read()
assert_equal(response_hex_str[0:160], response_header_hex_str[0:160]) assert_equal(response_hex_str[0:160], response_header_hex_str[0:160])
assert_equal(encode(response_header_str, "hex_codec")[0:160], response_header_hex_str[0:160]) assert_equal(encode(response_header_str, "hex_codec")[0:160], response_header_hex_str[0:160])
# Check json format # Check json format
block_json_string = http_get_call(url.hostname, url.port, '/rest/block/{}.json'.format(bb_hash)) block_json_obj = self.test_rest_request("/block/{}".format(bb_hash))
block_json_obj = json.loads(block_json_string)
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
response_header_json = http_get_call(url.hostname, url.port, '/rest/headers/1/{}.json'.format(bb_hash), True) json_obj = self.test_rest_request("/headers/1/{}".format(bb_hash))
assert_equal(response_header_json.status, 200)
response_header_json_str = response_header_json.read().decode('utf-8')
json_obj = json.loads(response_header_json_str, parse_float=Decimal)
assert_equal(len(json_obj), 1) # ensure that there is one header in the json response assert_equal(len(json_obj), 1) # ensure that there is one header in the json response
assert_equal(json_obj[0]['hash'], bb_hash) # request/response hash should be the same assert_equal(json_obj[0]['hash'], bb_hash) # request/response hash should be the same
@ -266,23 +243,18 @@ class RESTTest (BitcoinTestFramework):
# See if we can get 5 headers in one response # See if we can get 5 headers in one response
self.nodes[1].generate(5) self.nodes[1].generate(5)
self.sync_all() self.sync_all()
response_header_json = http_get_call(url.hostname, url.port, '/rest/headers/5/{}.json'.format(bb_hash), True) json_obj = self.test_rest_request("/headers/5/{}".format(bb_hash))
assert_equal(response_header_json.status, 200)
response_header_json_str = response_header_json.read().decode('utf-8')
json_obj = json.loads(response_header_json_str)
assert_equal(len(json_obj), 5) # now we should have 5 header objects assert_equal(len(json_obj), 5) # now we should have 5 header objects
self.log.info("Test the /tx URI") self.log.info("Test the /tx URI")
tx_hash = block_json_obj['tx'][0]['txid'] tx_hash = block_json_obj['tx'][0]['txid']
json_string = http_get_call(url.hostname, url.port, '/rest/tx/{}.json'.format(tx_hash)) json_obj = self.test_rest_request("/tx/{}".format(tx_hash))
json_obj = json.loads(json_string)
assert_equal(json_obj['txid'], tx_hash) assert_equal(json_obj['txid'], tx_hash)
# Check hex format response # Check hex format response
hex_string = http_get_call(url.hostname, url.port, '/rest/tx/{}.hex'.format(tx_hash), True) hex_response = self.test_rest_request("/tx/{}".format(tx_hash), req_type=ReqType.HEX, ret_type=RetType.OBJ)
assert_equal(hex_string.status, 200) assert_greater_than(int(hex_response.getheader('content-length')), 10)
assert_greater_than(int(response.getheader('content-length')), 10)
self.log.info("Test tx inclusion in the /mempool and /block URIs") self.log.info("Test tx inclusion in the /mempool and /block URIs")
@ -294,17 +266,15 @@ class RESTTest (BitcoinTestFramework):
self.sync_all() self.sync_all()
# Check that there are exactly 3 transactions in the TX memory pool before generating the block # Check that there are exactly 3 transactions in the TX memory pool before generating the block
json_string = http_get_call(url.hostname, url.port, '/rest/mempool/info.json') json_obj = self.test_rest_request("/mempool/info")
json_obj = json.loads(json_string)
assert_equal(json_obj['size'], 3) assert_equal(json_obj['size'], 3)
# the size of the memory pool should be greater than 3x ~100 bytes # the size of the memory pool should be greater than 3x ~100 bytes
assert_greater_than(json_obj['bytes'], 300) assert_greater_than(json_obj['bytes'], 300)
# Check that there are our submitted transactions in the TX memory pool # Check that there are our submitted transactions in the TX memory pool
json_string = http_get_call(url.hostname, url.port, '/rest/mempool/contents.json') json_obj = self.test_rest_request("/mempool/contents")
json_obj = json.loads(json_string)
for i, tx in enumerate(txs): for i, tx in enumerate(txs):
assert_equal(tx in json_obj, True) assert tx in json_obj
assert_equal(json_obj[tx]['spentby'], txs[i + 1:i + 2]) assert_equal(json_obj[tx]['spentby'], txs[i + 1:i + 2])
assert_equal(json_obj[tx]['depends'], txs[i - 1:i]) assert_equal(json_obj[tx]['depends'], txs[i - 1:i])
@ -313,24 +283,21 @@ class RESTTest (BitcoinTestFramework):
self.sync_all() self.sync_all()
# Check if the 3 tx show up in the new block # Check if the 3 tx show up in the new block
json_string = http_get_call(url.hostname, url.port, '/rest/block/{}.json'.format(newblockhash[0])) json_obj = self.test_rest_request("/block/{}".format(newblockhash[0]))
json_obj = json.loads(json_string) non_coinbase_txs = {tx['txid'] for tx in json_obj['tx']
for tx in json_obj['tx']: if 'coinbase' not in tx['vin'][0]}
if 'coinbase' not in tx['vin'][0]: # exclude coinbase assert_equal(non_coinbase_txs, set(txs))
assert_equal(tx['txid'] in txs, True)
# Check the same but without tx details # Check the same but without tx details
json_string = http_get_call(url.hostname, url.port, '/rest/block/notxdetails/{}.json'.format(newblockhash[0])) json_obj = self.test_rest_request("/block/notxdetails/{}".format(newblockhash[0]))
json_obj = json.loads(json_string)
for tx in txs: for tx in txs:
assert_equal(tx in json_obj['tx'], True) assert tx in json_obj['tx']
self.log.info("Test the /chaininfo URI") self.log.info("Test the /chaininfo URI")
bb_hash = self.nodes[0].getbestblockhash() bb_hash = self.nodes[0].getbestblockhash()
json_string = http_get_call(url.hostname, url.port, '/rest/chaininfo.json') json_obj = self.test_rest_request("/chaininfo")
json_obj = json.loads(json_string)
assert_equal(json_obj['bestblockhash'], bb_hash) assert_equal(json_obj['bestblockhash'], bb_hash)
if __name__ == '__main__': if __name__ == '__main__':