full_test_deposition.py

import argparse
import copy
import glob
import os
import sys
import unittest

from integration.partial.integration_test_utils import IntegrationTestUtils


class TestFullDeposition(unittest.TestCase):

    TEST_OUT_PATH: str = "not_set"
    TEST_START_DATE: str = 'NOT_SET'
    TEST_JOB_DIR: str = "DEPOSITION_" + TEST_START_DATE


    @classmethod
    def setUpClass(cls) -> None:
        TestFullDeposition.write_temp_run_config_file()
        TestFullDeposition.run_depo_pipeline()
        pass


    @staticmethod
    def write_temp_run_config_file():
        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'] = TestFullDeposition.TEST_OUT_PATH
        run_dict['WorkspacePath'] = TestFullDeposition.TEST_OUT_PATH

        IntegrationTestUtils.TEMP_CONFIG_FILE_PATH = TestFullDeposition.TEST_OUT_PATH + "temp_config.json"
        IntegrationTestUtils.write_json_file(run_dict, IntegrationTestUtils.TEMP_CONFIG_FILE_PATH)


    @staticmethod
    def run_depo_pipeline():
        component = 'Deposition'
        IntegrationTestUtils.run_external_pipeline(component,
                                                   TestFullDeposition.TEST_START_DATE,
                                                   IntegrationTestUtils.EMAIL_CRED_PATH)


    def test_standard_run_input_status_success(self):
        status_file_path = os.path.join(TestFullDeposition.TEST_OUT_PATH, TestFullDeposition.TEST_JOB_DIR, "STATUS_SUCCESS")
        success_file_exists: bool = os.path.isfile(status_file_path)
        self.assertTrue(success_file_exists)


    def test_standard_run_input_all_regions_ran(self):
        """
        working on the assumption that if there are images for each region, it must have run through
        (at least past the region iteration)
        """

        east_africa_image_path = os.path.join(TestFullDeposition.TEST_OUT_PATH, TestFullDeposition.TEST_JOB_DIR,
                                              "plotting", "eastafrica", "images", "Weekly",
                                              "deposition_eastafrica_leaf_rust_total_*_map.png")
        ethiopia_image_path = os.path.join(TestFullDeposition.TEST_OUT_PATH, TestFullDeposition.TEST_JOB_DIR,
                                           "plotting", "ethiopia", "images", "Weekly",
                                           "deposition_ethiopia_leaf_rust_total_*_map.png")

        ea_file_exists: bool = len(glob.glob(east_africa_image_path)) > 0
        eth_file_exists: bool = len(glob.glob(ethiopia_image_path)) > 0
        self.assertTrue(ea_file_exists)
        self.assertTrue(eth_file_exists)


    def test_standard_run_all_input_csvs_produced(self):
        east_africa_csv_path = os.path.join(TestFullDeposition.TEST_OUT_PATH, TestFullDeposition.TEST_JOB_DIR,
                                            "plotting", "eastafrica", "input_csvs", "*.csv")
        ethiopia_csv_path = os.path.join(TestFullDeposition.TEST_OUT_PATH, TestFullDeposition.TEST_JOB_DIR,
                                         "plotting", "ethiopia", "input_csvs", "*.csv")

        ea_csv_count: int = IntegrationTestUtils.count_files_in_wildcard(east_africa_csv_path)
        eth_csv_count: int = IntegrationTestUtils.count_files_in_wildcard(ethiopia_csv_path)
        self.assertEqual(9, ea_csv_count)
        self.assertEqual(9, eth_csv_count)


    def test_standard_run_all_images_produced(self):
        east_africa_image_path = os.path.join(TestFullDeposition.TEST_OUT_PATH, TestFullDeposition.TEST_JOB_DIR,
                                              "plotting", "eastafrica", "images", "Weekly", "*.png")
        ethiopia_image_path = os.path.join(TestFullDeposition.TEST_OUT_PATH, TestFullDeposition.TEST_JOB_DIR,
                                           "plotting", "ethiopia", "images", "Weekly", "*.png")

        ea_csv_count: int = IntegrationTestUtils.count_files_in_wildcard(east_africa_image_path)
        eth_csv_count: int = IntegrationTestUtils.count_files_in_wildcard(ethiopia_image_path)
        self.assertEqual(3, ea_csv_count)
        self.assertEqual(3, eth_csv_count)


if __name__ == '__main__':
    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('--run_date_type', required = False)
    parser.add_argument('--custom_run_date', required = False)
    parser.add_argument('unittest_args', nargs='*')

    _args = parser.parse_args()
    _config_file: str = _args.config
    _outdir: str = _args.outdir
    _email_cred_path: str = _args.email_cred
    _run_date_type: str = _args.run_date_type
    _custom_run_date: str = _args.custom_run_date

    IntegrationTestUtils.DEFAULT_CONFIG_FILE_PATH = _config_file

    TestFullDeposition.TEST_OUT_PATH = _outdir
    IntegrationTestUtils.EMAIL_CRED_PATH = _email_cred_path

    if _run_date_type == "today":
        print("today")
    elif _run_date_type == "yesterday":
        print("yesterday")
    elif _run_date_type == "custom":
        print(_custom_run_date)
        TestFullDeposition.TEST_START_DATE = _custom_run_date
        TestFullDeposition.TEST_JOB_DIR = "DEPOSITION_" + _custom_run_date
    else:
        print("default")
        TestFullDeposition.TEST_START_DATE = "20230126"
        TestFullDeposition.TEST_JOB_DIR = "DEPOSITION_" + TestFullDeposition.TEST_START_DATE

    # Now set the sys.argv to the unittest_args (leaving sys.argv[0] alone)
    sys.argv[1:] = _args.unittest_args
    unittest.main()