Merge branch 'UnitTesting' into dev

hitier
2 parents c074c029 aca1013e
Showing 5 changed files with 201 additions and 107 deletions Show diff stats
DEV.md
pytest.ini
tests/BackendTests.py
tests/FrontEndTests.py
web/run.py
@@ -5,4 +5,15 @@
 * go to config.yml
     1. add new entry to 'inputs' section
     2. inside each target, add entry in the 'models' sub-section
-* possible to set default input in the defaults.input_slug parameter
 \ No newline at end of file
+* possible to set default input in the defaults.input_slug parameter
+
+## Testing
+
+    # basic
+    PYTHONPATH=. pytest
+
+    # debug messages enabled
+    DEBUG=true PYTHONPATH=. pytest
+
+    # output to stderr activated
+    DEBUG=true PYTHONPATH=. pytest -rP 
@@ -0,0 +1,6 @@
+[pytest]
+testpaths = tests
+python_files = *Tests.py
+log_level = INFO
+log_cli = True
+console_output_style = classic
@@ -0,0 +1,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(25, 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)
-import urllib
-
-from selenium import webdriver
-from flask_testing import LiveServerTestCase
-from tryflask import app
-
-_SERVER_PORT = 8943
-_ROOT_URL = "http://localhost:{}".format(_SERVER_PORT)
-
-
-class BaseTestCase(LiveServerTestCase):
-    def create_app(self):
-        # # remove logging lines on test output
-        import logging
-        log = logging.getLogger('werkzeug')
-        log.setLevel(logging.INFO)
-        log.disabled = True
-        # os.environ['WERKZEUG_RUN_MAIN'] = 'true'
-
-        # pass in test configurations
-        app.config.update(
-            # Change the port that the liveserver listens on as we dont want to conflict with running:5000
-            LIVESERVER_PORT=_SERVER_PORT
-        )
-
-        self.app_context = app.app_context()
-        self.app_context.push()
-        return app
-
-    def setUp(self):
-        # os.environ["PYTHONTRACEMALLOC"] = '1'
-
-        self.driver = self.create_chrome_driver()
-
-    def tearDown(self):
-        self.app_context.pop()
-        self.driver.close()
-        self.driver.quit()
-
-    def create_chrome_driver(self):
-        """
-        Create then return the chrome driver.
-
-        :return: the chrome driver.
-        """
-        from selenium.webdriver.chrome.options import Options
-        options = Options()
-        options.add_argument('--headless')
-
-        return webdriver.Chrome(options=options)
-
-
-class AccessTestCase(BaseTestCase):
-    def test_ping(self):
-        self.assertEqual(_ROOT_URL, self.get_server_url())
-
-    def test_selenium_access(self):
-        # open browser on servers adress
-        self.driver.get(self.get_server_url())
-        # L'adresse dans l'url doit être celle que l'on attend.
-        self.assertEqual(_ROOT_URL+'/', self.driver.current_url)
-
-    def test_server_is_up_and_running(self):
-        response = urllib.request.urlopen(self.get_server_url())
-        self.assertEqual(200, response.code)
+# import urllib
+#
+# from selenium import webdriver
+# from flask_testing import LiveServerTestCase
+#
+# _SERVER_PORT = 8943
+# _ROOT_URL = "http://localhost:{}".format(_SERVER_PORT)
+#
+#
+# class BaseTestCase(LiveServerTestCase):
+#     def create_app(self):
+#         # # remove logging lines on test output
+#         import logging
+#         log = logging.getLogger('werkzeug')
+#         log.setLevel(logging.INFO)
+#         log.disabled = True
+#         # os.environ['WERKZEUG_RUN_MAIN'] = 'true'
+#
+#         # pass in test configurations
+#         app.config.update(
+#             # Change the port that the liveserver listens on as we dont want to conflict with running:5000
+#             LIVESERVER_PORT=_SERVER_PORT
+#         )
+#
+#         self.app_context = app.app_context()
+#         self.app_context.push()
+#         return app
+#
+#     def setUp(self):
+#         # os.environ["PYTHONTRACEMALLOC"] = '1'
+#
+#         self.driver = self.create_chrome_driver()
+#
+#     def tearDown(self):
+#         self.app_context.pop()
+#         self.driver.close()
+#         self.driver.quit()
+#
+#     def create_chrome_driver(self):
+#         """
+#         Create then return the chrome driver.
+#
+#         :return: the chrome driver.
+#         """
+#         from selenium.webdriver.chrome.options import Options
+#         options = Options()
+#         options.add_argument('--headless')
+#
+#         return webdriver.Chrome(options=options)
+#
+#
+# class AccessTestCase(BaseTestCase):
+#     def test_ping(self):
+#         self.assertEqual(_ROOT_URL, self.get_server_url())
+#
+#     def test_selenium_access(self):
+#         # open browser on servers adress
+#         self.driver.get(self.get_server_url())
+#         # L'adresse dans l'url doit être celle que l'on attend.
+#         self.assertEqual(_ROOT_URL+'/', self.driver.current_url)
+#
+#     def test_server_is_up_and_running(self):
+#         response = urllib.request.urlopen(self.get_server_url())
+#         self.assertEqual(200, response.code)
@@ -11,7 +11,8 @@ import time
 import urllib.request as urllib_request
 import requests
 import re  # regex
