Merge pull request #1216 from jmchilton/workflow_tweaks_1

More small workflow tweaks.

lib/galaxy/managers/workflows.py

@@ -507,8 +507,16 @@ def callback( input, value, prefixed_name, prefixed_label ):
             input_conn_dict = {}
             unique_input_names = set( [conn.input_name for conn in input_connections] )
             for input_name in unique_input_names:
-                input_conn_dict[ input_name ] = \
-                    [ dict( id=conn.output_step.order_index, output_name=conn.output_name ) for conn in input_connections if conn.input_name == input_name ]
+                input_conn_dicts = []
+                for conn in input_connections:
+                    if conn.input_name != input_name:
+                        continue
+                    input_conn = dict(
+                        id=conn.output_step.order_index,
+                        output_name=conn.output_name
+                    )
+                    input_conn_dicts.append(input_conn)
+                input_conn_dict[ input_name ] = input_conn_dicts
 
             # Preserve backward compatability. Previously Galaxy
             # assumed input connections would be dictionaries not
@@ -537,22 +545,17 @@ def _workflow_to_dict_instance(self, stored):
         item['url'] = url_for('workflow', id=item['id'])
         item['owner'] = stored.user.username
         inputs = {}
-        for step in workflow.steps:
+        for step in workflow.input_steps:
             step_type = step.type
-            if step_type in ['data_input', 'data_collection_input']:
-                if step.tool_inputs and "name" in step.tool_inputs:
-                    label = step.tool_inputs['name']
-                elif step_type == "data_input":
-                    label = "Input Dataset"
-                elif step_type == "data_collection_input":
-                    label = "Input Dataset Collection"
-                else:
-                    raise ValueError("Invalid step_type %s" % step_type)
-                inputs[step.id] = {'label': label, 'value': ""}
+            if step.tool_inputs and "name" in step.tool_inputs:
+                label = step.tool_inputs['name']
+            elif step_type == "data_input":
+                label = "Input Dataset"
+            elif step_type == "data_collection_input":
+                label = "Input Dataset Collection"
             else:
-                pass
-                # Eventually, allow regular tool parameters to be inserted and modified at runtime.
-                # p = step.get_required_parameters()
+                raise ValueError("Invalid/unknown input step_type %s" % step_type)
+            inputs[step.id] = {'label': label, 'value': ""}
         item['inputs'] = inputs
         item['annotation'] = self.get_item_annotation_str( sa_session, stored.user, stored )
         steps = {}

lib/galaxy/model/__init__.py

@@ -3471,6 +3471,7 @@ class Workflow( object, Dictifiable ):
 
     dict_collection_visible_keys = ( 'name', 'has_cycles', 'has_errors' )
     dict_element_visible_keys = ( 'name', 'has_cycles', 'has_errors' )
+    input_step_types = ['data_input', 'data_collection_input']
 
     def __init__( self, uuid=None ):
         self.user = None
@@ -3497,6 +3498,32 @@ def to_dict( self, view='collection', value_mapper=None):
         rval['uuid'] = ( lambda uuid: str( uuid ) if uuid else None )( self.uuid )
         return rval
 
+    @property
+    def steps_by_id( self ):
+        steps = {}
+        for step in self.steps:
+            step_id = step.id
+            steps[ step_id ] = step
+        return steps
+
+    def step_by_index(self, order_index):
+        for step in self.steps:
+            if order_index == step.order_index:
+                return step
+        raise KeyError("Workflow has no step with order_index '%s'" % order_index)
+
+    @property
+    def input_steps(self):
+        for step in self.steps:
+            if step.type in Workflow.input_step_types:
+                yield step
+
+    @property
+    def workflow_outputs(self):
+        for step in self.steps:
+            for workflow_output in step.workflow_outputs:
+                yield workflow_output
+
 
 class WorkflowStep( object ):
 

lib/galaxy/workflow/run.py

@@ -231,12 +231,17 @@ def remaining_steps(self):
         remaining_steps = []
         step_invocations_by_id = self.workflow_invocation.step_invocations_by_step_id()
         for step in steps:
+            step_id = step.id
             if not hasattr( step, 'module' ):
                 self.module_injector.inject( step )
-                runtime_state = step_states[ step.id ].value
+                if step_id not in step_states:
+                    template = "Workflow invocation [%s] has no step state for step id [%s]. States are [%s]."
+                    message = template % (self.workflow_invocation.id, step_id, step_states)
+                    raise Exception(message)
+                runtime_state = step_states[ step_id ].value
                 step.state = step.module.recover_runtime_state( runtime_state )
 
-            invocation_steps = step_invocations_by_id.get( step.id, None )
+            invocation_steps = step_invocations_by_id.get( step_id, None )
             if invocation_steps:
                 self._recover_mapping( step, invocation_steps )
             else:
@@ -263,7 +268,12 @@ def replacement_for_tool_input( self, step, input, prefixed_name ):
         return replacement
 
     def replacement_for_connection( self, connection ):
-        step_outputs = self.outputs[ connection.output_step.id ]
+        output_step_id = connection.output_step.id
+        if output_step_id not in self.outputs:
+            template = "No outputs found for step id %s, outputs are %s"
+            message = template % (output_step_id, self.outputs)
+            raise Exception(message)
+        step_outputs = self.outputs[ output_step_id ]
         if step_outputs is STEP_OUTPUT_DELAYED:
             raise modules.DelayedWorkflowEvaluation()
         output_name = connection.output_name

test/unit/workflows/test_workflow_progress.py

@@ -43,7 +43,7 @@ def setUp(self):
         self.progress = {}
 
     def _setup_workflow(self, workflow_yaml):
-        workflow = yaml_to_model(TEST_WORKFLOW_YAML)
+        workflow = yaml_to_model(workflow_yaml)
         self.invocation.workflow = workflow
 
     def _new_workflow_progress( self ):