Merge pull request #6807 from jmchilton/gxformat2_030

Support for Enhancements to gxformat2.

lib/galaxy/dependencies/pipfiles/default/Pipfile

@@ -73,6 +73,7 @@ paramiko = "*"
 python-genomespaceclient = "<2.0"
 social_auth_core = {version = "==1.5.0", extras = ['openidconnect']}
 cloudauthz = "<=0.2.0"
+gxformat2 = "*"
 
 [requires]
 python_version = "2.7"

lib/galaxy/dependencies/pipfiles/default/pinned-requirements.txt

@@ -55,6 +55,7 @@ functools32==3.2.3.post2; python_version == '2.7'
 future==0.16.0
 futures==3.2.0; python_version == '2.6' or python_version == '2.7'
 galaxy-sequence-utils==1.1.3
+gxformat2==0.3.0
 h5py==2.8.0
 html5lib==1.0.1
 idna==2.7

lib/galaxy/workflow/modules.py

@@ -78,7 +78,8 @@ def __init__(self, trans, content_id=None, **kwds):
     @classmethod
     def from_dict(Class, trans, d, **kwds):
         module = Class(trans, **kwds)
-        module.recover_state(d.get("tool_state"), **kwds)
+        input_connections = d.get("input_connections", {})
+        module.recover_state(d.get("tool_state"), input_connections=input_connections, **kwds)
         module.label = d.get("label")
         return module
 
@@ -867,8 +868,67 @@ def recover_state(self, state, **kwds):
         super(ToolModule, self).recover_state(state, **kwds)
         if kwds.get("fill_defaults", False) and self.tool:
             self.compute_runtime_state(self.trans, step_updates=None)
+            self.augment_tool_state_for_input_connections(**kwds)
             self.tool.check_and_update_param_values(self.state.inputs, self.trans, workflow_building_mode=True)
 
+    def augment_tool_state_for_input_connections(self, **kwds):
+        """Update tool state to accommodate specified input connections.
+
+        Top-level and conditional inputs will automatically get populated with connected
+        data outputs at runtime, but if there are not enough repeat instances in the tool
+        state - the runtime replacement code will never visit the input elements it needs
+        to in order to connect the data parameters to the tool state. This code then
+        populates the required repeat instances in the tool state in order for these
+        instances to be visited and inputs properly connected at runtime. I believe
+        this should be run before check_and_update_param_values in recover_state so non-data
+        parameters are properly populated with default values. The need to populate
+        defaults is why this is done here instead of at runtime - but this might also
+        be needed at runtime at some point (for workflows installed before their corresponding
+        tools?).
+
+        See the test case test_inputs_to_steps for an example of a workflow test
+        case that exercises this code.
+        """
+
+        # Ensure any repeats defined only by input_connections are populated.
+        input_connections = kwds.get("input_connections", {})
+        expected_replacement_keys = input_connections.keys()
+
+        def augment(expected_replacement_key, inputs, inputs_state):
+            if "|" not in expected_replacement_key:
+                return
+
+            prefix, rest = expected_replacement_key.split("|", 1)
+            if "_" not in prefix:
+                return
+
+            repeat_name, index = prefix.rsplit("_", 1)
+            if not index.isdigit():
+                return
+
+            index = int(index)
+            repeat = self.tool.inputs[repeat_name]
+            if repeat.type != "repeat":
+                return
+
+            if repeat_name not in inputs_states:
+                inputs_states[repeat_name] = []
+
+            repeat_values = inputs_states[repeat_name]
+            repeat_instance_state = None
+            while index >= len(repeat_values):
+                repeat_instance_state = {"__index__": len(repeat_values)}
+                repeat_values.append(repeat_instance_state)
+
+            if repeat_instance_state:
+                # TODO: untest branch - no test case for nested repeats yet...
+                augment(rest, repeat.inputs, repeat_instance_state)
+
+        for expected_replacement_key in expected_replacement_keys:
+            inputs_states = self.state.inputs
+            inputs = self.tool.inputs
+            augment(expected_replacement_key, inputs, inputs_states)
+
     def get_runtime_state(self):
         state = DefaultToolState()
         state.inputs = self.state.inputs

test/api/test_workflows.py

@@ -40,9 +40,8 @@
 steps:
   - tool_id: cat1
     label: first_cat
-    state:
-      input1:
-        $link: outer_input
+    in:
+      input1: outer_input
   - run:
       class: GalaxyWorkflow
       inputs:
@@ -61,8 +60,8 @@
               seed_source_selector: set_seed
               seed: asdf
     label: nested_workflow
