import argparse import datetime import glob import json import os from typing import List from unittest import TestSuite, TestLoader, TestCase, TestResult from zipfile import ZipFile from HTMLTestRunner import HTMLTestRunner from ews.coordinator.processor_base import ProcessorBase class IntegrationTestUtils: EMAIL_CRED_PATH: str = "../../test_data/test_deployment/envs/Cred_gmail.json" LOGGING_CONFIG_PATH: str = "../../test_data/test_deployment/envs/test_log_config.json" DEFAULT_SYS_CONFIG_FILE_PATH: str = "../../test_data/test_deployment/regions/EastAfrica/resources/coordinator/configs/sys_config_EastAfrica_fc_live.json" DEFAULT_DEPO_CONFIG_FILE_PATH: str = "../../test_data/test_deployment/regions/EastAfrica/resources/coordinator/configs/depo_config_EastAfrica_fc_live.json" DEFAULT_ENV_SUIT_CONFIG_FILE_PATH: str = "../../test_data/test_deployment/regions/EastAfrica/resources/coordinator/configs/env_suit_config_EastAfrica_fc_live.json" DEFAULT_EPI_CONFIG_FILE_PATH: str = "../../test_data/test_deployment/regions/EastAfrica/resources/coordinator/configs/epi_config_EastAfrica_fc_live.json" DEFAULT_SURVEY_CONFIG_FILE_PATH: str = "../../test_data/test_deployment/regions/EastAfrica/resources/coordinator/configs/survey_config_EastAfrica_fc_live.json" DEFAULT_ADVISORY_CONFIG_FILE_PATH: str = "../../test_data/test_deployment/regions/EastAfrica/resources/coordinator/configs/advisory_config_EastAfrica_fc_live.json" TEST_WORKSPACE_PATH: str = "../../test_data/test_deployment/regions/EastAfrica/workspace/" TEMP_SYS_CONFIG_FILE_NAME: str = None TEMP_RUN_CONFIG_FILE_NAME: str = None TEST_ASSETS_PATH: str = "../../test_data/test_deployment/regions/EastAfrica/resources/coordinator/assets/" EXAMPLE_SURVEY_FILE_PATH: str = TEST_ASSETS_PATH + "example_survey_run.zip" EXAMPLE_DEPO_FILE_PATH: str = TEST_ASSETS_PATH + "example_depo_run.zip" EXAMPLE_ENV_SUIT_FILE_PATH: str = TEST_ASSETS_PATH + "example_env_suit_run.zip" TEST_OUT_PATH: str = None TEST_START_DATE: str = None TEST_JOB_DIR: str = None @staticmethod def check_resources_exist(): """ do a single check for the email cred file, which will raise an error suggesting that the user may not have set their pythonpath correctly """ emailcred_exists = os.path.exists(IntegrationTestUtils.EMAIL_CRED_PATH) if not emailcred_exists: raise FileNotFoundError(f"email cred file {IntegrationTestUtils.EMAIL_CRED_PATH} not found - have you set " f"the working directory to the same directory as the tests you are running?" f" (tests use a path relative to the test directory)") @staticmethod def build_arg_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser() parser.add_argument('--config', required = True) parser.add_argument('--outdir', required = True) parser.add_argument('--email_cred', required = True) parser.add_argument('--test_report_dir', required = False) parser.add_argument('--run_date_type', required = False) parser.add_argument('--custom_run_date', required = False) parser.add_argument('unittest_args', nargs='*') return parser @staticmethod def run_full_integration_test_pipeline(test_case: [TestCase], test_prefix: str, processor_dir: str) -> bool: """ Runs the full integration tests on the production server GitlabRunner. The full integration tests are run on a GitlabRunner that runs on the production server, and mounts the local file system to use the production configuration files. :param test_case: :param test_prefix: :param processor_dir: :return: """ _parser = IntegrationTestUtils.build_arg_parser() _args = _parser.parse_args() _config_file: str = _args.config _outdir: str = _args.outdir _email_cred_path: str = _args.email_cred _test_report_dir: str = _args.test_report_dir _run_date_type: str = _args.run_date_type _custom_run_date: str = _args.custom_run_date IntegrationTestUtils.DEFAULT_CONFIG_FILE_PATH = _config_file nowstring: str = IntegrationTestUtils.get_now_string() prefix: str = f"temp_{test_prefix}_" + nowstring # prefix: str = f"temp_{test_prefix}" IntegrationTestUtils.TEST_OUT_PATH = _outdir + prefix + os.sep IntegrationTestUtils.EMAIL_CRED_PATH = _email_cred_path IntegrationTestUtils.TEST_START_DATE = IntegrationTestUtils.generate_run_date(_run_date_type, _custom_run_date) IntegrationTestUtils.TEST_JOB_DIR = os.path.join(IntegrationTestUtils.TEST_OUT_PATH, f"{processor_dir}_" + IntegrationTestUtils.TEST_START_DATE) tests: TestSuite = TestLoader().loadTestsFromTestCase(test_case) if _test_report_dir is None: _test_report_dir = IntegrationTestUtils.TEST_JOB_DIR """ the HTMLTestRunner will create the directory if it does not exist, we are putting the output into the TEST_OUT_PATH, which is the top-level test directory for this run, not the TEST_JOB_DIR, which is the dir for the pipeline being tested """ runner = HTMLTestRunner(output = IntegrationTestUtils.TEST_OUT_PATH) result: TestResult = runner.run(tests) return result.wasSuccessful() @staticmethod def load_json_file(file: str) -> dict: with open(file) as config_file: config: dict = json.load(config_file) return config @staticmethod def write_json_file(values: dict, file: str): with open(file, 'w') as file: json.dump(values, file, indent = 4) @staticmethod def get_now_string() -> str: nowstring: str = datetime.datetime.today().strftime('%Y-%m-%d_%H%M%S') return nowstring @staticmethod def unpack_zip(zip_to_unpack: str, out_file: str): with ZipFile(zip_to_unpack) as zf: # open the zip file for target_file in zf.namelist(): # check if the file exists in the archive zf.extract(target_file, out_file) @staticmethod def count_files_in_wildcard(wildcard: str) -> int: results = glob.glob(wildcard) return len(results) @staticmethod def check_file_not_empty(file_path: str) -> bool: return os.stat(file_path).st_size != 0 @staticmethod def check_file_exists(file_path: str) -> bool: return os.path.isfile(file_path) @staticmethod def check_wildcard_exists_and_not_empty(wildcard: str) -> bool: """ requires at least one file matching the wildcard to exist and not be empty """ result = False files: List[str] = glob.glob(wildcard) for file in files: result = IntegrationTestUtils.check_file_not_empty(file) if result is False: break return result @staticmethod def check_file_exists_and_not_empty(file_path: str) -> bool: file_exists = IntegrationTestUtils.check_file_exists(file_path) file_not_empty = IntegrationTestUtils.check_file_not_empty(file_path) return file_exists and file_not_empty @staticmethod def run_partial_integration_test_pipeline(component: str, shortname: str, start_date: str, processor: ProcessorBase, **kwargs): """ Runs the "run_Process" function in Processor.py with the given arguments for the partial integration tests. The full integration pipeline is run in the "run_full_integration_test_pipeline" function. :param processor: :param component: :param shortname: :param start_date: :param kwargs: :return: """ args_dict: dict = {} sys_config_path = IntegrationTestUtils.TEMP_SYS_CONFIG_FILE_NAME config_paths = [IntegrationTestUtils.TEMP_RUN_CONFIG_FILE_NAME] # note, possible to override these values in the kwargs loop below args_dict['live'] = False args_dict['noupload'] = True args_dict['start_date'] = start_date args_dict['component'] = component args_dict['short_name'] = shortname args_dict['sys_config_path'] = sys_config_path args_dict['config_paths'] = config_paths args_dict['log_level'] = 'info' args_dict['clearup'] = True for key, value in kwargs.items(): args_dict[key] = value # need EMAIL_CRED in the environment before we run a Processor os.environ["EMAIL_CRED"] = IntegrationTestUtils.EMAIL_CRED_PATH # need LOGGING_CONFIG_PATH in the environment before we run a Processor os.environ["LOGGING_CONFIG"] = IntegrationTestUtils.LOGGING_CONFIG_PATH try: processor.run_process(args_dict) except SystemExit: # we will eventually want to throw these to the calling class to be dealt with pass @staticmethod def run_external_pipeline(component: str, short_name: str, start_date: str, processor: ProcessorBase, **kwargs): args_dict: dict = {} # note, possible to override these values in the kwargs loop below args_dict['live'] = False args_dict['noupload'] = True args_dict['start_date'] = start_date args_dict['component'] = component args_dict['short_name'] = short_name args_dict['config_paths'] = [IntegrationTestUtils.TEMP_RUN_CONFIG_FILE_NAME] args_dict['log_level'] = 'info' args_dict['clearup'] = True for key, value in kwargs.items(): args_dict[key] = value # need EMAIL_CRED in the environment before we run a Processor os.environ["EMAIL_CRED"] = IntegrationTestUtils.EMAIL_CRED_PATH # need LOGGING_CONFIG_PATH in the environment before we run a Processor os.environ["LOGGING_CONFIG"] = IntegrationTestUtils.LOGGING_CONFIG_PATH try: processor.run_process(args_dict) except SystemExit: # we will eventually want to throw these to the calling class to be dealt with pass @staticmethod def get_day_before_as_string(input_date_string: str) -> str: date_format = "%Y%m%d" input_date: datetime = datetime.datetime.strptime(input_date_string, date_format) yesterday_date = input_date - datetime.timedelta(days = 1) yesterday_string = yesterday_date.strftime(date_format) return yesterday_string @staticmethod def generate_run_date(run_date_type: str, custom_run_date: str) -> str: if run_date_type == "today": print("today") today = datetime.date.today() today_string = today.strftime("%Y%m%d") result = today_string elif run_date_type == "yesterday": print("yesterday") today = datetime.date.today() yesterday: datetime = today - datetime.timedelta(days = 1) yesterday_string = yesterday.strftime("%Y%m%d") result = yesterday_string elif run_date_type == "custom": result = custom_run_date else: print("default") result = "20231127" return result