Merge pull request #411 from dcs4cop/forman-xxx-gen2_python_client

xcube generator service CLI and Python API

CHANGES.md

@@ -3,6 +3,11 @@
 * The S3 data store  `xcube.core.store.stores.s3.S3DataStore` now implements the `describe_data()` method. 
   It therefore can also be used as a data store from which data is queried and read.  
 
+* The `xcube gen2` data cube generator tool has been hidden from
+  the set of "official" xcube tools. It is considered as an internal tool 
+  that is subject to change at any time until its interface has stabilized.
+  Please refer to `xcube gen2 --help` for more information.
+
 * Added `coords` property to `DatasetDescriptor` class. 
   The `data_vars` property of the `DatasetDescriptor` class is now a dictionary. 
 

test/cli/test_main.py

@@ -1,26 +1,36 @@
-from test.cli.helpers import CliTest
+import warnings
+
+import pytest
 from click import command
+
 import xcube.cli.main
-from warnings import warn
-import pytest
+from test.cli.helpers import CliTest
 
 
 class ClickMainTest(CliTest):
-
-    WARNING_MESSAGE = 'This is a test warning.'
+    USER_WARNING_MESSAGE = 'This is a user warning.'
+    DEPRECATION_WARNING_MESSAGE = 'This is a deprecation warning.'
+    RUNTIME_WARNING_MESSAGE = 'This is a runtime warning.'
 
     def setUp(self):
         @command(name='test_warnings')
         def test_warnings():
-            warn(self.WARNING_MESSAGE)
+            warnings.warn(self.USER_WARNING_MESSAGE)
+            warnings.warn(self.DEPRECATION_WARNING_MESSAGE, category=DeprecationWarning)
+            warnings.warn(self.RUNTIME_WARNING_MESSAGE, category=RuntimeWarning)
 
         xcube.cli.main.cli.add_command(test_warnings)
 
     def test_warnings_not_issued_by_default(self):
         with pytest.warns(None) as record:
             self.invoke_cli(['test_warnings'])
-        self.assertEqual(0, len(record.list))
+        self.assertEqual(['This is a user warning.'],
+                         list(map(lambda r: str(r.message), record.list)))
 
     def test_warnings_issued_when_enabled(self):
-        with pytest.warns(UserWarning, match=self.WARNING_MESSAGE):
+        with pytest.warns(None) as record:
             self.invoke_cli(['--warnings', 'test_warnings'])
+        self.assertEqual(['This is a user warning.',
+                          'This is a deprecation warning.',
+                          'This is a runtime warning.'],
+                         list(map(lambda r: str(r.message), record.list)))

test/core/gen2/service/test_config.py

@@ -0,0 +1,16 @@
+import unittest
+
+from xcube.core.gen2.service.config import ServiceConfig
+
+
+class ServiceConfigTest(unittest.TestCase):
+
+    def test_from_json_instance(self):
+        json_instance = dict(endpoint_url='https://stage.xcube-gen.brockmann-consult.de/api/v2',
+                             access_token='02945ugjhklojg908ijr023jgbpij202jbv00897v0798v65472')
+        service_config = ServiceConfig.get_schema().from_instance(json_instance)
+        self.assertIsInstance(service_config, ServiceConfig)
+        self.assertEqual('https://stage.xcube-gen.brockmann-consult.de/api/v2/', service_config.endpoint_url)
+        self.assertEqual('02945ugjhklojg908ijr023jgbpij202jbv00897v0798v65472', service_config.access_token)
+        self.assertEqual(None, service_config.client_id)
+        self.assertEqual(None, service_config.client_secret)

test/core/gen2/service/test_generator.py

