import copy import os import unittest from integration.partial.integration_test_utils import IntegrationTestUtils from integration.test_suites.depo_test_suite import BaseDepoTestSuite class TestDeposition(BaseDepoTestSuite.DepoTestSuite): @classmethod def setUpClass(cls) -> None: """ This is overridden in the class setup function to ensure that the value is not set from a previous run when being run in a TestSuite """ IntegrationTestUtils.TEST_OUT_PATH = None def set_expected_values(self): IntegrationTestUtils.TEST_START_DATE = "20221001" IntegrationTestUtils.TEST_JOB_DIR = "DEPOSITION_" + IntegrationTestUtils.TEST_START_DATE self.EA_CSV_COUNT = 9 self.ETH_CSV_COUNT = 27 self.EA_PNG_COUNT = 3 self.ETH_PNG_COUNT = 6 def setUp(self) -> None: self.set_expected_values() if IntegrationTestUtils.TEST_OUT_PATH is None or not os.path.isdir(IntegrationTestUtils.TEST_OUT_PATH): TestDeposition.write_temp_run_config_file() # TestDeposition.run_depo_pipeline() else: print(f"output in {IntegrationTestUtils.TEST_OUT_PATH} already written, skipping rerun") @staticmethod def write_temp_run_config_file(): nowstring: str = IntegrationTestUtils.get_now_string() prefix: str = "temp_depo_" + nowstring # prefix: str = "temp_depo" 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) 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['ServerName'] = '' # nothing, as local machine full_server_path = os.path.abspath(IntegrationTestUtils.TEST_ASSETS_PATH) run_dict['Deposition']['ServerPathTemplate'] = full_server_path # 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 run_depo_pipeline(): component = 'Deposition' IntegrationTestUtils.run_unittest_pipeline(component, IntegrationTestUtils.TEST_START_DATE) if __name__ == '__main__': unittest.main()