import json import logging import jsonschema import resources.constants as constants from conftest import option def verify_json_schema(response, method): with open(f"{option.base_dir}/schemas/{method}", "r") as schema: jsonschema.validate(instance=response, schema=json.load(schema)) def get_suggested_routes(rpc_client, **kwargs): method = "wallet_getSuggestedRoutesAsync" required_params = ["uuid", "sendType", "addrFrom", "addrTo", "amountIn", "tokenID", "gasFeeMode"] input_params = {} for key, new_value in kwargs.items(): input_params[key] = new_value for key in required_params: if key not in input_params: logging.info(f"Warning: The key '{key}' does not exist in the input_params parameters and will be ignored.") params = [input_params] rpc_client.prepare_wait_for_signal("wallet.suggested.routes", 1) _ = rpc_client.rpc_valid_request(method, params) routes = rpc_client.wait_for_signal("wallet.suggested.routes") return routes["event"] def build_transactions_from_route(rpc_client, **kwargs): method = "wallet_buildTransactionsFromRoute" required_params = [] build_tx_params = {} for key, new_value in kwargs.items(): build_tx_params[key] = new_value for key in required_params: if key not in build_tx_params: logging.info(f"Warning: The key '{key}' does not exist in the build_tx_params parameters and will be ignored.") params = [build_tx_params] _ = rpc_client.rpc_valid_request(method, params) wallet_router_sign_transactions = rpc_client.wait_for_signal("wallet.router.sign-transactions") assert wallet_router_sign_transactions["event"]["signingDetails"]["signOnKeycard"] is False transaction_hashes = wallet_router_sign_transactions["event"]["signingDetails"]["hashes"] assert transaction_hashes, "Transaction hashes are empty!" return wallet_router_sign_transactions["event"] def sign_messages(rpc_client, hashes, address): tx_signatures = {} for hash in hashes: method = "wallet_signMessage" params = [hash, address, option.password] response = rpc_client.rpc_valid_request(method, params) result = response.json().get("result") assert result and result.startswith("0x"), f"Invalid transaction signature for hash {hash}: {result}" tx_signature = result[2:] signature = { "r": tx_signature[:64], "s": tx_signature[64:128], "v": tx_signature[128:], } tx_signatures[hash] = signature return tx_signatures def check_tx_details(rpc_client, tx_hash, network_id, address_to, expected_amount_in): method = "ethclient_transactionByHash" params = [network_id, tx_hash] response = rpc_client.rpc_valid_request(method, params) tx_details = response.json()["result"]["tx"] assert tx_details["value"] == expected_amount_in assert tx_details["to"].upper() == address_to.upper() def check_fees(fee_mode, base_fee, max_priority_fee_per_gas, max_fee_per_gas): assert base_fee.startswith("0x") assert max_priority_fee_per_gas.startswith("0x") assert max_fee_per_gas.startswith("0x") base_fee_int = int(base_fee, 16) max_priority_fee_per_gas_int = int(max_priority_fee_per_gas, 16) max_fee_per_gas_int = int(max_fee_per_gas, 16) if fee_mode == constants.gas_fee_mode_low: assert base_fee_int + max_priority_fee_per_gas_int == max_fee_per_gas_int elif fee_mode == constants.gas_fee_mode_medium: assert (2 * base_fee_int + max_priority_fee_per_gas_int) == max_fee_per_gas_int elif fee_mode == constants.gas_fee_mode_high: assert 3 * base_fee_int + max_priority_fee_per_gas_int == max_fee_per_gas_int elif fee_mode == constants.gas_fee_mode_custom: assert base_fee_int + max_priority_fee_per_gas_int == max_fee_per_gas_int else: assert False, "Invalid gas fee mode" def check_fees_for_path(path_name, gas_fee_mode, check_approval, route): for path_tx in route: if path_tx["ProcessorName"] != path_name: continue if check_approval: assert path_tx["ApprovalRequired"] check_fees(gas_fee_mode, path_tx["ApprovalBaseFee"], path_tx["ApprovalPriorityFee"], path_tx["ApprovalMaxFeesPerGas"]) return check_fees(gas_fee_mode, path_tx["TxBaseFee"], path_tx["TxPriorityFee"], path_tx["TxMaxFeesPerGas"]) def send_router_transactions_with_signatures(rpc_client, uuid, tx_signatures): method = "wallet_sendRouterTransactionsWithSignatures" params = [{"uuid": uuid, "Signatures": tx_signatures}] _ = rpc_client.rpc_valid_request(method, params) tx_status = rpc_client.wait_for_signal("wallet.transaction.status-changed") assert tx_status["event"]["status"] == "Success" return tx_status["event"] def send_router_transaction(rpc_client, **kwargs): routes = get_suggested_routes(rpc_client, **kwargs) router_build_tx_params = {} for key in kwargs: if key in ["uuid", "slippagePercentage"]: router_build_tx_params[key] = kwargs[key] build_tx = build_transactions_from_route(rpc_client, **router_build_tx_params) tx_signatures = sign_messages(rpc_client, build_tx["signingDetails"]["hashes"], kwargs.get("addrFrom")) tx_status = send_router_transactions_with_signatures(rpc_client, routes["Uuid"], tx_signatures) return { "routes": routes, "build_tx": build_tx, "tx_signatures": tx_signatures, "tx_status": tx_status, }