@@ -0,0 +1,157 @@
+import unittest
+
+import requests_mock
+
+from test.util.test_progress import TestProgressObserver
+from xcube.core.gen2 import CubeGeneratorError
+from xcube.core.gen2.request import CubeGeneratorRequest
+from xcube.core.gen2.service import CubeGeneratorService
+from xcube.core.gen2.service import ServiceConfig
+from xcube.core.gen2.service.response import CubeInfoWithCosts
+from xcube.util.progress import new_progress_observers
+
+
+def result(worked, total_work, failed=False, traceback: str = None):
+    json = {
+        "result": {
+            "cubegen_id": "93",
+            "status": {
+                "failed": True if failed else None,
+                "succeeded": True if worked == total_work else None,
+                "active": 1 if worked != total_work else None,
+            },
+            "progress": {
+                "worked": worked, "total_work": total_work,
+            }
+        }
+    }
+    if traceback:
+        json.update(traceback=traceback)
+    return dict(json=json)
+
+
+class CubeGeneratorServiceTest(unittest.TestCase):
+    ENDPOINT_URL = 'https://xcube-gen.com/api/v2/'
+
+    CUBE_GEN_CONFIG = dict(input_config=dict(store_id='memory',
+                                             data_id='S2L2A'),
+                           cube_config=dict(variable_names=['B01', 'B02', 'B03'],
+                                            crs='WGS84',
+                                            bbox=[12.2, 52.1, 13.9, 54.8],
+                                            spatial_res=0.05,
+                                            time_range=['2018-01-01', None],
+                                            time_period='4D'),
+                           output_config=dict(store_id='memory',
+                                              data_id='CHL'))
+
+    def setUp(self) -> None:
+        self.service = CubeGeneratorService(CubeGeneratorRequest.from_dict(self.CUBE_GEN_CONFIG),
+                                            ServiceConfig(endpoint_url=self.ENDPOINT_URL,
+                                                          client_id='itzibitzispider',
+                                                          client_secret='g3ergd36fd2983457fhjder'),
+                                            progress_period=0,
+                                            verbose=True)
+
+    @requests_mock.Mocker()
+    def test_generate_cube_success(self, m: requests_mock.Mocker):
+        m.post(f'{self.ENDPOINT_URL}oauth/token',
+               json={
+                   "access_token": "4ccsstkn983456jkfde",
+                   "token_type": "bearer"
+               })
+
+        m.put(f'{self.ENDPOINT_URL}cubegens',
+              response_list=[
+                  result(0, 4),
+              ])
+
+        m.get(f'{self.ENDPOINT_URL}cubegens/93',
+              response_list=[
+                  result(1, 4),
+                  result(2, 4),
+                  result(3, 4),
+                  result(4, 4),
+              ])
+
+        observer = TestProgressObserver()
+        with new_progress_observers(observer):
+            self.service.generate_cube()
+
+        self.assertEqual(
+            [
+                ('begin', [('Generating cube', 0.0, False)]),
+                ('update', [('Generating cube', 0.25, False)]),
+                ('update', [('Generating cube', 0.5, False)]),
+                ('update', [('Generating cube', 0.75, False)]),
+                ('end', [('Generating cube', 0.75, True)])
+            ],
+            observer.calls)
+
+    @requests_mock.Mocker()
+    def test_generate_cube_failure(self, m: requests_mock.Mocker):
+        m.post(f'{self.ENDPOINT_URL}oauth/token',
+               json={
+                   "access_token": "4ccsstkn983456jkfde",
+                   "token_type": "bearer"
+               })
+
+        m.put(f'{self.ENDPOINT_URL}cubegens',
+              response_list=[
+                  result(0, 4),
+              ])
+
+        m.get(f'{self.ENDPOINT_URL}cubegens/93',
+              response_list=[
+                  result(1, 4),
+                  result(2, 4, failed=True, traceback='1.that\n2.was\n3.bad'),
+              ])
+
+        observer = TestProgressObserver()
+        with new_progress_observers(observer):
+            with self.assertRaises(CubeGeneratorError) as cm:
+                self.service.generate_cube()
+            self.assertEqual('Cube generation failed', f'{cm.exception}')
+            self.assertEqual('1.that\n2.was\n3.bad', cm.exception.remote_traceback)
+
+        print(observer.calls)
+        self.assertEqual(
+            [
+                ('begin', [('Generating cube', 0.0, False)]),
+                ('update', [('Generating cube', 0.25, False)]),
+                ('end', [('Generating cube', 0.25, True)])
+            ],
+            observer.calls)
+
+    @requests_mock.Mocker()
+    def test_get_cube_info(self, m: requests_mock.Mocker):
+        m.post(f'{self.ENDPOINT_URL}oauth/token',
+               json={
+                   "access_token": "4ccsstkn983456jkfde",
+                   "token_type": "bearer"
+               })
+
+        m.post(f'{self.ENDPOINT_URL}cubegens/info',
+               json=dict(dims=dict(time=10 * 365, lat=720, lon=1440),
+                         chunks=dict(time=10, lat=720, lon=1440),
+                         data_vars=dict(CHL=dict(long_name='chlorophyll_concentration',
+                                                 units='mg/m^-1')),
+                         cost_info=dict(punits_input=300,
+                                        punits_output=400,
+                                        punits_combined=500)))
+
+        cube_info = self.service.get_cube_info()
+        self.assertIsInstance(cube_info, CubeInfoWithCosts)
+        self.assertEqual(dict(time=10 * 365, lat=720, lon=1440),
+                         cube_info.dims)
+        self.assertEqual(dict(time=10, lat=720, lon=1440),
+                         cube_info.chunks)
+        self.assertEqual(dict(CHL=dict(long_name='chlorophyll_concentration',
+                                       units='mg/m^-1')),
+                         cube_info.data_vars)
+        self.assertEqual(
+            {
+                "punits_input": 300,
+                "punits_output": 400,
+                "punits_combined": 500
+            },
+            cube_info.cost_info.additional_properties)

