Merge pull request #169 from roocs/fix-intake-filemapper

use FileMapper for search results

rook/director/director.py

@@ -4,6 +4,7 @@
 from daops.utils.normalise import ResultSet
 from pywps.app.exceptions import ProcessError
 from roocs_utils.project_utils import get_project_name
+from roocs_utils.utils.file_utils import FileMapper
 
 from rook import CONFIG
 from rook.exceptions import InvalidCollection
@@ -160,7 +161,7 @@ def process(self, runner):
             if self.search_result:
                 self.inputs["collection"] = []
                 for ds_id, file_uris in self.search_result.files().items():
-                    self.inputs["collection"].extend(file_uris)
+                    self.inputs["collection"].append(FileMapper(file_uris))
             try:
                 file_uris = runner(self.inputs)
             except Exception as e:

rook/workflow.py

@@ -131,15 +131,13 @@ def _run_step(self, step_id, step, inputs=None):
         if inputs:
             step["in"].update(inputs)
         if "subset" == step["run"]:
+            collection = step["in"]["collection"]
             result = self.subset_op.call(step["in"])
-            self.prov.add_operator(
-                step_id, step["in"], step["in"]["collection"], result
-            )
+            self.prov.add_operator(step_id, step["in"], collection, result)
         elif "average" == step["run"]:
+            collection = step["in"]["collection"]
             result = self.average_op.call(step["in"])
-            self.prov.add_operator(
-                step_id, step["in"], step["in"]["collection"], result
-            )
+            self.prov.add_operator(step_id, step["in"], collection, result)
         elif "diff" == step["run"]:
             result = self.diff_op.call(step["in"])
             self.prov.add_operator(step_id, step["in"], ["missing"], result)

setup.cfg

@@ -23,27 +23,27 @@ search = "version": "{current_version}",
 replace = "version": "{new_version}",
 
 [tool:pytest]
-addopts = 
+addopts =
 	--strict
 	--tb=native
 	tests/
 python_files = test_*.py
-markers = 
+markers =
 	online: mark test to need internet connection
 	slow: mark test to be slow
 	smoke: mark test as a smoke/sanity test
 
 [flake8]
 max-line-length = 120
-exclude = 
+exclude =
 	.git,
 	__pycache__,
 	docs/source/conf.py,
 	build,
 	dist,
 	src,
 	mini-esgf-data,
-ignore = 
+ignore =
 	F401
 	E402
 	W503

tests/smoke/conftest.py

@@ -30,12 +30,13 @@ def execute(self, identifier, inputs):
         outputs = [("output", True, None)]
         execution = self.wps.execute(identifier, inputs, output=outputs)
         monitorExecution(execution)
+        print(execution.errors)
         assert execution.isSucceded() is True
         assert len(execution.processOutputs) > 0
         ml_url = execution.processOutputs[0].reference
         xml = requests.get(ml_url).text
-        url = parse_metalink(xml)
-        return url
+        urls = parse_metalink(xml)
+        return urls
 
 
 @pytest.fixture

tests/smoke/test_smoke_checks.py

@@ -78,9 +78,10 @@ def test_smoke_execute_subset(wps, tmp_path):
         ("collection", C3S_CMIP6_MON_COLLECTION),
         ("time", "2020-01-01/2020-12-30"),
     ]
-    url = wps.execute("subset", inputs)
-    assert "rlds_Amon_INM-CM5-0_ssp245_r1i1p1f1_gr1_20200116-20201216.nc" in url
-    ds = open_dataset(url, tmp_path)
+    urls = wps.execute("subset", inputs)
+    assert len(urls) == 1
+    assert "rlds_Amon_INM-CM5-0_ssp245_r1i1p1f1_gr1_20200116-20201216.nc" in urls[0]
+    ds = open_dataset(urls[0], tmp_path)
     assert "rlds" in ds.variables
 
 
@@ -90,14 +91,17 @@ def test_smoke_execute_subset_original_files(wps):
         ("time", "2020-01-01/2020-12-30"),
         ("original_files", "1"),
     ]
-    url = wps.execute("subset", inputs)
-    assert "data.mips.copernicus-climate.eu" in url
+    urls = wps.execute("subset", inputs)
+    assert len(urls) == 1
+    assert "data.mips.copernicus-climate.eu" in urls[0]
 
 
 def test_smoke_execute_subset_collection_only(wps):
     inputs = [("collection", C3S_CMIP6_DAY_COLLECTION)]
-    url = wps.execute("subset", inputs)
-    assert "data.mips.copernicus-climate.eu" in url
+    urls = wps.execute("subset", inputs)
+    print(urls)
+    assert len(urls) == 2
+    assert "data.mips.copernicus-climate.eu" in urls[0]
 
 
 def test_smoke_execute_subset_time_and_area_cross_meridian(wps):
@@ -106,19 +110,30 @@ def test_smoke_execute_subset_time_and_area_cross_meridian(wps):
         ("time", "2020-01-01/2020-12-30"),
         ("area", "-50,-50,50,50"),
     ]
-    url = wps.execute("subset", inputs)
-    assert "rlds_Amon_INM-CM5-0_ssp245_r1i1p1f1_gr1_20200116-20201216.nc" in url
+    urls = wps.execute("subset", inputs)
+    assert len(urls) == 1
+    assert "rlds_Amon_INM-CM5-0_ssp245_r1i1p1f1_gr1_20200116-20201216.nc" in urls[0]
 
 
-def test_smoke_execute_average_time(wps):
+def test_smoke_execute_c3s_cmip6_mon_average_time(wps):
     inputs = [("collection", C3S_CMIP6_MON_COLLECTION), ("dims", "time")]
-    url = wps.execute("average", inputs)
-    assert "rlds_Amon_INM-CM5-0_ssp245_r1i1p1f1_gr1_avg-t.nc" in url
+    urls = wps.execute("average", inputs)
+    assert len(urls) == 1
+    assert "rlds_Amon_INM-CM5-0_ssp245_r1i1p1f1_gr1_avg-t.nc" in urls[0]
+
+
+def test_smoke_execute_c3s_cmip6_day_average_time(wps):
+    inputs = [("collection", C3S_CMIP6_DAY_COLLECTION), ("dims", "time")]
+    urls = wps.execute("average", inputs)
+    print(urls)
+    assert len(urls) == 1
+    assert "tas_day_HadGEM3-GC31-LL_ssp245_r1i1p1f3_gn_avg-t.nc" in urls[0]
 
 
 def test_smoke_execute_orchestrate(wps):
     inputs = [
         ("workflow", ComplexDataInput(WF_SUBSET_AVERAGE)),
     ]
-    url = wps.execute("orchestrate", inputs)
-    assert "rlds_Amon_INM-CM5-0_ssp245_r1i1p1f1_gr1_avg-t.nc" in url
+    urls = wps.execute("orchestrate", inputs)
+    assert len(urls) == 1
+    assert "rlds_Amon_INM-CM5-0_ssp245_r1i1p1f1_gr1_avg-t.nc" in urls[0]

tests/smoke/utils.py

@@ -23,5 +23,5 @@ def open_dataset(url, tmp_path):
 def parse_metalink(xml):
     xml_ = xml.replace(' xmlns="', ' xmlnamespace="')
     tree = etree.fromstring(xml_.encode())
-    url = tree.xpath("//metaurl")[0].text
-    return url
+    urls = [m.text for m in tree.xpath("//metaurl")]
+    return urls