Merge pull request #6850 from jmchilton/workflows_track_step_inputs

Track workflow step input definitions in our model.

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

@@ -56,7 +56,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.7.1
+gxformat2==0.8.0
 h5py==2.8.0
 idna==2.7
 ipaddress==1.0.22 ; python_version < '3.3'

lib/galaxy/managers/workflows.py

@@ -771,6 +771,14 @@ class WorkflowContentsManager(UsesAnnotations):
                 for output in module.get_data_outputs():
                     step_dict['outputs'].append({'name': output['name'], 'type': output['extensions'][0]})
 
+            step_in = {}
+            for step_input in step.inputs:
+                if step_input.default_value_set:
+                    step_in[step_input.name] = {"default": step_input.default_value}
+
+            if step_in:
+                step_dict["in"] = step_in
+
             # Connections
             input_connections = step.input_connections
             if step.type is None or step.type == 'tool':
@@ -998,6 +1006,16 @@ class WorkflowContentsManager(UsesAnnotations):
                     label=label,
                 )
                 trans.sa_session.add(m)
+
+        if "in" in step_dict:
+            for input_name, input_dict in step_dict["in"].items():
+                step_input = step.get_or_add_input(input_name)
+                NO_DEFAULT_DEFINED = object()
+                default = input_dict.get("default", NO_DEFAULT_DEFINED)
+                if default is not NO_DEFAULT_DEFINED:
+                    step_input.default_value = default
+                    step_input.default_value_set = True
+
         return module, step
 
     def __load_subworkflow_from_step_dict(self, trans, step_dict, subworkflow_id_map, **kwds):
@@ -1041,23 +1059,21 @@ class WorkflowContentsManager(UsesAnnotations):
                     continue
                 if not isinstance(conn_list, list):  # Older style singleton connection
                     conn_list = [conn_list]
+
                 for conn_dict in conn_list:
                     if 'output_name' not in conn_dict or 'id' not in conn_dict:
                         template = "Invalid connection [%s] - must be dict with output_name and id fields."
                         message = template % conn_dict
                         raise exceptions.MessageException(message)
-                    conn = model.WorkflowStepConnection()
-                    conn.input_step = step
-                    conn.input_name = input_name
-                    conn.output_name = conn_dict['output_name']
                     external_id = conn_dict['id']
                     if external_id not in steps_by_external_id:
                         raise KeyError("Failed to find external id %s in %s" % (external_id, steps_by_external_id.keys()))
-                    conn.output_step = steps_by_external_id[external_id]
+                    output_step = steps_by_external_id[external_id]
 
+                    output_name = conn_dict["output_name"]
                     input_subworkflow_step_index = conn_dict.get('input_subworkflow_step_id', None)
-                    if input_subworkflow_step_index is not None:
-                        conn.input_subworkflow_step = step.subworkflow.step_by_index(input_subworkflow_step_index)
+
+                    step.add_connection(input_name, output_name, output_step, input_subworkflow_step_index)
 
             del step.temp_input_connections
 

lib/galaxy/model/__init__.py

@@ -4071,13 +4071,45 @@ class WorkflowStep(RepresentById):
         self.tool_inputs = None
         self.tool_errors = None
         self.position = None
-        self.input_connections = []
+        self.inputs = []
         self.config = None
         self.label = None
         self.uuid = uuid4()
         self.workflow_outputs = []
         self._input_connections_by_name = None
 
+    def get_input(self, input_name):
+        for step_input in self.inputs:
+            if step_input.name == input_name:
+                return step_input
+
+        return None
+
+    def get_or_add_input(self, input_name):
+        step_input = self.get_input(input_name)
+
+        if step_input is None:
+            step_input = WorkflowStepInput(self)
+            step_input.name = input_name
+        return step_input
+
+    def add_connection(self, input_name, output_name, output_step, input_subworkflow_step_index=None):
+        step_input = self.get_or_add_input(input_name)
+
+        conn = WorkflowStepConnection()
+        conn.input_step_input = step_input
+        conn.output_name = output_name
+        conn.output_step = output_step
+        if input_subworkflow_step_index is not None:
+            input_subworkflow_step = self.subworkflow.step_by_index(input_subworkflow_step_index)
+            conn.input_subworkflow_step = input_subworkflow_step
+        return conn
+
+    @property
+    def input_connections(self):
+        connections = [_ for step_input in self.inputs for _ in step_input.connections]
+        return connections
+
     @property
     def unique_workflow_outputs(self):
         # Older Galaxy workflows may have multiple WorkflowOutputs
