import re from dataclasses import dataclass from datetime import datetime from http.client import RemoteDisconnected from os import environ import pytest from _pytest.runner import runtestprotocol from sauceclient import SauceClient, SauceException import tests from support.device_stats_db import DeviceStatsDB from support.test_rerun import should_rerun_test from tests import test_suite_data, appium_container # from support.testrail_report import TestrailReport # # testrail_report = TestrailReport() sauce_username = environ.get('SAUCE_USERNAME') sauce_access_key = environ.get('SAUCE_ACCESS_KEY') github_token = environ.get('GIT_HUB_TOKEN') sauce = SauceClient(sauce_username, sauce_access_key) def pytest_addoption(parser): parser.addoption("--build", action="store", default=datetime.now().strftime('%Y-%m-%d-%H-%M'), help="Specify build name") parser.addoption('--apk', action='store', default=None, help='Url or local path to apk') parser.addoption('--env', action='store', default='sauce', help='Specify environment: local/sauce/api') parser.addoption('--datacenter', action='store', default='eu-central-1', help='For sauce only: us-west-1, eu-central-1') parser.addoption('--platform_version', action='store', default='8.0', help='Android device platform version') parser.addoption('--log_steps', action='store', default=False, help='Display each test step in terminal as plain text: True/False') parser.addoption('--pr_number', action='store', default=None, help='Pull Request number') parser.addoption('--testrail_report', action='store', default=False, help='boolean; For creating testrail report per run') parser.addoption('--network', action='store', default='ropsten', help='string; ropsten or rinkeby') parser.addoption('--rerun_count', action='store', default=0, help='How many times tests should be re-run if failed') parser.addoption("--run_testrail_ids", action="store", metavar="NAME", default=None, help="only run tests matching the environment NAME.") parser.addoption("--apk_upgrade", action="store", metavar="NAME", default=None, help='Url or local path to apk for upgrade') # chat bot parser.addoption('--messages_number', action='store', default=20, help='Messages number') parser.addoption('--public_keys', action='store', default='', help='List of public keys for one-to-one chats') parser.addoption('--running_time', action='store', default=600, help='Running time in seconds') parser.addoption('--chat_name', action='store', default='test_chat', help='Public chat name') parser.addoption('--device_number', action='store', default=2, help='Public chat name') # running tests using appium docker instance parser.addoption('--docker', action='store', default=False, help='Are you using the appium docker container to run the tests?') parser.addoption('--docker_shared_volume', action='store', default=None, help='Path to a directory with .apk that will be shared with docker instance. Test reports will be also saved there') parser.addoption('--device_ip', action='store', default=None, help='Android device IP address used for battery tests') parser.addoption('--bugreport', action='store', default=False, help='Should generate bugreport for each test?') parser.addoption('--stats_db_host', action='store', default=None, help='Host address for device stats database') parser.addoption('--stats_db_port', action='store', default=8086, help='Port for device stats db') parser.addoption('--stats_db_username', action='store', default=None, help='Username for device stats db') parser.addoption('--stats_db_password', action='store', default=None, help='Password for device stats db') parser.addoption('--stats_db_database', action='store', default='example9', help='Database name for device stats db') @dataclass class Option: datacenter: str = None option = Option() testrail_report = None github_report = None def is_master(config): return not hasattr(config, 'workerinput') def is_uploaded(): from tests.cloudbase_test_api import sauce stored_files = sauce.storage.get_stored_files() for i in range(len(stored_files['files'])): if stored_files['files'][i]['name'] == test_suite_data.apk_name: return True def pytest_configure(config): global option option = config.option from support.testrail_report import TestrailReport global testrail_report testrail_report = TestrailReport() from support.github_report import GithubHtmlReport global github_report github_report = GithubHtmlReport() tests.pytest_config_global = vars(config.option) config.addinivalue_line("markers", "testrail_id(name): empty") if config.getoption('log_steps'): import logging logging.basicConfig(level=logging.INFO) if config.getoption('env') != 'api': test_suite_data.apk_name = ([i for i in [i for i in config.getoption('apk').split('/') if '.apk' in i]])[0] if is_master(config): pr_number = config.getoption('pr_number') if config.getoption('testrail_report'): if pr_number: run_number = len(testrail_report.get_runs(pr_number)) + 1 run_name = 'PR-%s run #%s' % (pr_number, run_number) else: run_name = test_suite_data.apk_name testrail_report.add_run(run_name) if pr_number: from github import Github repo = Github(github_token).get_user('status-im').get_repo('status-mobile') pull = repo.get_pull(int(pr_number)) pull.get_commits()[0].create_status(state='pending', context='Mobile e2e tests', description='e2e tests are running') if config.getoption('env') == 'sauce': if not is_uploaded(): from tests.cloudbase_test_api import sauce, upload_from_url if 'http' in config.getoption('apk'): upload_from_url(config.getoption('apk')) else: sauce.storage.upload_file(config.getoption('apk')) def pytest_unconfigure(config): if is_master(config): if config.getoption('testrail_report'): testrail_report.add_results() if config.getoption('pr_number'): from github import Github repo = Github(github_token).get_user('status-im').get_repo('status-mobile') pull = repo.get_pull(int(config.getoption('pr_number'))) comment = pull.create_issue_comment(github_report.build_html_report(testrail_report.run_id)) if not testrail_report.is_run_successful(): pull.get_commits()[0].create_status(state='failure', context='Mobile e2e tests', description='Failure - e2e tests are failed', target_url=comment.html_url) else: pull.get_commits()[0].create_status(state='success', context='Mobile e2e tests', description='Success - e2e tests are passed', target_url=comment.html_url) def should_save_device_stats(config): db_args = [config.getoption(option) for option in ('stats_db_host', 'stats_db_port', 'stats_db_username', 'stats_db_password', 'stats_db_database')] return all(db_args) @pytest.mark.hookwrapper def pytest_runtest_makereport(item, call): outcome = yield report = outcome.get_result() is_sauce_env = item.config.getoption('env') == 'sauce' def catch_error(): error = report.longreprtext failure_pattern = 'E.*Message:|E.*Error:|E.*Failed:' exception = re.findall(failure_pattern, error) if exception: error = error.replace(re.findall(failure_pattern, report.longreprtext)[0], '') return error if report.when == 'setup': is_group = "xdist_group" in item.keywords._markers or "xdist_group" in item.parent.keywords._markers error_intro, error = 'Test setup failed:', '' final_error = '%s %s' % (error_intro, error) if hasattr(report, 'wasxfail') and str([mark.args[0] for mark in item.iter_markers(name='testrail_id')][0]) \ in str(item.config.getoption("run_testrail_ids")): if '[NOTRUN]' in report.wasxfail: test_suite_data.set_current_test(item.name, testrail_case_id=get_testrail_case_id(item)) test_suite_data.current_test.create_new_testrun() if is_group: test_suite_data.current_test.group_name = item.instance.__class__.__name__ error_intro, error = 'Test is not run, e2e blocker ', report.wasxfail final_error = "%s [[%s]]" % (error_intro, error) else: if is_group: test_suite_data.current_test.group_name = item.instance.__class__.__name__ error = catch_error() final_error = '%s %s [[%s]]' % (error_intro, error, report.wasxfail) else: if is_group and report.failed: test_suite_data.current_test.group_name = item.instance.__class__.__name__ error = catch_error() final_error = '%s %s' % (error_intro, error) if is_sauce_env: update_sauce_jobs(test_suite_data.current_test.group_name, test_suite_data.current_test.testruns[-1].jobs, report.passed) if error: test_suite_data.current_test.testruns[-1].error = final_error from support.github_report import GithubHtmlReport GithubHtmlReport().save_test(test_suite_data.current_test) if report.when == 'call': current_test = test_suite_data.current_test error = catch_error() if report.failed: current_test.testruns[-1].error = error in_run = str([mark.args[0] for mark in item.iter_markers(name='testrail_id')][0]) in str(item.config.getoption( "run_testrail_ids")) if hasattr(report, 'wasxfail') and in_run: current_test.testruns[-1].xfail = report.wasxfail if error: current_test.testruns[-1].error = '%s [[%s]]' % (error, report.wasxfail) if is_sauce_env: update_sauce_jobs(current_test.name, current_test.testruns[-1].jobs, report.passed) if item.config.getoption('docker'): device_stats = appium_container.get_device_stats() if item.config.getoption('bugreport'): appium_container.generate_bugreport(item.name) build_name = item.config.getoption('apk') # Find type of tests that are run on the device if 'battery_consumption' in item.keywords._markers: test_group = 'battery_consumption' else: test_group = None if should_save_device_stats(item.config): device_stats_db = DeviceStatsDB( item.config.getoption('stats_db_host'), item.config.getoption('stats_db_port'), item.config.getoption('stats_db_username'), item.config.getoption('stats_db_password'), item.config.getoption('stats_db_database'), ) device_stats_db.save_stats(build_name, item.name, test_group, not report.failed, device_stats) def update_sauce_jobs(test_name, job_ids, passed): from tests.cloudbase_test_api import sauce for job_id in job_ids.keys(): try: sauce.jobs.update_job(job_id, name=test_name, passed=passed) except (RemoteDisconnected, SauceException): pass def get_testrail_case_id(item): testrail_id = item.get_closest_marker('testrail_id') if testrail_id: return testrail_id.args[0] def pytest_runtest_setup(item): try: testrail_id = [mark.args[0] for mark in item.iter_markers(name='testrail_id')][0] except IndexError: pass run_testrail_ids = item.config.getoption("run_testrail_ids") if run_testrail_ids: if str(testrail_id) not in list(run_testrail_ids.split(",")): pytest.skip("test requires testrail case id %s" % testrail_id) test_suite_data.set_current_test(item.name, testrail_case_id=get_testrail_case_id(item)) test_suite_data.current_test.create_new_testrun() def pytest_runtest_protocol(item, nextitem): rerun_count = int(item.config.getoption('rerun_count')) for i in range(rerun_count): reports = runtestprotocol(item, nextitem=nextitem) for report in reports: is_in_group = [i for i in item.iter_markers(name='xdist_group')] if report.failed and should_rerun_test(report.longreprtext) and not is_in_group: break # rerun else: return True # no need to rerun # @pytest.fixture(scope="session", autouse=False) # def faucet_for_senders(): # network_api = NetworkApi() # for user in transaction_senders.values(): # network_api.faucet(address=user['address']) @pytest.fixture def messages_number(request): return int(request.config.getoption('messages_number')) @pytest.fixture def message_wait_time(request): return int(request.config.getoption('message_wait_time')) @pytest.fixture def participants_number(request): return int(request.config.getoption('participants_number')) @pytest.fixture def chat_name(request): return request.config.getoption('chat_name') @pytest.fixture def user_public_key(request): return request.config.getoption('user_public_key')