Skip to content

Commit

Permalink
test(wallet)_: added tests for selecting custom fees and setting cust…
Browse files Browse the repository at this point in the history 
…om tx props
  • Loading branch information
@saledjenic
saledjenic committed Nov 28, 2024
1 parent f04d88b commit 011767c
Show file tree
Hide file tree
Showing 5 changed files with 310 additions and 144 deletions.
1 change: 1 addition & 0 deletions tests-functional/clients/rpc.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -57,6 +57,7 @@ def rpc_request(self, method, params=[], request_id=13, url=None):

def rpc_valid_request(self, method, params=[], _id=None, url=None):
response = self.rpc_request(method, params, _id, url)
json_obj = response.json()
self.verify_is_valid_json_rpc_response(response, _id)
return response

Expand Down
7 changes: 7 additions & 0 deletions tests-functional/constants.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -21,3 +21,10 @@ class Account:
password="Strong12345",
passphrase="test test test test test test test test test test nest junk"
)

gas_fee_mode_low = 0
gas_fee_mode_medium = 1
gas_fee_mode_high = 2
gas_fee_mode_custom = 3

processor_name_transfer = "Transfer"
288 changes: 195 additions & 93 deletions tests-functional/tests/test_router.py
View file
Original file line number Diff line number Diff line change
@@ -1,16 +1,18 @@
import uuid
import uuid as uuid_lib

import pytest
import logging
import constants

from conftest import option
from constants import user_1, user_2
from test_cases import SignalTestCase

import wallet_utils

@pytest.mark.rpc
@pytest.mark.transaction
@pytest.mark.wallet
class TestTransactionFromRoute(SignalTestCase):
class TestRouter(SignalTestCase):
await_signals = [
"wallet.suggested.routes",
"wallet.router.sign-transactions",
Expand All @@ -20,96 +22,196 @@ class TestTransactionFromRoute(SignalTestCase):
]

def test_tx_from_route(self):
uuid = str(uuid_lib.uuid4())
amount_in = "0xde0b6b3a7640000"

params = {
"uuid": uuid,
"sendType": 0,
"addrFrom": constants.user_1.address,
"addrTo": constants.user_2.address,
"amountIn": amount_in,
"amountOut": "0x0",
"tokenID": "ETH",
"tokenIDIsOwnerToken": False,
"toTokenID": "",
"disabledFromChainIDs": [1, 10, 42161],
"disabledToChainIDs": [1, 10, 42161],
"gasFeeMode": 1,
"fromLockedAmount": {}
}

routes = wallet_utils.get_suggested_routes(self.rpc_client, self.signal_client, **params)
wallet_router_sign_transactions = wallet_utils.build_transactions_from_route(self.rpc_client, self.signal_client, **params)
transaction_hashes = wallet_router_sign_transactions["signingDetails"]["hashes"]
tx_signatures = wallet_utils.sign_messages(self.rpc_client, transaction_hashes, constants.user_1.address)
tx_status = wallet_utils.send_router_transactions_with_signatures(self.rpc_client, self.signal_client, uuid, tx_signatures)
wallet_utils.check_tx_details(self.rpc_client, tx_status["hash"], constants.user_1.address, constants.user_2.address, amount_in)


def test_setting_different_fee_modes(self):
uuid = str(uuid_lib.uuid4())
gas_fee_mode = constants.gas_fee_mode_medium
amount_in = "0xde0b6b3a7640000"

_uuid = str(uuid.uuid4())
router_input_params = {
"uuid": uuid,
"sendType": 0,
"addrFrom": constants.user_1.address,
"addrTo": constants.user_2.address,
"amountIn": amount_in,
"amountOut": "0x0",
"tokenID": "ETH",
"tokenIDIsOwnerToken": False,
"toTokenID": "",
"disabledFromChainIDs": [1, 10, 42161],
"disabledToChainIDs": [1, 10, 42161],
"gasFeeMode": gas_fee_mode,
"fromLockedAmount": {}
}

logging.info("Step: getting the best route")
routes = wallet_utils.get_suggested_routes(self.rpc_client, self.signal_client, **router_input_params)
assert len(routes["Best"]) > 0
wallet_utils.check_fees_for_path(constants.processor_name_transfer, gas_fee_mode, routes["Best"][0]["ApprovalRequired"], routes["Best"])