test/core/gen2/test_config.py

@@ -1,7 +1,7 @@
 import unittest
 
+from xcube.core.gen2.config import CallbackConfig
 from xcube.core.gen2.config import CubeConfig
-from xcube.core.gen2.config import CubeGeneratorConfig
 from xcube.core.gen2.config import InputConfig
 from xcube.core.gen2.config import OutputConfig
 
@@ -59,46 +59,18 @@ def test_to_dict(self):
         self.assertEqual(json_instance, cube_config.to_dict())
 
 
-class GenConfigTest(unittest.TestCase):
-
-    def test_from_dict(self):
-        request_dict = dict(input_configs=[dict(store_id='memory',
-                                                data_id='S2L2A')],
-                            cube_config=dict(variable_names=['B01', 'B02'],
-                                             crs='WGS84',
-                                             bbox=[12.2, 52.1, 13.9, 54.8],
-                                             spatial_res=0.05,
-                                             time_range=['2018-01-01', None],
-                                             time_period='4D'),
-                            output_config=dict(store_id='memory',
-                                               data_id='CHL'))
-        gen_config = CubeGeneratorConfig.from_dict(request_dict)
-        self.assertIsInstance(gen_config, CubeGeneratorConfig)
-        self.assertEqual(1, len(gen_config.input_configs))
-        self.assertIsInstance(gen_config.input_configs[0], InputConfig)
-        self.assertEqual('memory', gen_config.input_configs[0].store_id)
-        self.assertEqual('S2L2A', gen_config.input_configs[0].data_id)
-        self.assertIsInstance(gen_config.output_config, OutputConfig)
-        self.assertEqual('memory', gen_config.output_config.store_id)
-        self.assertEqual('CHL', gen_config.output_config.data_id)
-        self.assertIsInstance(gen_config.cube_config, CubeConfig)
-        self.assertEqual(('B01', 'B02'), gen_config.cube_config.variable_names)
-        self.assertEqual('WGS84', gen_config.cube_config.crs)
-        self.assertEqual((12.2, 52.1, 13.9, 54.8), gen_config.cube_config.bbox)
-        self.assertEqual(0.05, gen_config.cube_config.spatial_res)
-        self.assertEqual(('2018-01-01', None), gen_config.cube_config.time_range)
-        self.assertEqual('4D', gen_config.cube_config.time_period)
+class CallbackConfigTest(unittest.TestCase):
 
     def test_to_dict(self):
-        request_dict = dict(input_config=dict(store_id='memory',
-                                              data_id='S2L2A'),
-                            cube_config=dict(variable_names=['B01', 'B02'],
-                                             crs='WGS84',
-                                             bbox=[12.2, 52.1, 13.9, 54.8],
-                                             spatial_res=0.05,
-                                             time_range=['2018-01-01', None],
-                                             time_period='4D'),
-                            output_config=dict(store_id='memory',
-                                               data_id='CHL'))
-        request = CubeGeneratorConfig.from_dict(request_dict)
-        self.assertEqual(request_dict, request.to_dict())
+        with self.assertRaises(ValueError) as e:
+            CallbackConfig()
+        self.assertEqual('Both, api_uri and access_token must be given', str(e.exception))
+
+        expected = {
+            "api_uri": 'https://bla.com',
+            "access_token": 'dfasovjdasoövjidfs'
+        }
+        callback = CallbackConfig(**expected)
+        res = callback.to_dict()
+
+        self.assertDictEqual(expected, res)

