Merge pull request #10109 from mvdbeek/add_subworkflow_invocation_id_to_workflow_invocation_step

Add subworkflow_invocation_id to invocation step

lib/galaxy/model/__init__.py

@@ -5415,8 +5415,8 @@ class WorkflowInvocationToSubworkflowInvocationAssociation(Dictifiable, Represen
 
 
 class WorkflowInvocationStep(Dictifiable, RepresentById):
-    dict_collection_visible_keys = ['id', 'update_time', 'job_id', 'workflow_step_id', 'state', 'action']
-    dict_element_visible_keys = ['id', 'update_time', 'job_id', 'workflow_step_id', 'state', 'action']
+    dict_collection_visible_keys = ['id', 'update_time', 'job_id', 'workflow_step_id', 'subworkflow_invocation_id', 'state', 'action']
+    dict_element_visible_keys = ['id', 'update_time', 'job_id', 'workflow_step_id', 'subworkflow_invocation_id', 'state', 'action']
     states = Bunch(
         NEW='new',  # Brand new workflow invocation step
         READY='ready',  # Workflow invocation step ready for another iteration of scheduling.

lib/galaxy/model/mapping.py

@@ -2587,7 +2587,14 @@ mapper(model.WorkflowInvocationToSubworkflowInvocationAssociation, model.Workflo
 simple_mapping(model.WorkflowInvocationStep,
     workflow_step=relation(model.WorkflowStep),
     job=relation(model.Job, backref=backref('workflow_invocation_step', uselist=False), uselist=False),
-    implicit_collection_jobs=relation(model.ImplicitCollectionJobs, backref=backref('workflow_invocation_step', uselist=False), uselist=False),)
+    implicit_collection_jobs=relation(model.ImplicitCollectionJobs, backref=backref('workflow_invocation_step', uselist=False), uselist=False),
+    subworkflow_invocation_id=column_property(
+        select([(model.WorkflowInvocationToSubworkflowInvocationAssociation.table.c.subworkflow_invocation_id)]).where(and_(
+            model.WorkflowInvocationToSubworkflowInvocationAssociation.table.c.workflow_invocation_id == model.WorkflowInvocationStep.table.c.workflow_invocation_id,
+            model.WorkflowInvocationToSubworkflowInvocationAssociation.table.c.workflow_step_id == model.WorkflowInvocationStep.table.c.workflow_step_id,
+        )),
+    ),
+)
 
 
 simple_mapping(model.WorkflowRequestInputParameter,

lib/galaxy_test/api/test_workflows.py

@@ -1133,7 +1133,7 @@ test_data:
 
     def test_run_subworkflow_simple(self):
         with self.dataset_populator.test_history() as history_id:
-            self._run_jobs(WORKFLOW_NESTED_SIMPLE, test_data="""
+            run_response = self._run_jobs(WORKFLOW_NESTED_SIMPLE, test_data="""
 outer_input:
   value: 1.bed
   type: File
@@ -1141,6 +1141,13 @@ outer_input:
 
             content = self.dataset_populator.get_history_dataset_content(history_id)
             self.assertEqual("chrX\t152691446\t152691471\tCCDS14735.1_cds_0_0_chrX_152691447_f\t0\t+\nchrX\t152691446\t152691471\tCCDS14735.1_cds_0_0_chrX_152691447_f\t0\t+\n", content)
+            steps = self.workflow_populator.get_invocation(run_response.invocation_id)['steps']
+            assert sum(1 for step in steps if step['subworkflow_invocation_id'] is None) == 3
+            subworkflow_invocation_id = [step['subworkflow_invocation_id'] for step in steps if step['subworkflow_invocation_id']][0]
+            subworkflow_invocation = self.workflow_populator.get_invocation(subworkflow_invocation_id)
+            # inner_input should be step 0, random_lines should be step 1, but step order gets lost on python < 3.6
+            assert [step for step in subworkflow_invocation['steps'] if step['workflow_step_label'] == 'inner_input']
+            assert [step for step in subworkflow_invocation['steps'] if step['workflow_step_label'] == 'random_lines']
 
     @skip_without_tool("random_lines1")
     def test_run_subworkflow_runtime_parameters(self):

lib/galaxy_test/base/populators.py

@@ -808,6 +808,11 @@ class BaseWorkflowPopulator:
         self.wait_for_invocation(workflow_id, invocation_id, timeout=timeout)
         self.dataset_populator.wait_for_history_jobs(history_id, assert_ok=assert_ok, timeout=timeout)
 
+    def get_invocation(self, invocation_id):
+        r = self._get("invocations/%s" % invocation_id)
+        r.raise_for_status()
+        return r.json()
+
     def invoke_workflow_raw(self, workflow_id, request):
         url = "workflows/%s/usage" % (workflow_id)
         invocation_response = self._post(url, data=request)