logging.info("Step: update gas fee mode without providing path tx identity params via wallet_setFeeMode endpoint")
method = "wallet_setFeeMode"
response = self.rpc_client.rpc_request(method, [None, gas_fee_mode])
self.rpc_client.verify_is_json_rpc_error(response)

logging.info("Step: update gas fee mode with incomplete details for path tx identity params via wallet_setFeeMode endpoint")
tx_identity_params = {
"routerInputParamsUuid": uuid,
}
response = self.rpc_client.rpc_request(method, [tx_identity_params, gas_fee_mode])
self.rpc_client.verify_is_json_rpc_error(response)

logging.info("Step: update gas fee mode to low")
gas_fee_mode = constants.gas_fee_mode_low
tx_identity_params = {
"routerInputParamsUuid": uuid,
"pathName": routes["Best"][0]["ProcessorName"],
"chainID": routes["Best"][0]["FromChain"]["chainId"],
"isApprovalTx": routes["Best"][0]["ApprovalRequired"],
}
self.signal_client.prepare_wait_for_signal("wallet.suggested.routes", 1)
_ = self.rpc_client.rpc_valid_request(method, [tx_identity_params, gas_fee_mode])
response = self.signal_client.wait_for_signal("wallet.suggested.routes")
routes = response['event']
assert len(routes["Best"]) > 0
wallet_utils.check_fees_for_path(constants.processor_name_transfer, gas_fee_mode, routes["Best"][0]["ApprovalRequired"], routes["Best"])

logging.info("Step: update gas fee mode to high")
gas_fee_mode = constants.gas_fee_mode_high
self.signal_client.prepare_wait_for_signal("wallet.suggested.routes", 1)
_ = self.rpc_client.rpc_valid_request(method, [tx_identity_params, gas_fee_mode])
response = self.signal_client.wait_for_signal("wallet.suggested.routes")
routes = response['event']
assert len(routes["Best"]) > 0
wallet_utils.check_fees_for_path(constants.processor_name_transfer, gas_fee_mode, routes["Best"][0]["ApprovalRequired"], routes["Best"])

logging.info("Step: try to set custom gas fee mode via wallet_setFeeMode endpoint")
gas_fee_mode = constants.gas_fee_mode_custom
response = self.rpc_client.rpc_request(method, [tx_identity_params, gas_fee_mode])
self.rpc_client.verify_is_json_rpc_error(response)


def test_setting_custom_fee_mode(self):
uuid = str(uuid_lib.uuid4())
gas_fee_mode = constants.gas_fee_mode_medium
amount_in = "0xde0b6b3a7640000"

method = "wallet_getSuggestedRoutesAsync"
params = [
{
"uuid": _uuid,
"sendType": 0,
"addrFrom": user_1.address,
"addrTo": user_2.address,
"amountIn": amount_in,
"amountOut": "0x0",
"tokenID": "ETH",
"tokenIDIsOwnerToken": False,
"toTokenID": "",
"disabledFromChainIDs": [10, 42161],
"disabledToChainIDs": [10, 42161],
"gasFeeMode": 1,
"fromLockedAmount": {}
}
]
response = self.rpc_client.rpc_valid_request(method, params)

routes = self.signal_client.wait_for_signal("wallet.suggested.routes")
assert routes['event']['Uuid'] == _uuid

method = "wallet_buildTransactionsFromRoute"
params = [
{
"uuid": _uuid,
"slippagePercentage": 0
}
]
response = self.rpc_client.rpc_valid_request(method, params)

wallet_router_sign_transactions = self.signal_client.wait_for_signal(
"wallet.router.sign-transactions")

assert wallet_router_sign_transactions['event']['signingDetails']['signOnKeycard'] == False
transaction_hashes = wallet_router_sign_transactions['event']['signingDetails']['hashes']

assert transaction_hashes, "Transaction hashes are empty!"

tx_signatures = {}

for hash in transaction_hashes:

method = "wallet_signMessage"
params = [
hash,
user_1.address,
option.password
]

response = self.rpc_client.rpc_valid_request(method, params)