@@ -4151,7 +4183,7 @@ class WorkflowStep(RepresentById):
         copied_step.position = self.position
         copied_step.config = self.config
         copied_step.label = self.label
-        copied_step.input_connections = copy_list(self.input_connections)
+        copied_step.inputs = copy_list(self.inputs, copied_step)
 
         subworkflow_step_mapping = {}
         subworkflow = self.subworkflow
@@ -4162,8 +4194,7 @@ class WorkflowStep(RepresentById):
                 subworkflow_step_mapping[subworkflow_step.id] = copied_subworkflow_step
 
         for old_conn, new_conn in zip(self.input_connections, copied_step.input_connections):
-            # new_conn.input_step = new_
-            new_conn.input_step = step_mapping[old_conn.input_step_id]
+            new_conn.input_step_input = copied_step.get_or_add_input(old_conn.input_name)
             new_conn.output_step = step_mapping[old_conn.output_step_id]
             if old_conn.input_subworkflow_step_id:
                 new_conn.input_subworkflow_step = subworkflow_step_mapping[old_conn.input_subworkflow_step_id]
@@ -4178,6 +4209,30 @@ class WorkflowStep(RepresentById):
         return "WorkflowStep[index=%d,type=%s]" % (self.order_index, self.type)
 
 
+class WorkflowStepInput(RepresentById):
+    default_merge_type = None
+    default_scatter_type = None
+
+    def __init__(self, workflow_step):
+        self.workflow_step = workflow_step
+        self.name = None
+        self.default_value = None
+        self.default_value_set = False
+        self.merge_type = self.default_merge_type
+        self.scatter_type = self.default_scatter_type
+
+    def copy(self, copied_step):
+        copied_step_input = WorkflowStepInput(copied_step)
+        copied_step_input.name = self.name
+        copied_step_input.default_value = self.default_value
+        copied_step_input.default_value_set = self.default_value_set
+        copied_step_input.merge_type = self.merge_type
+        copied_step_input.scatter_type = self.scatter_type
+
+        copied_step_input.connections = copy_list(self.connections)
+        return copied_step_input
+
+
 class WorkflowStepConnection(RepresentById):
     # Constant used in lieu of output_name and input_name to indicate an
     # implicit connection between two steps that is not dependent on a dataset
@@ -4189,18 +4244,29 @@ class WorkflowStepConnection(RepresentById):
     def __init__(self):
         self.output_step_id = None
         self.output_name = None
-        self.input_step_id = None
-        self.input_name = None
+        self.input_step_input_id = None
 
     @property
     def non_data_connection(self):
         return (self.output_name == self.input_name == WorkflowStepConnection.NON_DATA_CONNECTION)
 
+    @property
+    def input_name(self):
+        return self.input_step_input.name
+
+    @property
+    def input_step(self):
+        return self.input_step_input and self.input_step_input.workflow_step
+
+    @property
+    def input_step_id(self):
+        input_step = self.input_step
+        return input_step and input_step.id
+
     def copy(self):
         # TODO: handle subworkflow ids...
         copied_connection = WorkflowStepConnection()
         copied_connection.output_name = self.output_name
-        copied_connection.input_name = self.input_name
         return copied_connection
 
 

lib/galaxy/model/mapping.py

@@ -908,6 +908,23 @@ model.WorkflowStep.table = Table(
     # Column( "input_connections", JSONType ),
     Column("label", Unicode(255)))
 