-import numpy
+import numpy as np
+import sys
  
 from csv import writer as csv_writer, DictWriter as csv_dict_writer
 from math import sqrt, isnan
@@ -79,7 +80,6 @@ logHandler.setFormatter(logging.Formatter(
 ))
 log.addHandler(logHandler)
  
-
 # HARDCODED CONFIGURATION #####################################################
  
 ASTRONOMICAL_UNIT_IN_KM = 1.496e8
@@ -333,6 +333,40 @@ def datetime_from_list(time_list):
         "%Y%j%H%M%S%f"
     )
  
+# Override these using the model configuration in config.yml
+default_nc_keys = {
+    'hee': 'HEE',
+    'vtot': 'V',
+    'magn': 'B',
+    'temp': 'T',
+    'dens': 'N',
+    'pdyn': 'P_dyn',
+    'atse': 'Delta_angle',
+}
+
+def _sta_sto(_cnf, _sta, _sto):
+    if 'started_at' in _cnf:
+        _s0 = datetime.datetime.strptime(_cnf['started_at'], FILE_DATE_FMT)
+        _s0 = max(_s0, _sta)
+    else:
+        _s0 = _sta
+    if 'stopped_at' in _cnf:
+        _s1 = datetime.datetime.strptime(_cnf['stopped_at'], FILE_DATE_FMT)
+        _s1 = min(_s1, _sto)
+    else:
+        _s1 = _sto
+    return _s0, _s1
+
+
+def _read_var(_nc, _keys, _key, mandatory=False):
+    try:
+        return _nc.variables[_keys[_key]][:]
+    except KeyError:
+        pass
+    if mandatory:
+        raise Exception("No variable '%s' found in NetCDF." % _keys[_key])
+    return [None] * len(_nc.variables['Time'])  # slow -- use numpy!
+
  
 def get_local_filename(url):
     """
@@ -483,10 +517,6 @@ ORDER BY time_min, granule_gid
                 'external_link': row[3].text,
             })
  
-        # print("Auroral emission query")
-        # print(query)
-        # print("Auroral emission rows")
-        # print(rows)
  
         return rows
     except Exception as e:
@@ -654,38 +684,6 @@ def get_data_for_target(target_config, input_slug,
         # abort(500, "Invalid orbit configuration for '%s' : %s"
         #       % (target_config['slug'], str(e)))
  
-    def _sta_sto(_cnf, _sta, _sto):
-        if 'started_at' in _cnf:
-            _s0 = datetime.datetime.strptime(_cnf['started_at'], FILE_DATE_FMT)
-            _s0 = max(_s0, _sta)
-        else:
-            _s0 = _sta
-        if 'stopped_at' in _cnf:
-            _s1 = datetime.datetime.strptime(_cnf['stopped_at'], FILE_DATE_FMT)
-            _s1 = min(_s1, _sto)
-        else:
-            _s1 = _sto
-        return _s0, _s1
-
-    def _read_var(_nc, _keys, _key, mandatory=False):
-        try:
-            return _nc.variables[_keys[_key]]
-        except KeyError:
-            pass
-        if mandatory:
-            raise Exception("No variable '%s' found in NetCDF." % _keys[_key])
-        return [None] * len(_nc.variables['Time'])  # slow -- use numpy!
-
-    # Override these using the model configuration in config.yml
-    default_nc_keys = {
-        'hee': 'HEE',
-        'vtot': 'V',
-        'magn': 'B',
-        'temp': 'T',
-        'dens': 'N',
-        'pdyn': 'P_dyn',
-        'atse': 'Delta_angle',
-    }
  
     precision = "%Y-%m-%dT%H"  # model and orbits times are only equal-ish
     orbit_data = {}  # keys are datetime as str, values arrays of XY
@@ -914,7 +912,6 @@ def generate_csv_file_if_needed(target_slug, input_slug,
             if isfile(local_csv_file):
                 log.warning("Removing failed CSV '%s'..." % local_csv_file)
                 removefile(local_csv_file)
-            # pprint(e)
             abort(500, "Failed creating CSV '%s' : %s" % (filename, e))
  
  
@@ -1523,7 +1520,6 @@ def download_targets_cdf(targets, inp, started_at, stopped_at):
  
                 for param in params:
                     k = "%s_%s" % (target_slug, param)
-                    # print("PARAM %s" % k)
                     values = []
                     i = available_params.index(param)
                     has_nones = False
@@ -1603,7 +1599,6 @@ def download_auroral_catalog_csv(target):
     # Filter the emissions
     def _keep_emission(emission):
         ok = thumbnail_url_filter.match(emission['thumbnail_url'])
-        # print("ok", ok, emission['thumbnail_url'])
         return bool(ok)
  
     emissions = [e for e in emissions if _keep_emission(e)]