if response.json()["result"].startswith("0x"):
tx_signature = response.json()["result"][2:]

signature = {
"r": tx_signature[:64],
"s": tx_signature[64:128],
"v": tx_signature[128:]
}

tx_signatures[hash] = signature

method = "wallet_sendRouterTransactionsWithSignatures"
params = [
{
"uuid": _uuid,
"Signatures": tx_signatures
}
]
response = self.rpc_client.rpc_valid_request(method, params)

tx_status = self.signal_client.wait_for_signal(
"wallet.transaction.status-changed")

assert tx_status["event"]["chainId"] == 31337
assert tx_status["event"]["status"] == "Success"
tx_hash = tx_status["event"]["hash"]

method = "eth_getTransactionByHash"
params = [tx_hash]

response = self.rpc_client.rpc_valid_request(method, params, url=option.anvil_url)
tx_details = response.json()["result"]

assert tx_details["value"] == amount_in
assert tx_details["to"].upper() == user_2.address.upper()
assert tx_details["from"].upper() == user_1.address.upper()
router_input_params = {
"uuid": uuid,
"sendType": 0,
"addrFrom": constants.user_1.address,
"addrTo": constants.user_2.address,
"amountIn": amount_in,
"amountOut": "0x0",
"tokenID": "ETH",
"tokenIDIsOwnerToken": False,
"toTokenID": "",
"disabledFromChainIDs": [1, 10, 42161],
"disabledToChainIDs": [1, 10, 42161],
"gasFeeMode": gas_fee_mode,
"fromLockedAmount": {}
}

logging.info("Step: getting the best route")
routes = wallet_utils.get_suggested_routes(self.rpc_client, self.signal_client, **router_input_params)

assert len(routes["Best"]) > 0

wallet_utils.check_fees_for_path(constants.processor_name_transfer, gas_fee_mode, routes["Best"][0]["ApprovalRequired"], routes["Best"])

logging.info("Step: try to set custom tx details with empty params via wallet_setCustomTxDetails endpoint")
method = "wallet_setCustomTxDetails"
response = self.rpc_client.rpc_request(method, [None, None])
self.rpc_client.verify_is_json_rpc_error(response)

logging.info("Step: try to set custom tx details with incomplete details for path tx identity params via wallet_setCustomTxDetails endpoint")
tx_identity_params = {
"routerInputParamsUuid": uuid,
}
response = self.rpc_client.rpc_request(method, [tx_identity_params, None])
self.rpc_client.verify_is_json_rpc_error(response)

logging.info("Step: try to set custom tx details providing other than the custom gas fee mode via wallet_setCustomTxDetails endpoint")
tx_identity_params = {
"routerInputParamsUuid": uuid,
"pathName": routes["Best"][0]["ProcessorName"],
"chainID": routes["Best"][0]["FromChain"]["chainId"],
"isApprovalTx": routes["Best"][0]["ApprovalRequired"],
}
tx_custom_params = {
"gasFeeMode": constants.gas_fee_mode_low,
}
response = self.rpc_client.rpc_request(method, [tx_identity_params, tx_custom_params])
self.rpc_client.verify_is_json_rpc_error(response)

logging.info("Step: try to set custom tx details without providing maxFeesPerGas via wallet_setCustomTxDetails endpoint")
tx_custom_params = {
"gasFeeMode": gas_fee_mode,
}
response = self.rpc_client.rpc_request(method, [tx_identity_params, tx_custom_params])
self.rpc_client.verify_is_json_rpc_error(response)

logging.info("Step: try to set custom tx details without providing PriorityFee via wallet_setCustomTxDetails endpoint")
tx_custom_params = {
"gasFeeMode": gas_fee_mode,
"maxFeesPerGas": "0x77359400",
}
response = self.rpc_client.rpc_request(method, [tx_identity_params, tx_custom_params])
self.rpc_client.verify_is_json_rpc_error(response)