+
+model.WorkflowStepInput.table = Table(
+    "workflow_step_input", metadata,
+    Column("id", Integer, primary_key=True),
+    Column("workflow_step_id", Integer, ForeignKey("workflow_step.id"), index=True),
+    Column("name", TEXT),
+    Column("merge_type", TEXT),
+    Column("scatter_type", TEXT),
+    Column("value_from", JSONType),
+    Column("value_from_type", TEXT),
+    Column("default_value", JSONType),
+    Column("default_value_set", Boolean, default=False),
+    Column("runtime_value", Boolean, default=False),
+    UniqueConstraint("workflow_step_id", "name"),
+)
+
+
 model.WorkflowRequestStepState.table = Table(
     "workflow_request_step_states", metadata,
     Column("id", Integer, primary_key=True),
@@ -953,9 +970,8 @@ model.WorkflowStepConnection.table = Table(
     "workflow_step_connection", metadata,
     Column("id", Integer, primary_key=True),
     Column("output_step_id", Integer, ForeignKey("workflow_step.id"), index=True),
-    Column("input_step_id", Integer, ForeignKey("workflow_step.id"), index=True),
+    Column("input_step_input_id", Integer, ForeignKey("workflow_step_input.id"), index=True),
     Column("output_name", TEXT),
-    Column("input_name", TEXT),
     Column("input_subworkflow_step_id", Integer, ForeignKey("workflow_step.id"), index=True),
 )
 
@@ -2218,6 +2234,13 @@ mapper(model.WorkflowStep, model.WorkflowStep.table, properties=dict(
         backref="workflow_steps")
 ))
 
