integration_test_utils.py

import argparse
import datetime
import glob
import json
import os
from importlib import reload
from typing import List
from zipfile import ZipFile


class IntegrationTestUtils:

    EMAIL_CRED_PATH: str = "../../test_data/test_deployment/envs/Cred_gmail.json"
    DEFAULT_CONFIG_FILE_PATH: str = "../../test_data/test_deployment/regions/EastAfrica/resources/coordinator/configs/config_EastAfrica_fc_live.json"
    TEST_WORKSPACE_PATH: str = "../../test_data/test_deployment/regions/EastAfrica/workspace/"
    TEMP_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"

    @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 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_unittest_pipeline(component: str,
                              start_date: str,
                              **kwargs):

        #  need EMAIL_CRED in the environment before we import Processor
        os.environ["EMAIL_CRED"] = IntegrationTestUtils.EMAIL_CRED_PATH
        import Processor
        reload(Processor)
        from Processor import run_Process, set_log_level

        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['config_paths'] = [IntegrationTestUtils.TEMP_CONFIG_FILE_NAME]
        args_dict['log_level'] = 'info'
        args_dict['clearup'] = True

        for key, value in kwargs.items():
            args_dict[key] = value

        log_level = args_dict['log_level']
        set_log_level(log_level)

        try:
            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,
                              start_date: str,
                              email_cred_path: str,
                              **kwargs):

        #  need EMAIL_CRED in the environment before we import Processor
        os.environ["EMAIL_CRED"] = email_cred_path

        import Processor
        reload(Processor)
        from Processor import run_Process, set_log_level

        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['config_paths'] = [IntegrationTestUtils.TEMP_CONFIG_FILE_NAME]
        args_dict['log_level'] = 'info'
        args_dict['clearup'] = True

        for key, value in kwargs.items():
            args_dict[key] = value

        log_level = args_dict['log_level']
        set_log_level(log_level)

        try:
            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 = "20230126"

        return result