diff --git a/tests/integration/partial/test_advisory.py b/tests/integration/partial/test_advisory.py
index c559f04da2f3a74a8565962f49794189e469bb76..d79c423ff0efd88e49a0852fb2bccff9a321d69c 100644
--- a/tests/integration/partial/test_advisory.py
+++ b/tests/integration/partial/test_advisory.py
@@ -9,8 +9,8 @@ from integration.test_suites.advisory_test_suite import BaseAdvisoryTestSuite
class TestAdvisory(BaseAdvisoryTestSuite.AdvisoryTestSuite):
def set_expected_values(self):
- BaseAdvisoryTestSuite.AdvisoryTestSuite.TEST_START_DATE = '20221001'
- BaseAdvisoryTestSuite.AdvisoryTestSuite.TEST_JOB_DIR = "ADVISORY_" + BaseAdvisoryTestSuite.AdvisoryTestSuite.TEST_START_DATE
+ IntegrationTestUtils.TEST_START_DATE = '20221001'
+ IntegrationTestUtils.TEST_JOB_DIR = "ADVISORY_" + IntegrationTestUtils.TEST_START_DATE
self.EXPECTED_EA_IMAGE_COUNT = 3
self.EXPECTED_ETH_IMAGE_COUNT = 3
self.EXPECTED_EA_SHAPE_COUNT = 3
@@ -20,43 +20,46 @@ class TestAdvisory(BaseAdvisoryTestSuite.AdvisoryTestSuite):
self.set_expected_values()
- if BaseAdvisoryTestSuite.AdvisoryTestSuite.TEST_OUT_PATH is None or not os.path.isdir(BaseAdvisoryTestSuite.AdvisoryTestSuite.TEST_OUT_PATH):
+ if IntegrationTestUtils.TEST_OUT_PATH is None or not os.path.isdir(IntegrationTestUtils.TEST_OUT_PATH):
TestAdvisory.write_temp_run_config_files()
TestAdvisory.unpack_dependencies()
TestAdvisory.run_advisory_pipeline()
else:
- print(f"output in {BaseAdvisoryTestSuite.AdvisoryTestSuite.TEST_OUT_PATH} already written, skipping rerun")
+ print(f"output in {IntegrationTestUtils.TEST_OUT_PATH} already written, skipping rerun")
@staticmethod
def write_temp_run_config_files():
nowstring: str = IntegrationTestUtils.get_now_string()
- # prefix: str = "temp_advisory_" + nowstring
- prefix: str = "temp_advisory"
+ prefix: str = "temp_advisory_" + nowstring
+ # prefix: str = "temp_advisory"
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)
- BaseAdvisoryTestSuite.AdvisoryTestSuite.TEST_OUT_PATH = IntegrationTestUtils.TEST_WORKSPACE_PATH + prefix + os.sep
+ IntegrationTestUtils.TEST_OUT_PATH = IntegrationTestUtils.TEST_WORKSPACE_PATH + prefix + os.sep
# TestAdvisory.TEST_OUT_PATH = run_dict['WorkspacePathout'] + prefix + os.sep
- run_dict['WorkspacePathout'] = BaseAdvisoryTestSuite.AdvisoryTestSuite.TEST_OUT_PATH
- run_dict['WorkspacePath'] = BaseAdvisoryTestSuite.AdvisoryTestSuite.TEST_OUT_PATH
+ run_dict['WorkspacePathout'] = IntegrationTestUtils.TEST_OUT_PATH
+ run_dict['WorkspacePath'] = IntegrationTestUtils.TEST_OUT_PATH
run_dict['ServerName'] = '' # nothing, as local machine
+ # may be reusing a non-timestamped output file during development, so allow extant TEST_OUT_PATH
+ os.makedirs(IntegrationTestUtils.TEST_OUT_PATH, exist_ok = True)
+ IntegrationTestUtils.TEMP_CONFIG_FILE_NAME = IntegrationTestUtils.TEST_OUT_PATH + "temp_config.json"
IntegrationTestUtils.write_json_file(run_dict, IntegrationTestUtils.TEMP_CONFIG_FILE_NAME)
@staticmethod
def unpack_dependencies():
- IntegrationTestUtils.unpack_zip(IntegrationTestUtils.EXAMPLE_SURVEY_FILE_PATH, TestAdvisory.TEST_OUT_PATH)
- IntegrationTestUtils.unpack_zip(IntegrationTestUtils.EXAMPLE_DEPO_FILE_PATH, TestAdvisory.TEST_OUT_PATH)
- IntegrationTestUtils.unpack_zip(IntegrationTestUtils.EXAMPLE_ENV_SUIT_FILE_PATH, TestAdvisory.TEST_OUT_PATH)
+ IntegrationTestUtils.unpack_zip(IntegrationTestUtils.EXAMPLE_SURVEY_FILE_PATH, IntegrationTestUtils.TEST_OUT_PATH)
+ IntegrationTestUtils.unpack_zip(IntegrationTestUtils.EXAMPLE_DEPO_FILE_PATH, IntegrationTestUtils.TEST_OUT_PATH)
+ IntegrationTestUtils.unpack_zip(IntegrationTestUtils.EXAMPLE_ENV_SUIT_FILE_PATH, IntegrationTestUtils.TEST_OUT_PATH)
@staticmethod
def run_advisory_pipeline():
component = 'Advisory'
- IntegrationTestUtils.run_unittest_pipeline(component, TestAdvisory.TEST_START_DATE)
+ IntegrationTestUtils.run_unittest_pipeline(component, IntegrationTestUtils.TEST_START_DATE)
if __name__ == '__main__':
diff --git a/tests/integration/partial/test_survey.py b/tests/integration/partial/test_survey.py
index da26e014ded19f9edfb42040ab311dd8ea3f1b09..faca2fce1d3e031acf11466e90c01967a223565e 100644
--- a/tests/integration/partial/test_survey.py
+++ b/tests/integration/partial/test_survey.py
@@ -9,21 +9,21 @@ from integration.test_suites.survey_test_suite import BaseSurveyTestSuite
class TestSurvey(BaseSurveyTestSuite.SurveyTestSuite):
def set_expected_values(self):
- BaseSurveyTestSuite.SurveyTestSuite.TEST_START_DATE = '20221001'
- BaseSurveyTestSuite.SurveyTestSuite.TEST_JOB_DIR = "SURVEYDATA_" + BaseSurveyTestSuite.SurveyTestSuite.TEST_START_DATE
+ IntegrationTestUtils.TEST_START_DATE = '20221001'
+ IntegrationTestUtils.TEST_JOB_DIR = "SURVEYDATA_" + IntegrationTestUtils.TEST_START_DATE
def setUp(self) -> None:
self.set_expected_values()
- if BaseSurveyTestSuite.SurveyTestSuite.TEST_OUT_PATH is None or not os.path.isdir(
- BaseSurveyTestSuite.SurveyTestSuite.TEST_OUT_PATH):
+ if IntegrationTestUtils.TEST_OUT_PATH is None or not os.path.isdir(
+ IntegrationTestUtils.TEST_OUT_PATH):
TestSurvey.write_temp_run_config_file()
TestSurvey.unpack_dependencies()
TestSurvey.run_survey_pipeline()
else:
- print(f"output in {BaseSurveyTestSuite.SurveyTestSuite.TEST_OUT_PATH} already written, skipping rerun")
+ print(f"output in {IntegrationTestUtils.TEST_OUT_PATH} already written, skipping rerun")
@staticmethod
@@ -35,23 +35,27 @@ class TestSurvey(BaseSurveyTestSuite.SurveyTestSuite):
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)
- # TestSurvey.TEST_OUT_PATH = run_dict['WorkspacePathout'] + prefix + os.sep
- BaseSurveyTestSuite.SurveyTestSuite.TEST_OUT_PATH = IntegrationTestUtils.TEST_WORKSPACE_PATH + prefix + os.sep
- run_dict['WorkspacePathout'] = TestSurvey.TEST_OUT_PATH
- run_dict['WorkspacePath'] = TestSurvey.TEST_OUT_PATH
- run_dict['Survey']['SkipServerDownload'] = True
+ IntegrationTestUtils.TEST_OUT_PATH = IntegrationTestUtils.TEST_WORKSPACE_PATH + prefix + os.sep
+ run_dict['WorkspacePathout'] = IntegrationTestUtils.TEST_OUT_PATH
+ run_dict['WorkspacePath'] = IntegrationTestUtils.TEST_OUT_PATH
+ run_dict['Environment']['WORK_PATH'] = IntegrationTestUtils.TEST_OUT_PATH
+ run_dict['Environment']['INPUT_PATH'] = IntegrationTestUtils.TEST_OUT_PATH
+ run_dict['Environment']['OUTPUT_PATH'] = IntegrationTestUtils.TEST_OUT_PATH
run_dict['ServerName'] = '' # nothing, as local machine
+ # may be reusing a non-timestamped output file during development, so allow extant TEST_OUT_PATH
+ os.makedirs(IntegrationTestUtils.TEST_OUT_PATH, exist_ok = True)
+ IntegrationTestUtils.TEMP_CONFIG_FILE_NAME = IntegrationTestUtils.TEST_OUT_PATH + "temp_config.json"
IntegrationTestUtils.write_json_file(run_dict, IntegrationTestUtils.TEMP_CONFIG_FILE_NAME)
@staticmethod
def unpack_dependencies():
- IntegrationTestUtils.unpack_zip(IntegrationTestUtils.EXAMPLE_SURVEY_FILE_PATH, TestSurvey.TEST_OUT_PATH)
+ IntegrationTestUtils.unpack_zip(IntegrationTestUtils.EXAMPLE_SURVEY_FILE_PATH, IntegrationTestUtils.TEST_OUT_PATH)
@staticmethod
def run_survey_pipeline():
component = 'Survey'
- IntegrationTestUtils.run_unittest_pipeline(component, TestSurvey.TEST_START_DATE)
+ IntegrationTestUtils.run_unittest_pipeline(component, IntegrationTestUtils.TEST_START_DATE)
if __name__ == '__main__':
diff --git a/tests/integration/test_suites/advisory_test_suite.py b/tests/integration/test_suites/advisory_test_suite.py
index 03b3f6cebb7ea96a38a8819eee687fe7f72f052b..f46695f8869c8ca303c4f8f2758495babeca464f 100644
--- a/tests/integration/test_suites/advisory_test_suite.py
+++ b/tests/integration/test_suites/advisory_test_suite.py
@@ -10,10 +10,6 @@ class BaseAdvisoryTestSuite:
class AdvisoryTestSuite(unittest.TestCase):
- TEST_OUT_PATH: str = None
- TEST_START_DATE: str = None
- TEST_JOB_DIR: str = None
-
@abc.abstractmethod
def set_expected_values(self):
self.EXPECTED_EA_IMAGE_COUNT = 0
@@ -22,18 +18,18 @@ class BaseAdvisoryTestSuite:
def test_standard_run_input_status_success(self):
- status_file_path = os.path.join(BaseAdvisoryTestSuite.AdvisoryTestSuite.TEST_OUT_PATH, BaseAdvisoryTestSuite.AdvisoryTestSuite.TEST_JOB_DIR, "STATUS_SUCCESS")
+ status_file_path = os.path.join(IntegrationTestUtils.TEST_OUT_PATH, IntegrationTestUtils.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_docs_produced(self):
- east_africa_image_wildcard = os.path.join(BaseAdvisoryTestSuite.AdvisoryTestSuite.TEST_OUT_PATH,
- BaseAdvisoryTestSuite.AdvisoryTestSuite.TEST_JOB_DIR,
+ east_africa_image_wildcard = os.path.join(IntegrationTestUtils.TEST_OUT_PATH,
+ IntegrationTestUtils.TEST_JOB_DIR,
"tight-layout",
"wheat_rust_advisory_template_EastAfrica_*.docx")
- ethiopia_image_wildcard = os.path.join(BaseAdvisoryTestSuite.AdvisoryTestSuite.TEST_OUT_PATH,
- BaseAdvisoryTestSuite.AdvisoryTestSuite.TEST_JOB_DIR,
+ ethiopia_image_wildcard = os.path.join(IntegrationTestUtils.TEST_OUT_PATH,
+ IntegrationTestUtils.TEST_JOB_DIR,
"tight-layout",
"wheat_rust_advisory_template_Ethiopia_*.docx")
@@ -44,12 +40,12 @@ class BaseAdvisoryTestSuite:
def test_standard_run_input_images_produced(self):
- east_africa_image_wildcard = os.path.join(BaseAdvisoryTestSuite.AdvisoryTestSuite.TEST_OUT_PATH,
- BaseAdvisoryTestSuite.AdvisoryTestSuite.TEST_JOB_DIR,
+ east_africa_image_wildcard = os.path.join(IntegrationTestUtils.TEST_OUT_PATH,
+ IntegrationTestUtils.TEST_JOB_DIR,
"images",
"*eastafrica*.png")
- ethiopia_image_wildcard = os.path.join(BaseAdvisoryTestSuite.AdvisoryTestSuite.TEST_OUT_PATH,
- BaseAdvisoryTestSuite.AdvisoryTestSuite.TEST_JOB_DIR,
+ ethiopia_image_wildcard = os.path.join(IntegrationTestUtils.TEST_OUT_PATH,
+ IntegrationTestUtils.TEST_JOB_DIR,
"images",
"*ethiopia*.png")
@@ -60,8 +56,8 @@ class BaseAdvisoryTestSuite:
def test_standard_run_input_shapefiles_produced(self):
- east_africa_image_wildcard = os.path.join(BaseAdvisoryTestSuite.AdvisoryTestSuite.TEST_OUT_PATH,
- BaseAdvisoryTestSuite.AdvisoryTestSuite.TEST_JOB_DIR,
+ east_africa_image_wildcard = os.path.join(IntegrationTestUtils.TEST_OUT_PATH,
+ IntegrationTestUtils.TEST_JOB_DIR,
"images",
"*eastafrica*.shp")
diff --git a/tests/integration/test_suites/survey_test_suite.py b/tests/integration/test_suites/survey_test_suite.py
index 0ab19ee285591b566475db709c9e94e81639f145..6c6804dc06b0dbebef27a2146bda5e7b13332cde 100644
--- a/tests/integration/test_suites/survey_test_suite.py
+++ b/tests/integration/test_suites/survey_test_suite.py
@@ -9,25 +9,21 @@ class BaseSurveyTestSuite:
class SurveyTestSuite(unittest.TestCase):
- TEST_OUT_PATH: str = None
- TEST_START_DATE: str = None
- TEST_JOB_DIR: str = None
-
@abc.abstractmethod
def set_expected_values(self):
pass
def test_standard_run_input_status_success(self):
- status_file_path = os.path.join(BaseSurveyTestSuite.SurveyTestSuite.TEST_OUT_PATH,
- BaseSurveyTestSuite.SurveyTestSuite.TEST_JOB_DIR, "STATUS_SUCCESS")
+ status_file_path = os.path.join(IntegrationTestUtils.TEST_OUT_PATH,
+ IntegrationTestUtils.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_merged_survey_created(self):
- status_file_path = os.path.join(BaseSurveyTestSuite.SurveyTestSuite.TEST_OUT_PATH,
- BaseSurveyTestSuite.SurveyTestSuite.TEST_JOB_DIR,
+ status_file_path = os.path.join(IntegrationTestUtils.TEST_OUT_PATH,
+ IntegrationTestUtils.TEST_JOB_DIR,
"ExportCSV", "Merged_SurveyData.csv")
success_file_exists: bool = os.path.isfile(status_file_path)
file_not_empty: bool = IntegrationTestUtils.check_file_not_empty(status_file_path)