[WIP] implement basic CWL scatter semantics...

- count-lines3 scatter just works because it matches Galaxy's semantics without needing additional annotation it seems.

Author: jmchilton

lib/galaxy/tools/cwl/parser.py

@@ -40,6 +40,7 @@
     "EnvVarRequirement",
     "InlineJavascriptRequirement",
     "ShellCommandRequirement",
+    "ScatterFeatureRequirement",
 ]
 
 
@@ -527,16 +528,25 @@ def jsonld_id_to_label(self, id):
         return id.rsplit("#", 1)[-1]
 
     def cwl_input_to_galaxy_step(self, input, i):
-        assert input["type"] == "File"
-        return {
+        input_type = input["type"]
+        input_as_dict = {
             "id": i,
             "label": self.jsonld_id_to_label(input["id"]),
             "position": {"left": 0, "top": 0},
-            "type": "data_input",  # TODO: dispatch on type obviously...
             "annotation": self.cwl_object_to_annotation(input),
             "input_connections": {},  # Should the Galaxy API really require this? - Seems to.
         }
 
+        if input_type == "File":
+            input_as_dict["type"] = "data_input"
+        elif isinstance(input_type, dict) and input_type.get("type") == "array" and input_type.get("items") == "File":
+            input_as_dict["type"] = "data_collection_input"
+            input_as_dict["collection_type"] = "list"
+        else:
+            raise NotImplementedError()
+
+        return input_as_dict
+
     def cwl_object_to_annotation(self, cwl_obj):
         return cwl_obj.get("doc", None)
 

test/api/test_workflows_cwl.py

@@ -57,6 +57,36 @@ def test_count_line2_v1(self):
         """Test simple workflow v1.0/count-lines2-wf.cwl."""
         self._run_count_lines_wf("v1.0/count-lines2-wf.cwl")
 
+    def test_count_lines3_v1(self):
+        load_response = self._load_workflow("v1.0/count-lines3-wf.cwl")
+        self._assert_status_code_is(load_response, 200)
+        workflow = load_response.json()
+        workflow_id = workflow["id"]
+        history_id = self.dataset_populator.new_history()
+        hdca = self.dataset_collection_populator.create_list_in_history( history_id ).json()
+        inputs_map = {
+            "file1": {"src": "hdca", "id": hdca["id"]}
+        }
+        workflow_request = dict(
+            history="hist_id=%s" % history_id,
+            workflow_id=workflow_id,
+            inputs=json.dumps(inputs_map),
+            inputs_by="name",
+        )
+        url = "workflows/%s/invocations" % workflow_id
+        invocation_response = self._post(url, data=workflow_request)
+        self._assert_status_code_is(invocation_response, 200)
+        invocation_id = invocation_response.json()["id"]
+        self.wait_for_invocation_and_jobs(history_id, workflow_id, invocation_id)
+        hdca = self.dataset_populator.get_history_collection_details(history_id, hid=8)
+        assert hdca["collection_type"] == "list"
+        elements = hdca["elements"]
+        assert len(elements) == 3
+        element0 = elements[0]["object"]
+        assert element0["history_content_type"] == "dataset"
+        assert element0["state"] == "ok"
+        assert element0["file_ext"] == "expression.json"
+
     def _run_count_lines_wf(self, wf_path):
         load_response = self._load_workflow(wf_path)
         self._assert_status_code_is(load_response, 200)

test/unit/tools/test_cwl.py

@@ -117,6 +117,20 @@ def test_workflow_embedded_tools_proxy():
         assert exp_step["input_connections"]
 
 
+def test_workflow_scatter():
+    version = "v1.0"
+    proxy = workflow_proxy(_cwl_tool_path("%s/count-lines3-wf.cwl" % version))
+
+    step_proxies = proxy.step_proxies()
+    assert len(step_proxies) == 1
+
+    galaxy_workflow_dict = proxy.to_dict()
+    assert len(galaxy_workflow_dict["steps"]) == 2
+
+    wc_step = galaxy_workflow_dict["steps"][1]
+    assert wc_step["input_connections"]
+
+
 def test_load_proxy_simple():
     cat3 = _cwl_tool_path("draft3/cat3-tool.cwl")
     tool_source = get_tool_source(cat3)