BackendTests.py
3.67 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
import gzip
import os.path
import sys
import unittest
from datetime import datetime as ddatetime, datetime
from pprint import pprint
import numpy.ma.core
from netCDF4 import Dataset
from web.run import retrieve_amda_netcdf, get_data_for_target, _sta_sto, _read_var, default_nc_keys
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)
def test_var_read_nc_float32(self):
SCRIPT_PATH = os.path.dirname(__file__)
PROJECT_DIR = os.path.abspath(os.path.join(SCRIPT_PATH, os.pardir))
local_netc_file = os.path.join(PROJECT_DIR,'tests-resources', 'amda-irap-omp-eu-ddservice-base-data-omni-hour-omni202107010-nc')
cdf_handle = Dataset(local_netc_file, "r", format="NETCDF4")
nc_keys = default_nc_keys.copy()
data_v = _read_var(cdf_handle, nc_keys, 'vtot')
self.assertIsInstance(data_v[0], numpy.float32)