From 11fa76b841b1455fd763fd1df5a582bc427dcb09 Mon Sep 17 00:00:00 2001
From: lb584 <lb584@cam.ac.uk>
Date: Tue, 31 Jan 2023 14:15:27 +0000
Subject: [PATCH] initial commit of full integration tests (wont work)
---
.../integration/full/full_test_deposition.py | 111 ++++++++++++++++++
tests/integration/full/full_test_survey.py | 81 +++++++++++++
2 files changed, 192 insertions(+)
create mode 100644 tests/integration/full/full_test_deposition.py
create mode 100644 tests/integration/full/full_test_survey.py
diff --git a/tests/integration/full/full_test_deposition.py b/tests/integration/full/full_test_deposition.py
new file mode 100644
index 0000000..b4c0be1
--- /dev/null
+++ b/tests/integration/full/full_test_deposition.py
@@ -0,0 +1,111 @@
+import argparse
+import copy
+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 = '20230126'
+ 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_202301260000_202302020000_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_202301260000_202302020000_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(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(27, 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(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('unittest_args', nargs='*')
+
+ _args = parser.parse_args()
+ _config_file: str = _args.config
+ _outdir: str = _args.outdir
+ _email_cred_path: str = _args.email_cred
+
+ IntegrationTestUtils.DEFAULT_CONFIG_FILE_PATH = _config_file
+
+ TestFullDeposition.TEST_OUT_PATH = _outdir
+ IntegrationTestUtils.EMAIL_CRED_PATH = _email_cred_path
+
+ # Now set the sys.argv to the unittest_args (leaving sys.argv[0] alone)
+ sys.argv[1:] = _args.unittest_args
+ unittest.main()
diff --git a/tests/integration/full/full_test_survey.py b/tests/integration/full/full_test_survey.py
new file mode 100644
index 0000000..d9b2c16
--- /dev/null
+++ b/tests/integration/full/full_test_survey.py
@@ -0,0 +1,81 @@
+import argparse
+import copy
+import os
+import sys
+import unittest
+
+from integration.partial.integration_test_utils import IntegrationTestUtils
+
+
+class FullTestSurvey(unittest.TestCase):
+
+ TEST_OUT_PATH: str = "not_set"
+ TEST_START_DATE: str = '20230126'
+ TEST_JOB_DIR: str = "SURVEYDATA_" + TEST_START_DATE
+
+ @classmethod
+ def setUpClass(cls) -> None:
+ FullTestSurvey.write_temp_run_config_file()
+ FullTestSurvey.run_survey_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'] = FullTestSurvey.TEST_OUT_PATH
+ run_dict['WorkspacePath'] = FullTestSurvey.TEST_OUT_PATH
+ # run_dict['Survey']['SkipServerDownload'] = True
+ run_dict['Survey']['SkipServerDownload'] = False
+ # run_dict['ServerName'] = '' # nothing, as local machine
+
+ IntegrationTestUtils.TEMP_CONFIG_FILE_PATH = FullTestSurvey.TEST_OUT_PATH + "temp_config.json"
+ IntegrationTestUtils.write_json_file(run_dict, IntegrationTestUtils.TEMP_CONFIG_FILE_PATH)
+
+ @staticmethod
+ def unpack_dependencies():
+ IntegrationTestUtils.unpack_zip(IntegrationTestUtils.EXAMPLE_SURVEY_FILE_PATH, FullTestSurvey.TEST_OUT_PATH)
+
+ @staticmethod
+ def run_survey_pipeline():
+ component = 'Survey'
+ IntegrationTestUtils.run_external_pipeline(component,
+ FullTestSurvey.TEST_START_DATE,
+ IntegrationTestUtils.EMAIL_CRED_PATH)
+
+
+ def test_standard_run_input_status_success(self):
+ status_file_path = os.path.join(FullTestSurvey.TEST_OUT_PATH, FullTestSurvey.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(FullTestSurvey.TEST_OUT_PATH, FullTestSurvey.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)
+ self.assertTrue(success_file_exists)
+ self.assertTrue(file_not_empty)
+
+
+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('unittest_args', nargs='*')
+
+ _args = parser.parse_args()
+ _config_file: str = _args.config
+ _outdir: str = _args.outdir
+ _email_cred_path: str = _args.email_cred
+
+ IntegrationTestUtils.DEFAULT_CONFIG_FILE_PATH = _config_file
+
+ FullTestSurvey.TEST_OUT_PATH = _outdir
+ IntegrationTestUtils.EMAIL_CRED_PATH = _email_cred_path
+
+ # Now set the sys.argv to the unittest_args (leaving sys.argv[0] alone)
+ sys.argv[1:] = _args.unittest_args
+ unittest.main()
--
GitLab