Newer
Older
import copy
import os
import unittest
from integration.partial.integration_test_utils import IntegrationTestUtils
class TestEnvSuit(unittest.TestCase):
TEST_START_DATE: str = '20221001'
TEST_JOB_DIR: str = "ENVIRONMENT_2.0_" + TEST_START_DATE
@classmethod
def setUpClass(cls) -> None:
TestEnvSuit.write_temp_run_config_file()
# prefix: str = ""
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)
TestEnvSuit.TEST_OUT_PATH = run_dict['WorkspacePathout'] + prefix + os.sep
run_dict['WorkspacePathout'] = TestEnvSuit.TEST_OUT_PATH
run_dict['WorkspacePath'] = TestEnvSuit.TEST_OUT_PATH
run_dict['Environment']['WORK_PATH'] = TestEnvSuit.TEST_OUT_PATH
run_dict['Environment']['INPUT_PATH'] = TestEnvSuit.TEST_OUT_PATH
run_dict['Environment']['OUTPUT_PATH'] = TestEnvSuit.TEST_OUT_PATH
run_dict['ServerName'] = '' # nothing, as local machine
full_server_path = os.path.abspath(
"../../test_data/test_deployment/regions/EastAfrica/resources/coordinator/assets/")
run_dict['Environment']['ServerPathTemplate'] = full_server_path
IntegrationTestUtils.write_json_file(run_dict, IntegrationTestUtils.TEMP_CONFIG_FILE_PATH)
component = 'Environment'
IntegrationTestUtils.run_unittest_pipeline(component, TestEnvSuit.TEST_START_DATE)
def test_standard_run_input_status_success(self):
status_file_path = os.path.join(TestEnvSuit.TEST_OUT_PATH, TestEnvSuit.TEST_JOB_DIR, "STATUS_SUCCESS")
success_file_exists: bool = os.path.isfile(status_file_path)
self.assertTrue(success_file_exists)
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
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(TestEnvSuit.TEST_OUT_PATH, TestEnvSuit.TEST_JOB_DIR,
"plotting", "eastafrica", "images", "Weekly",
"suitability_eastafrica_leaf_rust_total_202210010000_202210070000_map.png")
ethiopia_image_path = os.path.join(TestEnvSuit.TEST_OUT_PATH, TestEnvSuit.TEST_JOB_DIR,
"plotting", "ethiopia", "images", "Weekly",
"suitability_ethiopia_leaf_rust_total_202210010000_202210070000_map.png")
ea_file_exists: bool = os.path.isfile(east_africa_image_path)
eth_file_exists: bool = os.path.isfile(ethiopia_image_path)
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(TestEnvSuit.TEST_OUT_PATH, TestEnvSuit.TEST_JOB_DIR,
"plotting", "eastafrica", "input_csvs", "*.csv")
ethiopia_csv_path = os.path.join(TestEnvSuit.TEST_OUT_PATH, TestEnvSuit.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(18, ea_csv_count)
self.assertEqual(18, eth_csv_count)
def test_standard_run_all_images_produced(self):
east_africa_image_path = os.path.join(TestEnvSuit.TEST_OUT_PATH, TestEnvSuit.TEST_JOB_DIR,
"plotting", "eastafrica", "images", "Weekly", "*.png")
ethiopia_image_path = os.path.join(TestEnvSuit.TEST_OUT_PATH, TestEnvSuit.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(6, ea_csv_count)
self.assertEqual(6, eth_csv_count)
if __name__ == '__main__':