Changeset View
Changeset View
Standalone View
Standalone View
test/functional/test_framework/util.py
| Show All 31 Lines | def assert_fee_amount(fee, tx_size, fee_per_kB, wiggleroom=2): | ||||
| wiggleroom defines an amount that the test expects the wallet to be off by | wiggleroom defines an amount that the test expects the wallet to be off by | ||||
| when estimating fees. This can be due to the dummy signature that is added | when estimating fees. This can be due to the dummy signature that is added | ||||
| during fee calculation, or due to the wallet funding transactions using the | during fee calculation, or due to the wallet funding transactions using the | ||||
| ceiling of the calculated fee. | ceiling of the calculated fee. | ||||
| """ | """ | ||||
| target_fee = tx_size * fee_per_kB / 1000 | target_fee = tx_size * fee_per_kB / 1000 | ||||
| if fee < (tx_size - wiggleroom) * fee_per_kB / 1000: | if fee < (tx_size - wiggleroom) * fee_per_kB / 1000: | ||||
| raise AssertionError( | raise AssertionError( | ||||
| "Fee of %s BCH too low! (Should be %s BCH)" % (str(fee), str(target_fee))) | "Fee of {} BCH too low! (Should be {} BCH)".format(str(fee), str(target_fee))) | ||||
| if fee > (tx_size + wiggleroom) * fee_per_kB / 1000: | if fee > (tx_size + wiggleroom) * fee_per_kB / 1000: | ||||
| raise AssertionError( | raise AssertionError( | ||||
| "Fee of %s BCH too high! (Should be %s BCH)" % (str(fee), str(target_fee))) | "Fee of {} BCH too high! (Should be {} BCH)".format(str(fee), str(target_fee))) | ||||
| def assert_equal(thing1, thing2, *args): | def assert_equal(thing1, thing2, *args): | ||||
| if thing1 != thing2 or any(thing1 != arg for arg in args): | if thing1 != thing2 or any(thing1 != arg for arg in args): | ||||
| raise AssertionError("not(%s)" % " == ".join(str(arg) | raise AssertionError("not({})".format(" == ".join(str(arg) | ||||
| for arg in (thing1, thing2) + args)) | for arg in (thing1, thing2) + args))) | ||||
| def assert_greater_than(thing1, thing2): | def assert_greater_than(thing1, thing2): | ||||
| if thing1 <= thing2: | if thing1 <= thing2: | ||||
| raise AssertionError("%s <= %s" % (str(thing1), str(thing2))) | raise AssertionError("{} <= {}".format(str(thing1), str(thing2))) | ||||
| def assert_greater_than_or_equal(thing1, thing2): | def assert_greater_than_or_equal(thing1, thing2): | ||||
| if thing1 < thing2: | if thing1 < thing2: | ||||
| raise AssertionError("%s < %s" % (str(thing1), str(thing2))) | raise AssertionError("{} < {}".format(str(thing1), str(thing2))) | ||||
| def assert_raises(exc, fun, *args, **kwds): | def assert_raises(exc, fun, *args, **kwds): | ||||
| assert_raises_message(exc, None, fun, *args, **kwds) | assert_raises_message(exc, None, fun, *args, **kwds) | ||||
| def assert_raises_message(exc, message, fun, *args, **kwds): | def assert_raises_message(exc, message, fun, *args, **kwds): | ||||
| try: | try: | ||||
| Show All 25 Lines | Args: | ||||
| fun (function): the function to call. This should execute a process. | fun (function): the function to call. This should execute a process. | ||||
| args*: positional arguments for the function. | args*: positional arguments for the function. | ||||
| kwds**: named arguments for the function. | kwds**: named arguments for the function. | ||||
| """ | """ | ||||
| try: | try: | ||||
| fun(*args, **kwds) | fun(*args, **kwds) | ||||
| except CalledProcessError as e: | except CalledProcessError as e: | ||||
| if returncode != e.returncode: | if returncode != e.returncode: | ||||
| raise AssertionError("Unexpected returncode %i" % e.returncode) | raise AssertionError( | ||||
| "Unexpected returncode {}".format(e.returncode)) | |||||
| if output not in e.output: | if output not in e.output: | ||||
| raise AssertionError("Expected substring not found:" + e.output) | raise AssertionError("Expected substring not found:" + e.output) | ||||
| else: | else: | ||||
| raise AssertionError("No exception raised") | raise AssertionError("No exception raised") | ||||
| def assert_raises_rpc_error(code, message, fun, *args, **kwds): | def assert_raises_rpc_error(code, message, fun, *args, **kwds): | ||||
| """Run an RPC and verify that a specific JSONRPC exception code and message is raised. | """Run an RPC and verify that a specific JSONRPC exception code and message is raised. | ||||
| Show All 20 Lines | def try_rpc(code, message, fun, *args, **kwds): | ||||
| Test against error code and message if the rpc fails. | Test against error code and message if the rpc fails. | ||||
| Returns whether a JSONRPCException was raised.""" | Returns whether a JSONRPCException was raised.""" | ||||
| try: | try: | ||||
| fun(*args, **kwds) | fun(*args, **kwds) | ||||
| except JSONRPCException as e: | except JSONRPCException as e: | ||||
| # JSONRPCException was thrown as expected. Check the code and message values are correct. | # JSONRPCException was thrown as expected. Check the code and message values are correct. | ||||
| if (code is not None) and (code != e.error["code"]): | if (code is not None) and (code != e.error["code"]): | ||||
| raise AssertionError( | raise AssertionError( | ||||
| "Unexpected JSONRPC error code %i" % e.error["code"]) | "Unexpected JSONRPC error code {}".format(e.error["code"])) | ||||
| if (message is not None) and (message not in e.error['message']): | if (message is not None) and (message not in e.error['message']): | ||||
| raise AssertionError( | raise AssertionError( | ||||
| "Expected substring not found:" + e.error['message']) | "Expected substring not found:" + e.error['message']) | ||||
| return True | return True | ||||
| except Exception as e: | except Exception as e: | ||||
| raise AssertionError( | raise AssertionError( | ||||
| "Unexpected exception raised: " + type(e).__name__) | "Unexpected exception raised: " + type(e).__name__) | ||||
| else: | else: | ||||
| return False | return False | ||||
| def assert_is_hex_string(string): | def assert_is_hex_string(string): | ||||
| try: | try: | ||||
| int(string, 16) | int(string, 16) | ||||
| except Exception as e: | except Exception as e: | ||||
| raise AssertionError( | raise AssertionError( | ||||
| "Couldn't interpret %r as hexadecimal; raised: %s" % (string, e)) | "Couldn't interpret {!r} as hexadecimal; raised: {}".format(string, e)) | ||||
| def assert_is_hash_string(string, length=64): | def assert_is_hash_string(string, length=64): | ||||
| if not isinstance(string, str): | if not isinstance(string, str): | ||||
| raise AssertionError("Expected a string, got type %r" % type(string)) | raise AssertionError( | ||||
| "Expected a string, got type {!r}".format(type(string))) | |||||
| elif length and len(string) != length: | elif length and len(string) != length: | ||||
| raise AssertionError( | raise AssertionError( | ||||
| "String of length %d expected; got %d" % (length, len(string))) | "String of length {} expected; got {}".format(length, len(string))) | ||||
| elif not re.match('[abcdef0-9]+$', string): | elif not re.match('[abcdef0-9]+$', string): | ||||
| raise AssertionError( | raise AssertionError( | ||||
| "String %r contains invalid characters for a hash." % string) | "String {!r} contains invalid characters for a hash.".format(string)) | ||||
| def assert_array_result(object_array, to_match, expected, should_not_find=False): | def assert_array_result(object_array, to_match, expected, should_not_find=False): | ||||
| """ | """ | ||||
| Pass in array of JSON objects, a dictionary with key/value pairs | Pass in array of JSON objects, a dictionary with key/value pairs | ||||
| to match against, and another dictionary with expected key/value | to match against, and another dictionary with expected key/value | ||||
| pairs. | pairs. | ||||
| If the should_not_find flag is true, to_match should not be found | If the should_not_find flag is true, to_match should not be found | ||||
| in object_array | in object_array | ||||
| """ | """ | ||||
| if should_not_find: | if should_not_find: | ||||
| assert_equal(expected, {}) | assert_equal(expected, {}) | ||||
| num_matched = 0 | num_matched = 0 | ||||
| for item in object_array: | for item in object_array: | ||||
| all_match = True | all_match = True | ||||
| for key, value in to_match.items(): | for key, value in to_match.items(): | ||||
| if item[key] != value: | if item[key] != value: | ||||
| all_match = False | all_match = False | ||||
| if not all_match: | if not all_match: | ||||
| continue | continue | ||||
| elif should_not_find: | elif should_not_find: | ||||
| num_matched = num_matched + 1 | num_matched = num_matched + 1 | ||||
| for key, value in expected.items(): | for key, value in expected.items(): | ||||
| if item[key] != value: | if item[key] != value: | ||||
| raise AssertionError("%s : expected %s=%s" % | raise AssertionError("{} : expected {}={}".format( | ||||
| (str(item), str(key), str(value))) | str(item), str(key), str(value))) | ||||
| num_matched = num_matched + 1 | num_matched = num_matched + 1 | ||||
| if num_matched == 0 and not should_not_find: | if num_matched == 0 and not should_not_find: | ||||
| raise AssertionError("No objects matched %s" % (str(to_match))) | raise AssertionError("No objects matched {}".format(str(to_match))) | ||||
| if num_matched > 0 and should_not_find: | if num_matched > 0 and should_not_find: | ||||
| raise AssertionError("Objects were found %s" % (str(to_match))) | raise AssertionError("Objects were found {}".format(str(to_match))) | ||||
| # Utility functions | # Utility functions | ||||
| ################### | ################### | ||||
| def check_json_precision(): | def check_json_precision(): | ||||
| """Make sure json library being used does not lose precision converting BCH values""" | """Make sure json library being used does not lose precision converting BCH values""" | ||||
| n = Decimal("20000000.00000003") | n = Decimal("20000000.00000003") | ||||
| ▲ Show 20 Lines • Show All 103 Lines • ▼ Show 20 Lines | |||||
| def rpc_port(n): | def rpc_port(n): | ||||
| return PORT_MIN + PORT_RANGE + n + (MAX_NODES * PortSeed.n) % (PORT_RANGE - 1 - MAX_NODES) | return PORT_MIN + PORT_RANGE + n + (MAX_NODES * PortSeed.n) % (PORT_RANGE - 1 - MAX_NODES) | ||||
| def rpc_url(datadir, host, port): | def rpc_url(datadir, host, port): | ||||
| rpc_u, rpc_p = get_auth_cookie(datadir) | rpc_u, rpc_p = get_auth_cookie(datadir) | ||||
| if host == None: | if host == None: | ||||
| host = '127.0.0.1' | host = '127.0.0.1' | ||||
| return "http://%s:%s@%s:%d" % (rpc_u, rpc_p, host, int(port)) | return "http://{}:{}@{}:{}".format(rpc_u, rpc_p, host, int(port)) | ||||
| # Node functions | # Node functions | ||||
| ################ | ################ | ||||
| def initialize_datadir(dirname, n): | def initialize_datadir(dirname, n): | ||||
| datadir = os.path.join(dirname, "node" + str(n)) | datadir = os.path.join(dirname, "node" + str(n)) | ||||
| if not os.path.isdir(datadir): | if not os.path.isdir(datadir): | ||||
| ▲ Show 20 Lines • Show All 143 Lines • ▼ Show 20 Lines | def find_output(node, txid, amount): | ||||
| """ | """ | ||||
| Return index to output of txid with value amount | Return index to output of txid with value amount | ||||
| Raises exception if there is none. | Raises exception if there is none. | ||||
| """ | """ | ||||
| txdata = node.getrawtransaction(txid, 1) | txdata = node.getrawtransaction(txid, 1) | ||||
| for i in range(len(txdata["vout"])): | for i in range(len(txdata["vout"])): | ||||
| if txdata["vout"][i]["value"] == amount: | if txdata["vout"][i]["value"] == amount: | ||||
| return i | return i | ||||
| raise RuntimeError("find_output txid %s : %s not found" % | raise RuntimeError("find_output txid {} : {} not found".format( | ||||
| (txid, str(amount))) | txid, str(amount))) | ||||
| def gather_inputs(from_node, amount_needed, confirmations_required=1): | def gather_inputs(from_node, amount_needed, confirmations_required=1): | ||||
| """ | """ | ||||
| Return a random set of unspent txouts that are enough to pay amount_needed | Return a random set of unspent txouts that are enough to pay amount_needed | ||||
| """ | """ | ||||
| assert(confirmations_required >= 0) | assert(confirmations_required >= 0) | ||||
| utxo = from_node.listunspent(confirmations_required) | utxo = from_node.listunspent(confirmations_required) | ||||
| random.shuffle(utxo) | random.shuffle(utxo) | ||||
| inputs = [] | inputs = [] | ||||
| total_in = Decimal("0.00000000") | total_in = Decimal("0.00000000") | ||||
| while total_in < amount_needed and len(utxo) > 0: | while total_in < amount_needed and len(utxo) > 0: | ||||
| t = utxo.pop() | t = utxo.pop() | ||||
| total_in += t["amount"] | total_in += t["amount"] | ||||
| inputs.append( | inputs.append( | ||||
| {"txid": t["txid"], "vout": t["vout"], "address": t["address"]}) | {"txid": t["txid"], "vout": t["vout"], "address": t["address"]}) | ||||
| if total_in < amount_needed: | if total_in < amount_needed: | ||||
| raise RuntimeError("Insufficient funds: need %d, have %d" % | raise RuntimeError("Insufficient funds: need {}, have {}".format( | ||||
| (amount_needed, total_in)) | amount_needed, total_in)) | ||||
| return (total_in, inputs) | return (total_in, inputs) | ||||
| def make_change(from_node, amount_in, amount_out, fee): | def make_change(from_node, amount_in, amount_out, fee): | ||||
| """ | """ | ||||
| Create change output(s), return them | Create change output(s), return them | ||||
| """ | """ | ||||
| outputs = {} | outputs = {} | ||||
| ▲ Show 20 Lines • Show All 129 Lines • Show Last 20 Lines | |||||