-    connect:
-      inner_input: first_cat#out_file1
+    in:
+      inner_input: first_cat/out_file1
   - tool_id: cat1
     label: second_cat
     state:
@@ -83,9 +82,8 @@
 steps:
   - tool_id: cat1
     label: first_cat
-    state:
-      input1:
-        $link: outer_input
+    in:
+      input1: outer_input
   - run:
       class: GalaxyWorkflow
       inputs:
@@ -102,8 +100,8 @@
               seed_source_selector: set_seed
               seed: asdf
     label: nested_workflow
-    connect:
-      inner_input: first_cat#out_file1
+    in:
+      inner_input: first_cat/out_file1
   - tool_id: cat1
     label: second_cat
     state:
@@ -525,17 +523,14 @@ def test_export_editor(self):
     def test_export_editor_collection_type_source(self):
         workflow_id = self._upload_yaml_workflow("""
 class: GalaxyWorkflow
-steps:
-  - label: text_input1
-    type: input_collection
+inputs:
+  - id: text_input1
+    type: collection
     collection_type: "list:paired"
+steps:
   - tool_id: collection_type_source
-    state:
-      input_collect:
-        $link: text_input1
-test_data:
-  text_input1:
-    type: "list:paired"
+    in:
+      input_collect: text_input1
 """)
         downloaded_workflow = self._download_workflow(workflow_id, style="editor")
         steps = downloaded_workflow['steps']
@@ -553,7 +548,7 @@ def test_export_editor_subworkflow_collection_type_source(self):
       class: GalaxyWorkflow
       inputs:
         - id: inner_input
-          type: data_collection
+          type: collection
           collection_type: "list:paired"
       outputs:
         - id: workflow_output
@@ -564,16 +559,12 @@ def test_export_editor_subworkflow_collection_type_source(self):
           collection_type: "list:paired"
         - tool_id: collection_type_source
           label: collection_type_source
-          state:
-            input_collect:
-              $link: inner_input
+          in:
+            input_collect: inner_input
     label: inner_workflow
-    connect:
+    in:
       inner_input: outer_input
-test_data:
-  outer_input:
-    type: "list:paired"
-    """)
+""")
         downloaded_workflow = self._download_workflow(workflow_id, style="editor")
         steps = downloaded_workflow['steps']
         assert len(steps) == 2
@@ -754,19 +745,19 @@ def test_workflow_resume_with_mapped_over_input(self):
         with self.dataset_populator.test_history() as history_id:
             job_summary = self._run_jobs("""
 class: GalaxyWorkflow
+inputs:
+  - id: input_datasets
+    type: collection
 steps:
-  - label: input_datasets
-    type: input_collection
   - label: fail_identifier_1
     tool_id: fail_identifier
     state:
-      input1:
-        $link: input_datasets
       failbool: true
+    in:
+      input1: input_datasets
   - tool_id: identifier_collection
-    state:
-      input1:
-        $link: fail_identifier_1#out_file1
+    in:
+      input1: fail_identifier_1/out_file1
 test_data:
   input_datasets:
     type: list

test/api/test_workflows_from_yaml.py

@@ -129,12 +129,9 @@ def test_inputs_to_steps(self):
 steps:
   - tool_id: cat1
     label: first_cat
-    state:
-      input1:
-        $link: input1
-      queries:
-        - input2:
-            $link: input1
+    connect:
+      input1: input1
+      queries_0|input2: input1
 
 test_data:
   input1: "hello world"
@@ -258,14 +255,12 @@ def test_pause(self):
         $link: test_input
   - label: the_pause
     type: pause
-    connect:
-      input:
-      - first_cat#out_file1
+    in:
+      input: first_cat/out_file1
   - label: second_cat
     tool_id: cat1
-    state:
-      input1:
-        $link: the_pause
+    in:
+      input1: the_pause
 """)
         print(self._get("workflows/%s/download" % workflow_id).json())
 

test/base/populators.py

@@ -16,16 +16,16 @@ def nottest(x):
         return x
 import requests
 import yaml
+from gxformat2 import (
+    convert_and_import_workflow,
+    ImporterGalaxyInterface,
+)
 from pkg_resources import resource_string
 from six import StringIO
 
 from galaxy.tools.verify.test_data import TestDataResolver
 from galaxy.util import unicodify
 from . import api_asserts
-from .workflows_format_2 import (
-    convert_and_import_workflow,
-    ImporterGalaxyInterface,
-)
 
 
 # Simple workflow that takes an input and call cat wrapper on it.