import argparse import copy import os import sys import unittest from unittest import TestSuite, TestLoader from HtmlTestRunner import HTMLTestRunner from integration.partial.integration_test_utils import IntegrationTestUtils from integration.test_suites.epi_test_suite import BaseEpiTestSuite class FullTestEpi(BaseEpiTestSuite.EpiTestSuite): def set_expected_values(self): # nothing to override here (outpouts overridden in the __main__ function pass def setUp(self) -> None: self.set_expected_values() if BaseEpiTestSuite.EpiTestSuite.TEST_OUT_PATH is None or not os.path.isdir(BaseEpiTestSuite.EpiTestSuite.TEST_OUT_PATH): # if True: FullTestEpi.write_temp_run_config_files() FullTestEpi.run_dependent_pipelines() FullTestEpi.run_epi_pipeline() else: print(f"output in {BaseEpiTestSuite.EpiTestSuite.TEST_OUT_PATH} already written, skipping rerun") @staticmethod def write_temp_run_config_files(): default_config = IntegrationTestUtils.DEFAULT_CONFIG_FILE_PATH default_config_dict: dict = IntegrationTestUtils.load_json_file(default_config) run_dict: dict = copy.deepcopy(default_config_dict) run_dict['WorkspacePathout'] = BaseEpiTestSuite.EpiTestSuite.TEST_OUT_PATH run_dict['WorkspacePath'] = BaseEpiTestSuite.EpiTestSuite.TEST_OUT_PATH # we need to run the env suit pipeline as well run_dict['Environment']['WORK_PATH'] = BaseEpiTestSuite.EpiTestSuite.TEST_OUT_PATH run_dict['Environment']['INPUT_PATH'] = BaseEpiTestSuite.EpiTestSuite.TEST_OUT_PATH run_dict['Environment']['OUTPUT_PATH'] = BaseEpiTestSuite.EpiTestSuite.TEST_OUT_PATH #make sure we activate server download run_dict['Survey']['SkipServerDownload'] = False # change the config for the Epi so it only runs on a single past date. Requires use of a different # FileListerFunction and a different calculation span run_dict['Epidemiology']['CalculationSpanDays'] = [0, 1] run_dict['Epidemiology']['Deposition']['FileListerFunction'] = "list_onefile_operational" run_dict['Epidemiology']['Environment']['FileListerFunction'] = "list_onefile_operational" # run_dict['Epidemiology']['Epi']['rescale_output_by_host_raster'] = False previous_day_string: str = IntegrationTestUtils.get_day_before_as_string(BaseEpiTestSuite.EpiTestSuite.TEST_START_DATE) run_dict['Survey']['SeasonStartString'] = previous_day_string # may be reusing a non-timestamped output file during development, so allow extant TEST_OUT_PATH os.makedirs(BaseEpiTestSuite.EpiTestSuite.TEST_OUT_PATH, exist_ok = True) IntegrationTestUtils.TEMP_CONFIG_FILE_NAME = BaseEpiTestSuite.EpiTestSuite.TEST_OUT_PATH + "temp_config.json" IntegrationTestUtils.write_json_file(run_dict, IntegrationTestUtils.TEMP_CONFIG_FILE_NAME) @staticmethod def run_dependent_pipelines(): IntegrationTestUtils.run_external_pipeline("Deposition", BaseEpiTestSuite.EpiTestSuite.TEST_START_DATE, IntegrationTestUtils.EMAIL_CRED_PATH) IntegrationTestUtils.run_external_pipeline("Environment", BaseEpiTestSuite.EpiTestSuite.TEST_START_DATE, IntegrationTestUtils.EMAIL_CRED_PATH) previous_day_string: str = IntegrationTestUtils.get_day_before_as_string(BaseEpiTestSuite.EpiTestSuite.TEST_START_DATE) IntegrationTestUtils.run_external_pipeline("Survey", previous_day_string, IntegrationTestUtils.EMAIL_CRED_PATH) pass @staticmethod def run_epi_pipeline(): component = 'Epidemiology' IntegrationTestUtils.run_external_pipeline(component, BaseEpiTestSuite.EpiTestSuite.TEST_START_DATE, IntegrationTestUtils.EMAIL_CRED_PATH) if __name__ == '__main__': _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 = "temp_epi_" + nowstring prefix: str = "temp_epi" BaseEpiTestSuite.EpiTestSuite.TEST_OUT_PATH = _outdir + prefix + os.sep IntegrationTestUtils.EMAIL_CRED_PATH = _email_cred_path BaseEpiTestSuite.EpiTestSuite.TEST_START_DATE = IntegrationTestUtils.generate_run_date(_run_date_type, _custom_run_date) BaseEpiTestSuite.EpiTestSuite.TEST_JOB_DIR = "EPI_" + BaseEpiTestSuite.EpiTestSuite.TEST_START_DATE # Now set the sys.argv to the unittest_args (leaving sys.argv[0] alone) sys.argv[1:] = _args.unittest_args tests: TestSuite = TestLoader().loadTestsFromTestCase(FullTestEpi) if _test_report_dir is None: _test_report_dir = BaseEpiTestSuite.EpiTestSuite.TEST_JOB_DIR runner = HTMLTestRunner(output=_test_report_dir, combine_reports = True) runner.run(tests)