status-react/test/appium/tests/base_test_case.py

482 lines
18 KiB
Python
Raw Normal View History

import asyncio
2022-11-03 11:26:53 +00:00
import base64
import logging
import re
import subprocess
import sys
from abc import ABCMeta, abstractmethod
2022-01-24 11:40:02 +00:00
from http.client import RemoteDisconnected
import pytest
2022-01-24 11:40:02 +00:00
import requests
from appium import webdriver
2023-09-06 03:07:12 +00:00
from appium.options.common import AppiumOptions
from appium.webdriver.common.mobileby import MobileBy
2022-01-24 11:40:02 +00:00
from sauceclient import SauceException
from selenium.common.exceptions import NoSuchElementException, TimeoutException, WebDriverException
2022-01-24 11:40:02 +00:00
from selenium.webdriver.support.wait import WebDriverWait
2023-06-08 02:06:43 +00:00
from urllib3.exceptions import MaxRetryError, ProtocolError
2022-01-24 11:40:02 +00:00
from support.api.network_api import NetworkApi
from tests import test_suite_data, start_threads, appium_container, pytest_config_global, transl
from tests.conftest import sauce_username, sauce_access_key, apibase, github_report, run_name
2022-11-03 11:26:53 +00:00
executor_sauce_lab = 'https://%s:%s@ondemand.%s:443/wd/hub' % (sauce_username, sauce_access_key, apibase)
executor_local = 'http://localhost:4723/wd/hub'
implicit_wait = 5
2023-02-17 00:32:28 +00:00
def get_capabilities_local():
desired_caps = dict()
if pytest_config_global['docker']:
# apk is in shared volume directory
apk = '/root/shared_volume/%s' % pytest_config_global['apk']
else:
apk = pytest_config_global['apk']
desired_caps['app'] = apk
desired_caps['deviceName'] = 'nexus_5'
desired_caps['platformName'] = 'Android'
desired_caps['appiumVersion'] = '1.9.1'
desired_caps['platformVersion'] = '10.0'
desired_caps['newCommandTimeout'] = 600
desired_caps['fullReset'] = False
desired_caps['unicodeKeyboard'] = True
desired_caps['automationName'] = 'UiAutomator2'
desired_caps['setWebContentDebuggingEnabled'] = True
return desired_caps
def add_local_devices_to_capabilities():
updated_capabilities = list()
raw_out = re.split(r'[\r\\n]+', str(subprocess.check_output(['adb', 'devices'])).rstrip())
for line in raw_out[1:]:
serial = re.findall(r"(([\d.\d:]*\d+)|\bemulator-\d+)", line)
if serial:
capabilities = get_capabilities_local()
capabilities['udid'] = serial[0][0]
updated_capabilities.append(capabilities)
return updated_capabilities
def get_capabilities_sauce_lab():
2023-08-12 18:04:28 +00:00
caps = dict()
caps['platformName'] = 'Android'
caps['idleTimeout'] = 1000
caps['appium:app'] = 'sauce-storage:' + test_suite_data.apk_name
caps['appium:deviceName'] = 'Android GoogleAPI Emulator'
caps['appium:deviceOrientation'] = 'portrait'
2023-12-09 00:25:19 +00:00
caps['appium:platformVersion'] = '14.0'
2023-08-12 18:04:28 +00:00
caps['appium:automationName'] = 'UiAutomator2'
caps['appium:newCommandTimeout'] = 600
caps['appium:idleTimeout'] = 1000
caps['appium:unicodeKeyboard'] = True
caps['appium:automationName'] = 'UiAutomator2'
caps['appium:setWebContentDebuggingEnabled'] = True
caps['appium:ignoreUnimportantViews'] = False
caps['ignoreUnimportantViews'] = False
caps['appium:enableNotificationListener'] = True
caps['enableNotificationListener'] = True
caps['appium:enforceXPath1'] = True
caps['enforceXPath1'] = True
caps['sauce:options'] = dict()
caps['sauce:options']['appiumVersion'] = '2.0.0'
caps['sauce:options']['username'] = sauce_username
caps['sauce:options']['accessKey'] = sauce_access_key
caps['sauce:options']['build'] = run_name
2023-08-12 18:04:28 +00:00
caps['sauce:options']['name'] = test_suite_data.current_test.name
caps['sauce:options']['maxDuration'] = 3600
caps['sauce:options']['idleTimeout'] = 1000
caps['sauce:options']['android.gpu.mode'] = 'hardware'
2023-09-06 03:07:12 +00:00
options = AppiumOptions()
options.load_capabilities(caps)
return options
2023-09-06 03:07:12 +00:00
# def update_capabilities_sauce_lab(new_capabilities: dict):
# caps = get_capabilities_sauce_lab().copy()
# caps.update(new_capabilities)
# return caps
2021-11-18 15:16:48 +00:00
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
2024-09-17 19:31:34 +00:00
def pull_geth(driver):
result = driver.pull_file(get_app_path() + 'geth.log')
return base64.b64decode(result)
2024-09-17 19:31:34 +00:00
def pull_requests_log(driver):
result = driver.pull_file(get_app_path() + 'requests.log')
return base64.b64decode(result)
class AbstractTestCase:
__metaclass__ = ABCMeta
def print_sauce_lab_info(self, driver):
sys.stdout = sys.stderr
print("SauceOnDemandSessionID=%s job-name=%s" % (driver.session_id, run_name))
def get_translation_by_key(self, key):
return transl[key]
@abstractmethod
def setup_method(self, method):
raise NotImplementedError('Should be overridden from a child class')
@abstractmethod
def teardown_method(self, method):
raise NotImplementedError('Should be overridden from a child class')
@property
def environment(self):
return pytest_config_global['env']
network_api = NetworkApi()
2021-11-18 15:16:48 +00:00
@staticmethod
2023-02-17 00:32:28 +00:00
def get_alert_text(driver):
try:
2023-02-17 00:32:28 +00:00
return driver.find_element(MobileBy.ID, 'android:id/message').text
except NoSuchElementException:
2023-02-17 00:32:28 +00:00
return None
def add_alert_text_to_report(self, driver):
2023-02-17 00:32:28 +00:00
try:
alert_text = self.get_alert_text(driver)
if alert_text:
test_suite_data.current_test.testruns[-1].error = "%s; also Unexpected Alert is shown: '%s'" % (
test_suite_data.current_test.testruns[-1].error, alert_text
)
2023-06-08 02:06:43 +00:00
except (RemoteDisconnected, ProtocolError):
2023-02-17 00:32:28 +00:00
test_suite_data.current_test.testruns[-1].error = "%s; \n RemoteDisconnected" % \
test_suite_data.current_test.testruns[-1].error
class Driver(webdriver.Remote):
@property
def number(self):
return test_suite_data.current_test.testruns[-1].jobs[self.session_id]
def info(self, text: str, device=True):
if device:
text = 'Device %s: %s ' % (self.number, text)
logging.info(text)
test_suite_data.current_test.testruns[-1].steps.append(text)
def fail(self, text: str):
pytest.fail('Device %s: %s' % (self.number, text))
class Errors(object):
def __init__(self):
self.errors = list()
def append(self, text=str()):
self.errors.append(text)
def verify_no_errors(self):
if self.errors:
pytest.fail('\n '.join([self.errors.pop(0) for _ in range(len(self.errors))]))
class SingleDeviceTestCase(AbstractTestCase):
def setup_method(self, method, **kwargs):
if pytest_config_global['docker']:
appium_container.start_appium_container(pytest_config_global['docker_shared_volume'])
appium_container.connect_device(pytest_config_global['device_ip'])
(executor, capabilities) = (executor_sauce_lab, get_capabilities_sauce_lab()) if \
self.environment == 'sauce' else (executor_local, get_capabilities_local())
for key, value in kwargs.items():
capabilities[key] = value
self.driver = Driver(executor, capabilities)
test_suite_data.current_test.testruns[-1].jobs[self.driver.session_id] = 1
self.driver.implicitly_wait(implicit_wait)
self.errors = Errors()
if pytest_config_global['docker']:
appium_container.reset_battery_stats()
def teardown_method(self, method):
if self.environment == 'sauce':
self.print_sauce_lab_info(self.driver)
try:
self.add_alert_text_to_report(self.driver)
geth_content = pull_geth(self.driver)
self.driver.quit()
if pytest_config_global['docker']:
appium_container.stop_container()
except (WebDriverException, AttributeError):
pass
finally:
2023-06-08 02:57:38 +00:00
github_report.save_test(test_suite_data.current_test,
{'%s_geth.log' % test_suite_data.current_test.name: geth_content})
class LocalMultipleDeviceTestCase(AbstractTestCase):
def setup_method(self, method):
self.drivers = dict()
self.errors = Errors()
def create_drivers(self, quantity):
capabilities = self.add_local_devices_to_capabilities()
for driver in range(quantity):
self.drivers[driver] = Driver(self.executor_local, capabilities[driver])
test_suite_data.current_test.testruns[-1].jobs[self.drivers[driver].session_id] = driver + 1
self.drivers[driver].implicitly_wait(self.implicitly_wait)
def teardown_method(self, method):
for driver in self.drivers:
try:
self.add_alert_text_to_report(self.drivers[driver])
self.drivers[driver].quit()
except WebDriverException:
pass
class SauceMultipleDeviceTestCase(AbstractTestCase):
@classmethod
def setup_class(cls):
cls.loop = asyncio.new_event_loop()
asyncio.set_event_loop(cls.loop)
def setup_method(self, method):
self.drivers = dict()
self.errors = Errors()
def create_drivers(self, quantity=2, max_duration=1800, custom_implicitly_wait=None):
self.drivers = self.loop.run_until_complete(start_threads(quantity,
Driver,
self.drivers,
executor_sauce_lab,
2023-09-06 03:07:12 +00:00
get_capabilities_sauce_lab()))
for driver in range(quantity):
test_suite_data.current_test.testruns[-1].jobs[self.drivers[driver].session_id] = driver + 1
self.drivers[driver].implicitly_wait(
custom_implicitly_wait if custom_implicitly_wait else implicit_wait)
def teardown_method(self, method):
geth_names, geth_contents = [], []
for driver in self.drivers:
try:
self.print_sauce_lab_info(self.drivers[driver])
self.add_alert_text_to_report(self.drivers[driver])
2021-11-18 15:16:48 +00:00
geth_names.append(
'%s_geth%s.log' % (test_suite_data.current_test.name, str(self.drivers[driver].number)))
geth_contents.append(pull_geth(self.drivers[driver]))
self.drivers[driver].quit()
except (WebDriverException, AttributeError):
pass
geth = {geth_names[i]: geth_contents[i] for i in range(len(geth_names))}
2023-06-08 02:57:38 +00:00
github_report.save_test(test_suite_data.current_test, geth)
@classmethod
def teardown_class(cls):
cls.loop.close()
def create_shared_drivers(quantity):
drivers = dict()
if pytest_config_global['env'] == 'local':
capabilities = add_local_devices_to_capabilities()
for i in range(quantity):
driver = Driver(executor_local, capabilities[i])
test_suite_data.current_test.testruns[-1].jobs[driver.session_id] = i + 1
driver.implicitly_wait(implicit_wait)
drivers[i] = driver
loop = None
return drivers, loop
else:
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
2022-11-03 11:26:53 +00:00
print('SC Executor: %s' % executor_sauce_lab)
try:
2023-07-04 17:26:05 +00:00
drivers = loop.run_until_complete(start_threads(test_suite_data.current_test.name,
quantity,
2023-06-08 02:06:43 +00:00
Driver,
drivers,
2023-09-06 03:07:12 +00:00
command_executor=executor_sauce_lab,
options=get_capabilities_sauce_lab()))
if len(drivers) < quantity:
test_suite_data.current_test.testruns[-1].error = "Not all %s drivers are created" % quantity
for i in range(quantity):
test_suite_data.current_test.testruns[-1].jobs[drivers[i].session_id] = i + 1
drivers[i].implicitly_wait(implicit_wait)
return drivers, loop
2023-07-04 17:26:05 +00:00
except (MaxRetryError, AttributeError) as e:
2023-12-09 00:25:19 +00:00
test_suite_data.current_test.testruns[-1].error = str(e)
for _, driver in drivers.items():
try:
driver.quit()
except (WebDriverException, AttributeError):
pass
2023-06-02 15:42:06 +00:00
raise e
class LocalSharedMultipleDeviceTestCase(AbstractTestCase):
def setup_method(self, method):
jobs = test_suite_data.current_test.testruns[-1].jobs
if not jobs:
for index, driver in self.drivers.items():
jobs[driver.session_id] = index + 1
self.errors = Errors()
def teardown_method(self, method):
for driver in self.drivers:
try:
self.add_alert_text_to_report(self.drivers[driver])
except WebDriverException:
pass
@pytest.fixture(scope='class', autouse=True)
def prepare(self, request):
try:
request.cls.prepare_devices(request)
finally:
for item, value in request.__dict__.items():
setattr(request.cls, item, value)
@classmethod
def teardown_class(cls):
for driver in cls.drivers:
try:
cls.drivers[driver].quit()
except WebDriverException:
pass
class SauceSharedMultipleDeviceTestCase(AbstractTestCase):
def setup_method(self, method):
if not self.drivers:
pytest.fail(test_suite_data.current_test.testruns[-1].error)
2022-01-24 11:40:02 +00:00
for _, driver in self.drivers.items():
driver.execute_script("sauce:context=Started %s" % method.__name__)
jobs = test_suite_data.current_test.testruns[-1].jobs
if not jobs:
for index, driver in self.drivers.items():
jobs[driver.session_id] = index + 1
self.errors = Errors()
test_suite_data.current_test.group_name = self.__class__.__name__
def teardown_method(self, method):
2024-09-17 19:31:34 +00:00
log_names, log_contents = [], []
for driver in self.drivers:
try:
self.print_sauce_lab_info(self.drivers[driver])
self.add_alert_text_to_report(self.drivers[driver])
2024-09-17 19:31:34 +00:00
log_names.append(
'%s_geth%s.log' % (test_suite_data.current_test.name, str(self.drivers[driver].number)))
2024-09-17 19:31:34 +00:00
log_contents.append(pull_geth(self.drivers[driver]))
log_names.append(
'%s_requests%s.log' % (test_suite_data.current_test.name, str(self.drivers[driver].number)))
log_contents.append(pull_requests_log(self.drivers[driver]))
2023-06-13 03:24:21 +00:00
except (WebDriverException, AttributeError, RemoteDisconnected, ProtocolError):
pass
finally:
2023-06-02 15:42:06 +00:00
try:
2024-09-17 19:31:34 +00:00
logs = {log_names[i]: log_contents[i] for i in range(len(log_names))}
test_suite_data.current_test.logs_paths = github_report.save_logs(logs)
2023-06-02 15:42:06 +00:00
except IndexError:
pass
@pytest.fixture(scope='class', autouse=True)
def prepare(self, request):
try:
request.cls.prepare_devices(request)
finally:
for item, value in request.__dict__.items():
setattr(request.cls, item, value)
@classmethod
def teardown_class(cls):
from tests.conftest import sauce
2022-01-24 11:40:02 +00:00
requests_session = requests.Session()
requests_session.auth = (sauce_username, sauce_access_key)
if test_suite_data.tests[0].testruns[-1].error and 'setup failed' in test_suite_data.tests[0].testruns[
-1].error:
group_setup_failed = True
else:
group_setup_failed = False
2024-09-17 19:31:34 +00:00
log_contents, log_names = list(), list()
2023-07-04 17:26:05 +00:00
try:
2024-09-17 19:31:34 +00:00
for i, driver in cls.drivers.items():
if group_setup_failed:
2024-09-17 19:31:34 +00:00
log_contents.append(pull_geth(driver=driver))
log_names.append('%s_geth%s.log' % (cls.__name__, i))
log_contents.append(pull_requests_log(driver=driver))
log_names.append('%s_requests%s.log' % (cls.__name__, i))
session_id = driver.session_id
try:
sauce.jobs.update_job(username=sauce_username, job_id=session_id, name=cls.__name__)
except (RemoteDisconnected, SauceException, requests.exceptions.ConnectionError):
pass
try:
driver.quit()
except WebDriverException:
pass
url = 'https://api.%s/rest/v1/%s/jobs/%s/assets/%s' % (apibase, sauce_username, session_id, "log.json")
try:
WebDriverWait(driver, 60, 2).until(lambda _: requests_session.get(url).status_code == 200)
commands = requests_session.get(url).json()
for command in commands:
try:
if command['message'].startswith("Started "):
for test in test_suite_data.tests:
if command['message'] == "Started %s" % test.name:
test.testruns[-1].first_commands[session_id] = commands.index(command) + 1
except KeyError:
continue
2023-07-20 10:55:06 +00:00
except (RemoteDisconnected, requests.exceptions.ConnectionError, TimeoutException):
pass
2023-07-04 17:26:05 +00:00
except AttributeError:
pass
finally:
2023-07-04 17:26:05 +00:00
try:
cls.loop.close()
except AttributeError:
2023-07-04 17:26:05 +00:00
pass
2024-09-17 19:31:34 +00:00
logs = dict(zip(log_names, log_contents))
logs_paths = github_report.save_logs(logs)
2022-01-24 11:40:02 +00:00
for test in test_suite_data.tests:
if group_setup_failed:
2024-09-17 19:31:34 +00:00
test.logs_paths = logs_paths
2023-06-08 02:57:38 +00:00
github_report.save_test(test)
if pytest_config_global['env'] == 'local':
MultipleDeviceTestCase = LocalMultipleDeviceTestCase
MultipleSharedDeviceTestCase = LocalSharedMultipleDeviceTestCase
else:
MultipleDeviceTestCase = SauceMultipleDeviceTestCase
MultipleSharedDeviceTestCase = SauceSharedMultipleDeviceTestCase
class NoDeviceTestCase(AbstractTestCase):
def setup_method(self, method, **kwargs):
pass
def teardown_method(self, method):
2023-06-08 02:57:38 +00:00
github_report.save_test(test_suite_data.current_test)