123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128 |
- # Copyright IBM Corp, All Rights Reserved.
- #
- # SPDX-License-Identifier: Apache-2.0
- #
- # Licensed under the Apache License, Version 2.0 (the "License");
- # you may not use this file except in compliance with the License.
- # You may obtain a copy of the License at
- # http://www.apache.org/licenses/LICENSE-2.0
- # Unless required by applicable law or agreed to in writing, software
- # distributed under the License is distributed on an "AS IS" BASIS,
- # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- # See the License for the specific language governing permissions and
- # limitations under the License.
- from __future__ import print_function
- from hyperledger.client import Client
- # import base64
- import json
- import sys
- import time
- API_URL = 'http://127.0.0.1:5000'
- def query_value(chaincode_name, arg_list):
- """
- Query a list of values.
- :param chaincode_name: The name of the chaincode.
- :param arg_list: List of arguments.
- :return: A list of values.
- """
- result, resp = [], {}
- print("Query value will try at most 20 times.")
- for arg in arg_list:
- for i in range(20):
- try:
- resp = c.chaincode_query(chaincode_name=chaincode_name,
- function="query",
- args=[arg])
- if resp['result']['status'] == 'OK':
- result.append(resp['result']['message'])
- break
- except KeyError:
- print("Wait 1 seconds for the {0} query".format(i))
- time.sleep(1)
- return result
- # Usage:
- # * python function_test.py [API_URL=http://127.0.0.1:5000] will deploy first
- # * python function_test.py [API_URL=http://127.0.0.1:5000] [chaincode_name]
- # E.g.,
- # "f389486d91f54d1f8775940f24b1d3bd9f8a8e75d364e158ac92328ddacad629607a3c42be156fc4a7da7173adca2ac7d7eef29afc59c6f07f3ad14abee34f68"
- if __name__ == '__main__':
- if len(sys.argv) not in [2, 3]:
- print("Usage: python function_test.py ["
- "API_URL=http://127.0.0.1:5000] [chaincode_name]")
- exit()
- API_URL = sys.argv[1]
- chaincode_name = ""
- if len(sys.argv) == 3:
- chaincode_name = sys.argv[2]
- c = Client(base_url=API_URL)
- print("Checking cluster at {}".format(API_URL))
- if not chaincode_name:
- print(">>>Test: deploy the default chaincode")
- res = c.chaincode_deploy(args=["a", "10000", "b", "20000"])
- chaincode_name = res['result']['message']
- assert res['result']['status'] == 'OK'
- print("Successfully deploy chaincode with returned name = " +
- chaincode_name)
- print("Wait 15 seconds to make sure deployment is done.")
- time.sleep(15)
- print(">>>Check the initial value: a, b")
- values = query_value(chaincode_name, ["a", "b"])
- print(values)
- assert values == ['10000', '20000']
- print(">>>Test: invoke a chaincode: a-->b 1")
- res = c.chaincode_invoke(chaincode_name=chaincode_name, function="invoke",
- args=["a", "b", "1"])
- assert res["result"]["status"] == "OK"
- transaction_uuid = res["result"]["message"]
- print("Transaction id = {0}".format(transaction_uuid))
- # TODO: sleep 3 seconds till invoke done.
- print("Wait 5 seconds to make sure invoke is done.")
- time.sleep(5)
- print(">>>Check the after value: a, b")
- values = query_value(chaincode_name, ["a", "b"])
- print(values)
- assert values == ['9999', '20001']
- time.sleep(5)
- print(">>>Test: Check the transaction content")
- res = c.transaction_get(transaction_uuid)
- # res["chaincodeID"] = base64.b64decode(res["chaincodeID"])
- print(json.dumps(res, sort_keys=True, indent=4))
- assert res["uuid"] == transaction_uuid
- print(">>>Test: list the peers")
- res = c.peer_list()
- print(json.dumps(res, sort_keys=True, indent=4))
- assert len(res['peers']) > 0
- print(">>>Test: list the chain")
- res = c.chain_list()
- print(json.dumps(res, sort_keys=True, indent=4))
- assert res['height'] > 0
- print("Existing block number = {0}".format(res["height"]))
- print(">>>Test: get the content of block 1")
- res = c.block_get(block='1')
- print(json.dumps(res, sort_keys=True, indent=4))
- print(">>>Test: get the content of block 2")
- res = c.block_get(block='2')
- print(json.dumps(res, sort_keys=True, indent=4))
|