logging.info("Step: try to set custom tx details via wallet_setCustomTxDetails endpoint")
gas_fee_mode = constants.gas_fee_mode_custom
tx_nonce = 4
tx_gas_amount = 30000
tx_max_fees_per_gas = "0x77359400"
tx_priority_fee = "0x1DCD6500"
tx_identity_params = {
"routerInputParamsUuid": uuid,
"pathName": routes["Best"][0]["ProcessorName"],
"chainID": routes["Best"][0]["FromChain"]["chainId"],
"isApprovalTx": routes["Best"][0]["ApprovalRequired"],
}
tx_custom_params = {
"gasFeeMode": gas_fee_mode,
"nonce": tx_nonce,
"gasAmount": tx_gas_amount,
"maxFeesPerGas": tx_max_fees_per_gas,
"priorityFee": tx_priority_fee,
}
self.signal_client.prepare_wait_for_signal("wallet.suggested.routes", 1)
_ = self.rpc_client.rpc_valid_request(method, [tx_identity_params, tx_custom_params])
response = self.signal_client.wait_for_signal("wallet.suggested.routes")
routes = response['event']
assert len(routes["Best"]) > 0
tx_nonce_int = int(routes["Best"][0]["TxNonce"], 16)
assert tx_nonce_int == tx_nonce
assert routes["Best"][0]["TxGasAmount"] == tx_gas_amount
assert routes["Best"][0]["TxMaxFeesPerGas"].upper() == tx_max_fees_per_gas.upper()
assert routes["Best"][0]["TxPriorityFee"].upper() == tx_priority_fee.upper()
wallet_utils.check_fees_for_path(constants.processor_name_transfer, gas_fee_mode, routes["Best"][0]["ApprovalRequired"], routes["Best"])
37 changes: 31 additions & 6 deletions tests-functional/tests/test_wallet_activity_session.py
View file
Original file line number Diff line number Diff line change
@@ -1,9 +1,9 @@
import json
import random

import uuid as uuid_lib
import pytest

from constants import user_1
from constants import user_1, user_2
from test_cases import SignalTestCase

import wallet_utils
Expand Down Expand Up @@ -33,9 +33,32 @@ def setup_method(self):
self.request_id = str(random.randint(1, 8888))

def test_wallet_start_activity_filter_session(self):

uuid = str(uuid_lib.uuid4())
amount_in = "0xde0b6b3a7640000"

input_params = {
"uuid": uuid,
"sendType": 0,
"addrFrom": user_1.address,
"addrTo": user_2.address,
"amountIn": amount_in,
"amountOut": "0x0",
"tokenID": "ETH",
"tokenIDIsOwnerToken": False,
"toTokenID": "",
"disabledFromChainIDs": [10, 42161],
"disabledToChainIDs": [10, 42161],
"gasFeeMode": 1,
"fromLockedAmount": {},

#params for building tx from route
"slippagePercentage": 0
}

tx_data = [] # (routes, build_tx, tx_signatures, tx_status)
# Set up a transactions for account before starting session
tx_data.append(wallet_utils.send_router_transaction(self.rpc_client, self.signal_client))
tx_data.append(wallet_utils.send_router_transaction(self.rpc_client, self.signal_client, **input_params))

# Start activity session
method = "wallet_startActivityFilterSessionV2"
Expand All @@ -47,7 +70,7 @@ def test_wallet_start_activity_filter_session(self):
response = self.rpc_client.rpc_valid_request(method, params, self.request_id)
event_response = self.signal_client.wait_for_signal("wallet", timeout=10)['event']

# Check response
# Check response
sessionID = int(response.json()["result"])
assert sessionID > 0

Expand All @@ -60,9 +83,11 @@ def test_wallet_start_activity_filter_session(self):
validate_entry(message['activities'][0], tx_data[-1])

# Trigger new transaction
uuid = str(uuid_lib.uuid4())
input_params["uuid"] = uuid

self.signal_client.prepare_wait_for_signal("wallet", 1, lambda signal : signal["event"]["type"] == EventActivitySessionUpdated and signal['event']['requestId'] == sessionID)
tx_data.append(wallet_utils.send_router_transaction(self.rpc_client, self.signal_client))
print(tx_data[-1])
tx_data.append(wallet_utils.send_router_transaction(self.rpc_client, self.signal_client, **input_params))
event_response = self.signal_client.wait_for_signal("wallet", timeout=10)['event']

# Check response event
Expand Down
Loading

0 comments on commit 011767c

Please sign in to comment.