test/core/gen2/test_generator.py

@@ -7,11 +7,13 @@
 
 from xcube.core.dsio import rimraf
 from xcube.core.gen2.generator import CubeGenerator
+from xcube.core.gen2.generator import LocalCubeGenerator
+from xcube.core.gen2.request import CubeGeneratorRequest
 from xcube.core.new import new_cube
 from xcube.core.store.stores.memory import MemoryDataStore
 
 
-class CubeGeneratorTest(unittest.TestCase):
+class LocalCubeGeneratorTest(unittest.TestCase):
     REQUEST = dict(input_config=dict(store_id='memory',
                                      data_id='S2L2A'),
                    cube_config=dict(variable_names=['B01', 'B02', 'B03'],
@@ -27,9 +29,9 @@ class CubeGeneratorTest(unittest.TestCase):
 
     def setUp(self) -> None:
         with open('_request.json', 'w') as fp:
-            json.dump(CubeGeneratorTest.REQUEST, fp)
+            json.dump(self.REQUEST, fp)
         with open('_request.yaml', 'w') as fp:
-            yaml.dump(CubeGeneratorTest.REQUEST, fp)
+            yaml.dump(self.REQUEST, fp)
         self.saved_cube_memory = MemoryDataStore.replace_global_data_dict(
             {'S2L2A': new_cube(variables=dict(B01=0.1, B02=0.2, B03=0.3))}
         )
@@ -39,17 +41,24 @@ def tearDown(self) -> None:
         rimraf('_request.yaml')
         MemoryDataStore.replace_global_data_dict(self.saved_cube_memory)
 
+    @requests_mock.Mocker()
+    def test_dict(self, m):
+        m.put('https://xcube-gen.test/api/v1/jobs/tomtom/iamajob/callback', json={})
+        LocalCubeGenerator(CubeGeneratorRequest.from_dict(self.REQUEST), verbose=True).generate_cube()
+        self.assertIsInstance(MemoryDataStore.get_global_data_dict().get('CHL'),
+                              xr.Dataset)
+
     @requests_mock.Mocker()
     def test_json(self, m):
         m.put('https://xcube-gen.test/api/v1/jobs/tomtom/iamajob/callback', json={})
-        CubeGenerator.from_file('_request.json', verbose=True).run()
+        CubeGenerator.from_file('_request.json', verbose=True).generate_cube()
         self.assertIsInstance(MemoryDataStore.get_global_data_dict().get('CHL'),
                               xr.Dataset)
 
     @requests_mock.Mocker()
     def test_yaml(self, m):
         m.put('https://xcube-gen.test/api/v1/jobs/tomtom/iamajob/callback', json={})
-        CubeGenerator.from_file('_request.yaml', verbose=True).run()
+        CubeGenerator.from_file('_request.yaml', verbose=True).generate_cube()
         self.assertIsInstance(MemoryDataStore.get_global_data_dict().get('CHL'),
                               xr.Dataset)
 

test/core/gen2/test_progress.py

@@ -3,7 +3,7 @@
 
 import requests_mock
 
-from xcube.core.gen2.config import CubeGeneratorConfig
+from xcube.core.gen2.request import CubeGeneratorRequest
 from xcube.core.gen2.progress import ApiProgressCallbackObserver
 from xcube.core.gen2.progress import TerminalProgressCallbackObserver
 from xcube.core.gen2.progress import _ThreadedProgressObserver
@@ -26,7 +26,7 @@ class TestThreadedProgressObservers(unittest.TestCase):
                                         access_token='dfsvdfsv'))
 
     def setUp(self) -> None:
-        self._request = CubeGeneratorConfig.from_dict(self.REQUEST)
+        self._request = CubeGeneratorRequest.from_dict(self.REQUEST)
         self._callback_config = self._request.callback_config
 
     @requests_mock.Mocker()