import sys import unittest from datetime import datetime as ddatetime, datetime from pprint import pprint from web.run import retrieve_amda_netcdf, get_data_for_target, _sta_sto FILE_DATE_FMT = "%Y-%m-%d %H:%M:%S" class BaseTestCase(unittest.TestCase): def setUp(self): # initialise app # self.app = create_app(TestConfig) # update flask context # self.app_context = self.app.app_context() # self.app_context.push() pass def tearDown(self): # if os.path.isfile(self.db_path): # os.remove(self.db_path) # self.app_context.pop() pass def test_always_true(self): self.assertTrue(True) class AmdaTestCase(BaseTestCase): def test_amda_retrieve(self): started_at = ddatetime.strptime("2021-12-17 00:00:00", FILE_DATE_FMT) stopped_at = ddatetime.strptime("2021-12-18 00:00:00", FILE_DATE_FMT) amda_list = retrieve_amda_netcdf("earth", "omni_hour_all", started_at, stopped_at) self.assertEqual(1, len(amda_list)) self.assertIn('amda-irap-omp-eu-ddservice-base-data-omni-hour-omni202107010-nc', amda_list[0]) def test_get_data_for_target(self): target_config = {'type': 'planet', 'slug': 'earth', 'name': 'Earth', 'title': 'Earth', 'orbit': {'models': []}, 'models': { 'om': [{'slug': 'omni_hour_all', 'parameters': {'pdyn': 'RamP'}}, {'slug': 'ace_swepam_real_1h', 'parameters': {'dens': 'Dens', 'vtot': 'Vel', 'temp': 'Temp'}}], 'sa': [{'slug': 'omni_hour_all', 'parameters': {'pdyn': 'RamP'}}, {'slug': 'ace_swepam_real_1h', 'parameters': {'dens': 'Dens', 'vtot': 'Vel', 'temp': 'Temp'}}], 'sb': [{'slug': 'omni_hour_all', 'parameters': {'pdyn': 'RamP'}}, {'slug': 'ace_swepam_real_1h', 'parameters': {'dens': 'Dens', 'vtot': 'Vel', 'temp': 'Temp'}}]}, 'locked': False, 'default': True, 'catalog_layers': {'cmecatalogs': []}} started_at = ddatetime.strptime("2021-12-17 00:00:00", FILE_DATE_FMT) stopped_at = ddatetime.strptime("2021-12-18 00:00:00", FILE_DATE_FMT) model = {'parameters': {'dens': 'Dens', 'temp': 'Temp', 'vtot': 'Vel'}, 'slug': 'ace_swepam_real_1h'} s0, s1 = _sta_sto(model, started_at, stopped_at) all_data = get_data_for_target(target_config, 'om', s0, s1 ) self.assertEqual(1, len(all_data)) def test_sta_sto(self): started_at = ddatetime.strptime("2021-12-17 00:00:00", FILE_DATE_FMT) stopped_at = ddatetime.strptime("2021-12-18 00:00:00", FILE_DATE_FMT) model = {'parameters': {'dens': 'Dens', 'temp': 'Temp', 'vtot': 'Vel'}, 'slug': 'ace_swepam_real_1h'} s0, s1 = _sta_sto(model, started_at, stopped_at) self.assertIsInstance(s0, datetime) self.assertIsInstance(s1, datetime)