+mapper(model.WorkflowStepInput, model.WorkflowStepInput.table, properties=dict(
+    workflow_step=relation(model.WorkflowStep,
+        backref=backref("inputs", uselist=True),
+        cascade="all",
+        primaryjoin=(model.WorkflowStepInput.table.c.workflow_step_id == model.WorkflowStep.table.c.id))
+))
+
 mapper(model.WorkflowOutput, model.WorkflowOutput.table, properties=dict(
     workflow_step=relation(model.WorkflowStep,
         backref='workflow_outputs',
@@ -2225,10 +2248,10 @@ mapper(model.WorkflowOutput, model.WorkflowOutput.table, properties=dict(
 ))
 
 mapper(model.WorkflowStepConnection, model.WorkflowStepConnection.table, properties=dict(
-    input_step=relation(model.WorkflowStep,
-        backref="input_connections",
+    input_step_input=relation(model.WorkflowStepInput,
+        backref="connections",
         cascade="all",
-        primaryjoin=(model.WorkflowStepConnection.table.c.input_step_id == model.WorkflowStep.table.c.id)),
+        primaryjoin=(model.WorkflowStepConnection.table.c.input_step_input_id == model.WorkflowStepInput.table.c.id)),
     input_subworkflow_step=relation(model.WorkflowStep,
         backref=backref("parent_workflow_input_connections", uselist=True),
         primaryjoin=(model.WorkflowStepConnection.table.c.input_subworkflow_step_id == model.WorkflowStep.table.c.id),

lib/galaxy/model/migrate/versions/0136_collection_and_workflow_state.py

@@ -93,14 +93,6 @@ def upgrade(migrate_engine):
     for table in tables.values():
         __create(table)
 
-    def nextval(table, col='id'):
-        if migrate_engine.name in ['postgres', 'postgresql']:
-            return "nextval('%s_%s_seq')" % (table, col)
-        elif migrate_engine.name in ['mysql', 'sqlite']:
-            return "null"
-        else:
-            raise Exception("Unhandled database type")
-
     # Set default for creation to scheduled, actual mapping has new as default.
     workflow_invocation_step_state_column = Column("state", TrimmedString(64), default="scheduled")
     if migrate_engine.name in ['postgres', 'postgresql']:

lib/galaxy/model/migrate/versions/0145_add_workflow_step_input.py

@@ -0,0 +1,123 @@
+"""
+Migration script for workflow step input table.
+"""
+from __future__ import print_function
+
+import logging
+
+from sqlalchemy import (
+    Boolean,
+    Column,
+    ForeignKey,
+    Integer,
+    MetaData,
+    Table,
+    TEXT,
+    UniqueConstraint
+)
+
+from galaxy.model.custom_types import JSONType
+
+log = logging.getLogger(__name__)
+metadata = MetaData()
+
+
+WorkflowStepInput_table = Table(
+    "workflow_step_input", metadata,
+    Column("id", Integer, primary_key=True),
+    Column("workflow_step_id", Integer, ForeignKey("workflow_step.id"), index=True),
+    Column("name", TEXT),
+    Column("merge_type", TEXT),
+    Column("scatter_type", TEXT),
+    Column("value_from", JSONType),
+    Column("value_from_type", TEXT),
+    Column("default_value", JSONType),
+    Column("default_value_set", Boolean, default=False),
+    Column("runtime_value", Boolean, default=False),
+    UniqueConstraint("workflow_step_id", "name"),
+)
+
+
+def upgrade(migrate_engine):
+    metadata.bind = migrate_engine
+    print(__doc__)
+    metadata.reflect()
+
+    OldWorkflowStepConnection_table = Table("workflow_step_connection", metadata, autoload=True)
+    for index in OldWorkflowStepConnection_table.indexes:
+        index.drop()
+    OldWorkflowStepConnection_table.rename("workflow_step_connection_preupgrade145")
+    # Try to deregister that table to work around some caching problems it seems.
+    OldWorkflowStepConnection_table.deregister()
+    metadata._remove_table("workflow_step_connection", metadata.schema)
+    metadata.reflect()
+
+    NewWorkflowStepConnection_table = Table(
+        "workflow_step_connection", metadata,
+        Column("id", Integer, primary_key=True),
+        Column("output_step_id", Integer, ForeignKey("workflow_step.id"), index=True),
+        Column("input_step_input_id", Integer, ForeignKey("workflow_step_input.id"), index=True),
+        Column("output_name", TEXT),
+        Column("input_subworkflow_step_id", Integer, ForeignKey("workflow_step.id"), index=True),
+    )
+    for table in (WorkflowStepInput_table, NewWorkflowStepConnection_table):
+        _create(table)
+
+    insert_step_inputs_cmd = \
+        "INSERT INTO workflow_step_input (workflow_step_id, name) " + \
+        "SELECT input_step_id, input_name FROM workflow_step_connection_preupgrade145"
+    migrate_engine.execute(insert_step_inputs_cmd)
+
+    insert_step_connections_cmd = \
+        "INSERT INTO workflow_step_connection (output_step_id, input_step_input_id, output_name, input_subworkflow_step_id) " + \
+        "SELECT wsc.output_step_id, wsi.id, wsc.output_name, wsc.input_subworkflow_step_id " + \
+        "FROM workflow_step_connection_preupgrade145 AS wsc JOIN workflow_step_input AS wsi ON wsc.input_step_id = wsi.workflow_step_id AND wsc.input_name = wsi.name ORDER BY wsc.id"
+    migrate_engine.execute(insert_step_connections_cmd)
+    _drop(OldWorkflowStepConnection_table)
+
+
+def downgrade(migrate_engine):
+    metadata.bind = migrate_engine
+
+    NewWorkflowStepConnection_table = Table("workflow_step_connection", metadata, autoload=True)
+    for index in NewWorkflowStepConnection_table.indexes:
+        index.drop()
+    NewWorkflowStepConnection_table.rename("workflow_step_connection_predowngrade145")
+    # Try to deregister that table to work around some caching problems it seems.
+    NewWorkflowStepConnection_table.deregister()
+    metadata._remove_table("workflow_step_connection", metadata.schema)
+    metadata.reflect()
+
+    OldWorkflowStepConnection_table = Table(
+        "workflow_step_connection", metadata,
+        Column("id", Integer, primary_key=True),
+        Column("output_step_id", Integer, ForeignKey("workflow_step.id"), index=True),
+        Column("input_step_id", Integer, ForeignKey("workflow_step.id"), index=True),
+        Column("output_name", TEXT),
+        Column("input_name", TEXT),
+        Column("input_subworkflow_step_id", Integer, ForeignKey("workflow_step.id"), index=True),
+    )
+    _create(OldWorkflowStepConnection_table)
+
+    insert_step_connections_cmd = \
+        "INSERT INTO workflow_step_connection (output_step_id, input_step_id, output_name, input_name, input_subworkflow_step_id) " + \
+        "SELECT wsc.output_step_id, wsi.workflow_step_id, wsc.output_name, wsi.name, wsc.input_subworkflow_step_id " + \
+        "FROM workflow_step_connection_predowngrade145 AS wsc JOIN workflow_step_input AS wsi ON wsc.input_step_input_id = wsi.id ORDER BY wsc.id"
+    migrate_engine.execute(insert_step_connections_cmd)
+
+    for table in (WorkflowStepInput_table, NewWorkflowStepConnection_table):
+        _drop(table)
+
+
+def _create(table):
+    try:
+        table.create()
+    except Exception:
+        log.exception("Creating %s table failed.", table.name)
+
+
+def _drop(table):
+    try:
+        table.drop()
+    except Exception:
+        log.exception("Dropping %s table failed.", table.name)

lib/galaxy/webapps/tool_shed/model/__init__.py

@@ -479,10 +479,34 @@ class WorkflowStep(object):
         self.tool_inputs = None
         self.tool_errors = None
         self.position = None
-        self.input_connections = []
+        self.inputs = []
         self.config = None
         self.label = None
 
+    def get_or_add_input(self, input_name):
+        for step_input in self.inputs:
+            if step_input.name == input_name:
+                return step_input
+
+        step_input = WorkflowStepInput()
+        step_input.workflow_step = self
+        step_input.name = input_name
+        self.inputs.append(step_input)
+        return step_input
+
+    @property
+    def input_connections(self):
+        connections = [_ for step_input in self.inputs for _ in step_input.connections]
+        return connections
+
+
+class WorkflowStepInput(object):
+
+    def __init__(self):
+        self.id = None
+        self.name = None
+        self.connections = []
+
 
 class WorkflowStepConnection(object):
 

lib/galaxy/workflow/extract.py

@@ -124,10 +124,10 @@ def extract_steps(trans, history=None, job_ids=None, dataset_ids=None, dataset_c
                 else:
                     log.info("Cannot find implicit input collection for %s" % input_name)
             if other_hid in hid_to_output_pair:
+                step_input = step.get_or_add_input(input_name)
                 other_step, other_name = hid_to_output_pair[other_hid]
                 conn = model.WorkflowStepConnection()
-                conn.input_step = step
-                conn.input_name = input_name
+                conn.input_step_input = step_input
                 # Should always be connected to an earlier step
                 conn.output_step = other_step
                 conn.output_name = other_name

lib/galaxy/workflow/modules.py

@@ -197,11 +197,14 @@ class WorkflowModule(object):
         """
         return {}
 
-    def compute_runtime_state(self, trans, step_updates=None):
+    def compute_runtime_state(self, trans, step=None, step_updates=None):
         """ Determine the runtime state (potentially different from self.state
         which describes configuration state). This (again unlike self.state) is
         currently always a `DefaultToolState` object.
 
+        If `step` is not `None`, it will be used to search for default values
+        defined in workflow input steps.
+
         If `step_updates` is `None`, this is likely for rendering the run form
         for instance and no runtime properties are available and state must be
         solely determined by the default runtime state described by the step.
@@ -211,6 +214,21 @@ class WorkflowModule(object):
         """
         state = self.get_runtime_state()
         step_errors = {}
+
+        if step is not None:
+
+            def update_value(input, context, prefixed_name, **kwargs):
+                step_input = step.get_input(prefixed_name)
+                if step_input is None:
+                    return NO_REPLACEMENT
+
+                if step_input.default_value_set:
+                    return step_input.default_value
+
+                return NO_REPLACEMENT
+
+            visit_input_values(self.get_runtime_inputs(), state.inputs, update_value, no_replacement_value=NO_REPLACEMENT)
+
         if step_updates:
 
             def update_value(input, context, prefixed_name, **kwargs):
@@ -489,6 +507,14 @@ class InputProxy(object):
         self.input = input
         self.prefixed_name = prefixed_name
 
+    @property
+    def name(self):
+        return self.prefixed_name
+
+    @property
+    def label(self):
+        return self.prefixed_name
+
     def to_dict(self, *args, **kwds):
         as_dict = self.input.to_dict(*args, **kwds)
         as_dict["name"] = self.prefixed_name
@@ -972,7 +998,7 @@ class ToolModule(WorkflowModule):
         """
         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.compute_runtime_state(self.trans, step=None, 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)
 
@@ -1042,14 +1068,14 @@ class ToolModule(WorkflowModule):
     def get_runtime_inputs(self, **kwds):
         return self.get_inputs()
 
-    def compute_runtime_state(self, trans, step_updates=None):
+    def compute_runtime_state(self, trans, step=None, step_updates=None):
         # Warning: This method destructively modifies existing step state.
         if self.tool:
             step_errors = {}
             state = self.state
             self.runtime_post_job_actions = {}
+            state, step_errors = super(ToolModule, self).compute_runtime_state(trans, step, step_updates)
             if step_updates:
-                state, step_errors = super(ToolModule, self).compute_runtime_state(trans, step_updates)
                 self.runtime_post_job_actions = step_updates.get(RUNTIME_POST_JOB_ACTIONS_KEY, {})
                 step_metadata_runtime_state = self.__step_meta_runtime_state()
                 if step_metadata_runtime_state:
@@ -1404,7 +1430,7 @@ class WorkflowModuleInjector(object):
             subworkflow = step.subworkflow
             populate_module_and_state(self.trans, subworkflow, param_map=unjsonified_subworkflow_param_map)
 
-        state, step_errors = module.compute_runtime_state(self.trans, step_args)
+        state, step_errors = module.compute_runtime_state(self.trans, step, step_args)
         step.state = state
 
         # Fix any missing parameters

lib/tool_shed/util/workflow_util.py

@@ -306,13 +306,12 @@ def get_workflow_from_dict(trans, workflow_dict, tools_metadata, repository_id,
         # Input connections.
         for input_name, conn_dict in step.temp_input_connections.items():
             if conn_dict:
+                step_input = step.get_or_add_input(input_name)
                 output_step = steps_by_external_id[conn_dict['id']]
                 conn = trans.model.WorkflowStepConnection()
-                conn.input_step = step
-                conn.input_name = input_name
+                conn.input_step_input = step_input
                 conn.output_step = output_step
                 conn.output_name = conn_dict['output_name']
-                step.input_connections.append(conn)
         del step.temp_input_connections
     # Order the steps if possible.
     attach_ordered_steps(workflow, steps)

test/api/test_workflows.py

@@ -1410,13 +1410,14 @@ steps:
       steps:
         random_lines:
           tool_id: random_lines1
-          state:
-            num_lines: 2
-            input:
-              $link: inner_input
-            seed_source:
-              seed_source_selector: set_seed
-              seed: asdf
+          in:
+            input: inner_input
+            num_lines:
+              default: 2
+            seed_source|seed_source_selector:
+              default: set_seed
+            seed_source|seed:
+              default: asdf
         split:
           tool_id: split
           in:
@@ -2868,6 +2869,73 @@ steps:
         self.dataset_populator.wait_for_history(history_id, assert_ok=True)
         self.assertEqual("2\n", self.dataset_populator.get_history_dataset_content(history_id))
 
+    @skip_without_tool("random_lines1")
+    def test_run_replace_params_over_default(self):
+        with self.dataset_populator.test_history() as history_id:
+            self._run_jobs("""
+class: GalaxyWorkflow
+inputs:
+  input: data
+steps:
+  randomlines:
+    tool_id: random_lines1
+    in:
+      input: input
+      num_lines:
+        default: 6
+""", test_data="""
+step_parameters:
+  '1':
+    num_lines: 4
+input:
+  value: 1.bed
+  type: File
+""", history_id=history_id, wait=True, assert_ok=True, round_trip_format_conversion=True)
+            result = self.dataset_populator.get_history_dataset_content(history_id)
+            assert result.count("\n") == 4
+
+    @skip_without_tool("random_lines1")
+    def test_run_replace_params_over_default_delayed(self):
+        with self.dataset_populator.test_history() as history_id:
+            run_summary = self._run_jobs("""
+class: GalaxyWorkflow
+inputs:
+  input: data
+steps:
+  first_cat:
+    tool_id: cat1
+    in:
+      input1: input
+  the_pause:
+    type: pause
+    in:
+      input: first_cat/out_file1
+  randomlines:
+    tool_id: random_lines1
+    in:
+      input: the_pause
+      num_lines:
+        default: 6
+""", test_data="""
+step_parameters:
+  '3':
+    num_lines: 4
+input:
+  value: 1.bed
+  type: File
+""", history_id=history_id, wait=False)
+            wait_on(lambda: len(self._history_jobs(history_id)) >= 2 or None, "history jobs")
+            self.dataset_populator.wait_for_history(history_id, assert_ok=True)
+
+            workflow_id = run_summary.workflow_id
+            invocation_id = run_summary.invocation_id
+
+            self.__review_paused_steps(workflow_id, invocation_id, order_index=2, action=True)
+            self.wait_for_invocation_and_jobs(history_id, workflow_id, invocation_id)
+
+            result = self.dataset_populator.get_history_dataset_content(history_id)
+            assert result.count("\n") == 4
+
     def test_pja_import_export(self):
         workflow = self.workflow_populator.load_workflow(name="test_for_pja_import", add_pja=True)
         uploaded_workflow_id = self.workflow_populator.create_workflow(workflow)

test/unit/test_galaxy_mapping.py

@@ -485,6 +485,11 @@ class MappingTests(unittest.TestCase):
         workflow_step_2.type = "subworkflow"
         workflow_step_2.subworkflow = child_workflow
 
+        workflow_step_1.get_or_add_input("moo1")
+        workflow_step_1.get_or_add_input("moo2")
+        workflow_step_2.get_or_add_input("moo")
+        workflow_step_1.add_connection("foo", "cow", workflow_step_2)
+
         workflow = workflow_from_steps([workflow_step_1, workflow_step_2])
         self.persist(workflow)
 

test/unit/workflows/test_modules.py

@@ -40,7 +40,7 @@ def test_data_input_step_modified_state():
 
 def test_data_input_compute_runtime_state_default():
     module = __from_step(type="data_input")
-    state, errors = module.compute_runtime_state(module.trans)
+    state, errors = module.compute_runtime_state(module.trans, module.test_step)
     assert not errors
     assert "input" in state.inputs
     assert state.inputs["input"] is None
@@ -52,7 +52,7 @@ def test_data_input_compute_runtime_state_args():
     hda = model.HistoryDatasetAssociation()
     with mock.patch("galaxy.workflow.modules.check_param") as check_method:
         check_method.return_value = (hda, None)
-        state, errors = module.compute_runtime_state(module.trans, {"input": 4, "tool_state": tool_state})
+        state, errors = module.compute_runtime_state(module.trans, module.test_step, {"input": 4, "tool_state": tool_state})
     assert not errors
     assert "input" in state.inputs
     assert state.inputs["input"] is hda
@@ -166,25 +166,28 @@ steps:
     label: "input2"
   - type: "tool"
     tool_id: "cat1"
-    input_connections:
-    -  input_name: "input1"
-       "@output_step": 0
-       output_name: "output"
+    inputs:
+      input1:
+        connections:
+        - "@output_step": 0
+          output_name: "output"
   - type: "tool"
     tool_id: "cat1"
-    input_connections:
-    -  input_name: "input1"
-       "@output_step": 0
-       output_name: "output"
+    inputs:
+      input1:
+        connections:
+        - "@output_step": 0
+          output_name: "output"
     workflow_outputs:
     -   output_name: "out_file1"
         label: "out1"
   - type: "tool"
     tool_id: "cat1"
-    input_connections:
-    -  input_name: "input1"
-       "@output_step": 2
-       output_name: "out_file1"
+    inputs:
+      input1:
+        connections:
+        - "@output_step": 2
+          output_name: "out_file1"
     workflow_outputs:
     -   output_name: "out_file1"
 """

test/unit/workflows/test_render.py

@@ -6,28 +6,28 @@ steps:
   - type: "data_input"
     order_index: 0
     tool_inputs: {"name": "input1"}
-    input_connections: []
     position: {"top": 3, "left": 3}
   - type: "data_input"
     order_index: 1
     tool_inputs: {"name": "input2"}
-    input_connections: []
     position: {"top": 6, "left": 4}
   - type: "tool"
     tool_id: "cat1"
     order_index: 2
-    input_connections:
-    -  input_name: "input1"
-       "@output_step": 0
-       output_name: "di1"
+    inputs:
+      input1:
+        connection:
+        - "@output_step": 0
+          output_name: "di1"
     position: {"top": 13, "left": 10}
   - type: "tool"
     tool_id: "cat1"
     order_index: 3
-    input_connections:
-    -  input_name: "input1"
-       "@output_step": 0
-       output_name: "di1"
+    inputs:
+      input1:
+        connection:
+        - "@output_step": 0
+          output_name: "di1"
     position: {"top": 33, "left": 103}
 """
 

test/unit/workflows/test_workflow_progress.py

@@ -12,22 +12,25 @@ steps:
     tool_inputs: {"name": "input2"}
   - type: "tool"
     tool_id: "cat1"
-    input_connections:
-    -  input_name: "input1"
-       "@output_step": 0
-       output_name: "output"
+    inputs:
+      "input1":
+        connections:
+        - "@output_step": 0
+          output_name: "output"
   - type: "tool"
     tool_id: "cat1"
-    input_connections:
-    -  input_name: "input1"
-       "@output_step": 0
-       output_name: "output"
+    inputs:
+      input1:
+        connections:
+        - "@output_step": 0
+          output_name: "output"
   - type: "tool"
     tool_id: "cat1"
-    input_connections:
-    -  input_name: "input1"
-       "@output_step": 2
-       output_name: "out_file1"
+    inputs:
+      "input1":
+        connections:
+        - "@output_step": 2
+          output_name: "out_file1"
 """
 
 TEST_SUBWORKFLOW_YAML = """
@@ -41,15 +44,17 @@ steps:
             tool_inputs: {"name": "inner_input"}
           - type: "tool"
             tool_id: "cat1"
-            input_connections:
-            -  input_name: "input1"
-               "@output_step": 0
-               output_name: "output"
-    input_connections:
-    -  input_name: "inner_input"
-       "@output_step": 0
-       output_name: "output"
-       "@input_subworkflow_step": 0
+            inputs:
+              "input1":
+                  connections:
+                  - "@output_step": 0
+                    output_name: "output"
+    inputs:
+      inner_input:
+        connections:
+        - "@output_step": 0
+          output_name: "output"
+          "@input_subworkflow_step": 0
 """
 
 UNSCHEDULED_STEP = object()

test/unit/workflows/workflow_support.py

@@ -99,20 +99,28 @@ def yaml_to_model(has_dict, id_offset=100):
 
         for key, value in step.items():
             if key == "input_connections":
-                connections = []
-                for conn_dict in value:
-                    conn = model.WorkflowStepConnection()
-                    for conn_key, conn_value in conn_dict.items():
-                        if conn_key == "@output_step":
-                            target_step = workflow.steps[conn_value]
-                            conn_value = target_step
-                            conn_key = "output_step"
-                        if conn_key == "@input_subworkflow_step":
-                            conn_value = step["subworkflow"].step_by_index(conn_value)
-                            conn_key = "input_subworkflow_step"
-                        setattr(conn, conn_key, conn_value)
-                    connections.append(conn)
-                value = connections
+                raise NotImplementedError()
+            if key == "inputs":
+                inputs = []
+                for input_name, input_def in value.items():
+                    step_input = model.WorkflowStepInput(workflow_step)
+                    step_input.name = input_name
+                    connections = []
+                    for conn_dict in input_def.get("connections", []):
+                        conn = model.WorkflowStepConnection()
+                        for conn_key, conn_value in conn_dict.items():
+                            if conn_key == "@output_step":
+                                target_step = workflow.steps[conn_value]
+                                conn_value = target_step
+                                conn_key = "output_step"
+                            if conn_key == "@input_subworkflow_step":
+                                conn_value = step["subworkflow"].step_by_index(conn_value)
+                                conn_key = "input_subworkflow_step"
+                            setattr(conn, conn_key, conn_value)
+                        connections.append(conn)
+                    step_input.connections = connections
+                    inputs.append(step_input)
+                value = inputs
             if key == "workflow_outputs":
                 value = [partial(_dict_to_workflow_output, workflow_step)(_) for _ in value]
             setattr(workflow_step, key, value)