FAQ | This is a LIVE service | Changelog

Skip to content
Snippets Groups Projects
Commit 25afa675 authored by L. Bower's avatar L. Bower
Browse files

adding full env suit test

parent af6c89b7
No related branches found
No related tags found
No related merge requests found
import argparse
import copy
import os
import sys
import unittest
from integration.partial.integration_test_utils import IntegrationTestUtils
class FullTestEnvSuit(unittest.TestCase):
TEST_OUT_PATH: str = "NOT_SET"
TEST_START_DATE: str = 'NOT_SET'
TEST_JOB_DIR: str = "NOT_SET"
@classmethod
def setUpClass(cls) -> None:
FullTestEnvSuit.write_temp_run_config_file()
FullTestEnvSuit.run_env_pipeline()
@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'] = FullTestEnvSuit.TEST_OUT_PATH
run_dict['WorkspacePath'] = FullTestEnvSuit.TEST_OUT_PATH
run_dict['Environment']['WORK_PATH'] = FullTestEnvSuit.TEST_OUT_PATH
run_dict['Environment']['INPUT_PATH'] = FullTestEnvSuit.TEST_OUT_PATH
run_dict['Environment']['OUTPUT_PATH'] = FullTestEnvSuit.TEST_OUT_PATH
IntegrationTestUtils.TEMP_CONFIG_FILE_PATH = FullTestEnvSuit.TEST_OUT_PATH + "temp_config.json"
IntegrationTestUtils.write_json_file(run_dict, IntegrationTestUtils.TEMP_CONFIG_FILE_PATH)
@staticmethod
def run_env_pipeline():
component = 'Environment'
IntegrationTestUtils.run_external_pipeline(component,
FullTestEnvSuit.TEST_START_DATE,
IntegrationTestUtils.EMAIL_CRED_PATH)
def test_standard_run_input_status_success(self):
status_file_path = os.path.join(FullTestEnvSuit.TEST_OUT_PATH, FullTestEnvSuit.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(FullTestEnvSuit.TEST_OUT_PATH, FullTestEnvSuit.TEST_JOB_DIR,
"plotting", "eastafrica", "images", "Weekly",
"suitability_eastafrica_leaf_rust_total_202210010000_202210070000_map.png")
ethiopia_image_path = os.path.join(FullTestEnvSuit.TEST_OUT_PATH, FullTestEnvSuit.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(FullTestEnvSuit.TEST_OUT_PATH, FullTestEnvSuit.TEST_JOB_DIR,
"plotting", "eastafrica", "input_csvs", "*.csv")
ethiopia_csv_path = os.path.join(FullTestEnvSuit.TEST_OUT_PATH, FullTestEnvSuit.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(FullTestEnvSuit.TEST_OUT_PATH, FullTestEnvSuit.TEST_JOB_DIR,
"plotting", "eastafrica", "images", "Weekly", "*.png")
ethiopia_image_path = os.path.join(FullTestEnvSuit.TEST_OUT_PATH, FullTestEnvSuit.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__':
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
FullTestEnvSuit.TEST_OUT_PATH = _outdir
IntegrationTestUtils.EMAIL_CRED_PATH = _email_cred_path
FullTestEnvSuit.TEST_START_DATE = IntegrationTestUtils.generate_run_date(_run_date_type, _custom_run_date)
FullTestEnvSuit.TEST_JOB_DIR = "ENVIRONMENT" \
"_" + FullTestEnvSuit.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()
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment