Tests/custom tx params and nonce (#22569)

* e2e: added new test for setting custom tx params and replacing tx nonce

Author: churik

test/appium/requirements.txt

@@ -16,4 +16,5 @@ mnemonic==0.21
 bip-utils==2.9.3
 testrail==0.3.15
 webcolors==24.8.0
-CurrencyConverter==0.17.34
+CurrencyConverter==0.17.34
+bs4==0.0.2

test/appium/support/api/lightweight_browser_api.py

@@ -0,0 +1,54 @@
+import requests
+from bs4 import BeautifulSoup
+import re
+
+
+class LightweightBrowserHandler:
+    
+    def __init__(self, base_url: str):
+        self.base_url = base_url
+
+
+    def load_tx_etherscan_page(self, tx_hash: str) -> BeautifulSoup:
+        headers = {
+            "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36",
+            "Accept-Language": "en-US,en;q=0.9",
+            "Cache-Control": "no-cache",
+            "Connection": "keep-alive",
+            "DNT": "1",
+        }
+        tx_url = '%s/tx/%s' % (self.base_url, tx_hash)
+        response = requests.get(tx_url, headers=headers, timeout=10)
+        response.raise_for_status()
+        return BeautifulSoup(response.text, "html.parser")
+
+    def extract_tx_details(self, soup: BeautifulSoup) -> dict:
+        tx_fee_elem = soup.select_one("#txfeebutton")
+        tx_fee = tx_fee_elem.text.strip() if tx_fee_elem else "Not Found"
+
+        confirmation_time_elem = soup.find(string=re.compile('Confirmed within'))
+        confirmation_time = (
+            confirmation_time_elem.text.replace("Confirmed within", "").strip()
+            if confirmation_time_elem
+            else "Not Found"
+        )
+
+        return {'fee': tx_fee, 'confirmation_time': confirmation_time}
+
+    def find_text(self, soup: BeautifulSoup, text: str) -> bool:
+        return soup.find(string=re.compile(re.escape(text))) is not None
+
+    def find_success_badge(self, soup: BeautifulSoup):
+        for span in soup.find_all("span", class_=lambda x: x and "text-green-600" in x):
+            if "Success" in span.get_text():
+                return span
+        return None
+
+    def get_fee_and_confirmation_time(self, tx_url: str) -> dict:
+        try:
+            soup = self.load_tx_etherscan_page(tx_url)
+            return self.extract_tx_details(soup)
+        except requests.RequestException as e:
+            print(f"An error occurred while fetching transaction details: {e}")
+            return {'fee': None, 'confirmation_time': None}
+    

test/appium/support/api/network_api.py

@@ -4,6 +4,7 @@
 from json import JSONDecodeError
 from os import environ
 from typing import List
+import json 
 
 import pytest
 import pytz
@@ -83,3 +84,20 @@ def wait_for_balance_to_be(self, address: str, expected_balance: int):
             time.sleep(20)
         raise TimeoutException(
             'balance is not updated on Etherscan, it is %s but expected to be %s' % (balance, expected_balance))
+
+
+class SepoliaNetworkApi:
+    def __init__(self):
+        self.rpc_url = 'https://ethereum-sepolia.rpc.subquery.network/public'
+
+    def is_tx_successful(self, tx_hash):
+        response = requests.post(headers={"Content-Type": "application/json"},
+                                 url=self.rpc_url,
+                                 data=json.dumps(
+                                     {"jsonrpc": "2.0", "method": "eth_getTransactionReceipt", "params": [tx_hash],
+                                      "id": 1})).json()
+        result = response.get('result')
+        if result is not None:
+            return result.get('status') == '0x1'
+        return None
+            

test/appium/tests/base_test_case.py

@@ -26,6 +26,18 @@
 
 implicit_wait = 5
 
+def get_app_package():
+    """
+    Determines the app package dynamically based on the APK configuration.
+    :return: The app package name.
+    """
+    apk = pytest_config_global['apk']
+    app_folder = "im.status.ethereum"  # Default app package name
+
+    if re.findall(r'pr\d\d\d\d\d', apk) or re.findall(r'\d\d\d\d\d.apk', apk):
+        app_folder += ".pr"  # Append `.pr` in specific cases
+
+    return app_folder
 
 def get_lambda_test_capabilities_real_device():
     capabilities = {
@@ -78,12 +90,7 @@ def get_lambda_test_capabilities_emulator(platform_version: int = 14, device_nam
 
 
 def get_app_path():
-    app_folder = 'im.status.ethereum'
-    apk = pytest_config_global['apk']
-    if re.findall(r'pr\d\d\d\d\d', apk) or re.findall(r'\d\d\d\d\d.apk', apk):
-        app_folder += '.pr'
-    app_path = '/storage/emulated/0/Android/data/%s/files/Download/' % app_folder
-    return app_path
+    return '/storage/emulated/0/Android/data/%s/files/Download/' % get_app_package()
 
 
 def pull_logs_folder(driver):

test/appium/tests/wallet_txs/test_wallet_mainnet_txs.py

@@ -65,7 +65,7 @@ def test_wallet_swap_dai_snt_real_tx(self):
                                                                    network=network_name)
         self.errors.append(self.wallet, tx_errors)
 
-        self.wallet.just_fyi("Check that balance is updated for receiver")
+        self.wallet.just_fyi("Check that From and To balance is updated")
         self.wallet.wait_for_wallet_balance_to_update(fiat_expected_amount_from, asset_name_from, fiat=True)
         self.wallet.wait_for_wallet_balance_to_update(fiat_expected_amount_to, asset_name_to, fiat=True)
 

test/appium/tests/wallet_txs/test_wallet_testnet.py

@@ -5,6 +5,7 @@
 from tests import marks, run_in_parallel
 from users import transaction_senders
 from views.sign_in_view import SignInView
+from support.api.network_api import SepoliaNetworkApi
 
 
 @pytest.mark.xdist_group(name="four_2")
@@ -99,4 +100,92 @@ def test_wallet_send_erc20_from_drawer(self, network, asset, asset_ticker, decim
 
         self.errors.verify_no_errors()
 
+@pytest.mark.xdist_group(name="two_1")
+@marks.nightly
+@marks.smoke
+class TestWalletCustomParamOneDevice(MultipleSharedDeviceTestCase):
+
+    def prepare_devices(self):
+        self.drivers, self.loop  = create_shared_drivers(1)
+        self.driver = self.drivers[0]
+        self.sign_in = SignInView(self.drivers[0])
+        self.sender, self.receiver = transaction_senders['ETH_ADI_STT_2'], transaction_senders['ETH_3']
+        self.sender['wallet_address'] = '0x' + self.sender['address']
+        self.receiver['wallet_address'] = '0x' + self.receiver['address']
+        self.home = self.sign_in.recover_access(self.sender['passphrase'])
+        self.wallet = self.sign_in.get_wallet_view()
+    
+    @marks.testrail_id(742910) 
+    def test_send_snt_custom_tx_params(self):
+        wallet = self.wallet
+        wallet.navigate_back_to_wallet_view()
+        network = 'Sepolia'
+        address, asset_name, amount, asset_ticker = self.receiver['wallet_address'], 'Status Test Token', '1', 'STT'
+        max_base_fee, max_prio, max_gas_amount = '10', '1', '60000'
+
+        wallet.just_fyi("Set amount and address")
+        wallet.set_amount_and_address(address, asset_name, amount)
+        wallet.button_one.click_until_presence_of_element(wallet.advanced_tx_button)
+
+        wallet.just_fyi("Set custom tx params to make pending tx")
+        wallet.advanced_tx_button.wait_and_click()
+        wallet.custom_tx_params_button.wait_and_click()
+        wallet.custom_max_base_fee_button.click()
+        wallet.clear_value_and_set_with_custom_keyboard(max_base_fee)
+        wallet.custom_max_prio_fee_button.wait_and_click()
+        wallet.clear_value_and_set_with_custom_keyboard(max_prio)
+        wallet.custom_max_gas_amount_button.wait_and_click()
+        wallet.clear_value_and_set_with_custom_keyboard(max_gas_amount)
+        wallet.button_one.click()
+
+        wallet.just_fyi("Check that custom tx params are preserved")
+        wallet.advanced_tx_button.wait_and_click()
+        wallet.custom_tx_params_button.wait_and_click()
+        for value in (max_base_fee, max_prio, max_gas_amount):
+            if not wallet.get_custom_tx_element(value).is_element_displayed():
+                self.errors.append(wallet, "Value %s is not preserved in custom tx settings drawer!" % value)
+        wallet.button_one.click()
+
+        wallet.just_fyi("Sign first tx (for dropping) and copy tx hash")
+        wallet.slide_and_confirm_with_password()
+
+        wallet.just_fyi("Verify send tx in the list for sender")
+        device_time_before_sending = wallet.driver.device_time
+        tx_errors = wallet.check_last_transaction_in_activity(device_time_before_sending, amount,
+                                                                   send_to_account=self.receiver['wallet_address'],
+                                                                   asset=asset_ticker,
+                                                                   tx_type='Send',
+                                                                   network=network, 
+                                                                   navigate_to_main_screen=False)
+        self.errors.append(wallet, tx_errors)
+        dropped_tx_hash = wallet.copy_tx_hash()
+
+        wallet.just_fyi("Send second tx (for replacing) with custom nonce")
+        wallet.navigate_back_to_wallet_view()
+        wallet.set_amount_and_address(address, asset_name, amount)
+        wallet.button_one.click_until_presence_of_element(wallet.advanced_tx_button)
+        wallet.advanced_tx_button.wait_and_click()
+        wallet.custom_tx_params_button.wait_and_click()
+        wallet.custom_nonce_button.wait_and_click()
+        current_nonce = int(wallet.amount_input.text)
+        wallet.clear_nonce_field()
+        wallet.set_amount(str(current_nonce - 1))
+        wallet.button_one.click()
+        wallet.button_one.click()
+        wallet.slide_and_confirm_with_password()
+        replacemant_tx_hash = wallet.copy_tx_hash()
+        
+        if replacemant_tx_hash == dropped_tx_hash:
+            wallet.just_fyi("Scenario1, nonce is too low: initial tx failed, check the status")
+            if SepoliaNetworkApi().is_tx_successful(dropped_tx_hash) is not False:
+                self.errors.append(wallet,
+                                   "Initial tx %s is not failed, so custom nonce is not set" % dropped_tx_hash)
+
+        else:
+            wallet.just_fyi("Scenario2, Nonce is replaced: check that initial tx was dropped as custom nonce has been set")
+            if SepoliaNetworkApi().is_tx_successful(dropped_tx_hash) is None:
+                self.errors.append(wallet, "Dropped tx %s receipt is available, so custom nonce is not set" % dropped_tx_hash)
+
+        self.errors.verify_no_errors()
+
 

test/appium/views/wallet_view.py

@@ -2,6 +2,7 @@
 import time
 from typing import Literal
 
+
 import pytest
 from selenium.common import NoSuchElementException
 
@@ -94,6 +95,10 @@ def header(self):
     def timestamp(self):
         return Text(self.driver, prefix=self.locator, xpath="//*[@content-desc='transaction-timestamp']").text
 
+    @property
+    def options(self):
+        return Button(self.driver, prefix=self.locator, xpath="//*[@content-desc='transaction-timestamp']/following::*[@content-desc='icon'][1]")
+
     @property
     def amount(self):
         return Text(self.driver, prefix=self.locator,
@@ -190,6 +195,14 @@ def __init__(self, driver):
         self.to_data_container = ConfirmationViewInfoContainer(self.driver, label_name='To')
         self.on_data_container = ConfirmationViewInfoContainer(self.driver, label_name='On')
 
+        # Advanced tx settings
+        self.advanced_tx_button = Button(self.driver, accessibility_id='advanced-button')
+        self.custom_tx_params_button = Button(self.driver, translation_id='custom')
+        self.custom_max_base_fee_button = Button(self.driver, translation_id='max-base-fee')
+        self.custom_max_prio_fee_button = Button(self.driver, translation_id='priority-fee')
+        self.custom_nonce_button = Button(self.driver, translation_id='nonce')
+        self.custom_max_gas_amount_button = Button(self.driver, translation_id='max-gas-amount')
+
         # Swap flow
         self.approve_swap_button = Button(self.driver, accessibility_id='Approve')
         self.spending_cap_approval_info_container = BaseElement(
@@ -234,6 +247,10 @@ def __init__(self, driver):
             self.driver, xpath="//*[@content-desc='expanded-collectible']//android.widget.ImageView")
         self.send_from_collectible_info_button = Button(self.driver, accessibility_id="icon, Send")
 
+
+        # Tx activity
+        self.copy_tx_hash_button = Button(self.driver, accessibility_id="copy-transaction-hash")
+
         # dApp adding
         self.add_dapp_button = Button(self.driver, accessibility_id='connected-dapps-add')
         self.wallet_connect_button = Button(self.driver, accessibility_id='wc-connect')
@@ -309,13 +326,34 @@ def set_amount(self, amount: str):
         for i in amount:
             Button(self.driver, accessibility_id='keyboard-key-%s' % i).click()
 
-    def send_asset(self, address: str, asset_name: str, amount, network_name: str, account='Account 1'):
-        self.send_button.click()
+    def clear_value_and_set_with_custom_keyboard(self, value: str):
+        self.clear_amount()
+        self.set_amount(value)
+        self.button_one.click()
+        
+    def clear_nonce_field(self):
+        Button(self.driver, xpath='//com.horcrux.svg.SvgView[@content-desc="icon"]').click()
+
+    def clear_amount(self):
+        """
+        Tap on the clear button until the amount input field is empty.
+        """
+        clear_button = Button(self.driver, accessibility_id="icon-label")
+        edit_box = EditBox(self.driver, xpath="//android.widget.EditText")
+        while edit_box.text != "" and edit_box.text != "0":
+            clear_button.click()
+    
+    def set_amount_and_address(self, address: str, asset_name: str, amount, network_name=""):
+        self.send_button.click_until_presence_of_element(self.address_text_input)
         self.address_text_input.send_keys(address)
         self.continue_button.click()
         self.select_asset(asset_name)
-        self.select_network(network_name)
+        if network_name:
+            self.select_network(network_name)
         self.set_amount(str(amount))
+
+    def send_asset(self, address: str, asset_name: str, amount, network_name: str, account='Account 1'):
+        self.set_amount_and_address(address, asset_name, amount, network_name)
         self.confirm_transaction()
 
     def send_asset_from_drawer(self, address: str, asset_name: str, amount, network_name: str):
@@ -345,6 +383,11 @@ def swap_asset_from_drawer(self, asset_name: str, amount, network_name: str, dec
                 'est_slippage': slippage,
                 'max_fees': fees}
 
+    def copy_tx_hash(self):
+        self.get_activity_element().options.click_until_presence_of_element(self.copy_tx_hash_button)
+        self.copy_tx_hash_button.click()
+        return self.driver.get_clipboard_text()
+
     def add_regular_account(self, account_name: str):
         if not self.add_account_button.is_element_displayed():
             self.get_account_element().swipe_left_on_element()
@@ -479,7 +522,8 @@ def check_last_transaction_in_activity(self, device_time, amount,
                                            send_to_account='',
                                            swap_asset_to='SNT',
                                            swap_amount_to='0.000000000000000000',
-                                           network='Status Network'):
+                                           network='Status Network',
+                                           navigate_to_main_screen=True):
         errors = list()
         current_time = datetime.datetime.strptime(device_time, "%Y-%m-%dT%H:%M:%S%z")
         expected_time = "Today %s" % current_time.strftime('%-I:%M %p')
@@ -528,7 +572,8 @@ def check_last_transaction_in_activity(self, device_time, amount,
         except NoSuchElementException:
             errors.append("Can't find the last transaction")
         finally:
-            self.close_account_button.click_until_presence_of_element(self.show_qr_code_button)
+            if navigate_to_main_screen:
+                self.close_account_button.click_until_presence_of_element(self.show_qr_code_button)
         return errors
 
     def get_balance(self, asset='Ether', fiat=False):
@@ -539,6 +584,10 @@ def get_balance(self, asset='Ether', fiat=False):
     def get_receive_swap_amount(self, decimals=18):
         self.just_fyi("Getting swap Receive amount for on review page")
         return self.round_amount_float(self.swap_receive_amount_summary_text.text.split()[0], decimals)
+    
+    def get_custom_tx_element(self, text):
+        return Text(self.driver, xpath="//*[@content-desc[contains(., '%s')]]" % text)
+
 
     def get_connected_dapp_element_by_name(self, dapp_name: str):
         class ConnectedDAppElement(BaseElement):
@@ -557,3 +606,5 @@ def disconnect(self):
     def select_account_to_connect_dapp(self, account_name: str):
         self.select_account_to_connect_dapp_button.click()
         Button(self.driver, xpath="//*[@content-desc='container']/*[@text='%s']" % account_name).click()
+
+