do not invoke workflow if not all tools are available

lib/galaxy/managers/workflows.py

@@ -539,7 +539,6 @@ class WorkflowContentsManager(UsesAnnotations):
         option describes the workflow in a context more tied to the current Galaxy instance and includes
         fields like 'url' and 'url' and actual unencoded step ids instead of 'order_index'.
         """
-
         def to_format_2(wf_dict, **kwds):
             return from_galaxy_native(wf_dict, None, **kwds)
 
@@ -1486,6 +1485,16 @@ class WorkflowContentsManager(UsesAnnotations):
             dry_run=refactor_request.dry_run,
         )
 
+    def get_all_tool_ids(self, workflow):
+        tool_ids = set()
+        for step in workflow.steps:
+            if step.type == 'tool':
+                if step.tool_id:
+                    tool_ids.add(step.tool_id)
+            elif step.type == 'subworkflow':
+                tool_ids.update(self.get_all_tool_ids(step.subworkflow))
+        return tool_ids
+
 
 class RefactorRequest(RefactorActions):
     style: str = "export"

lib/galaxy/webapps/galaxy/api/workflows.py

@@ -190,15 +190,10 @@ class WorkflowsAPIController(BaseGalaxyAPIController, UsesStoredWorkflowMixin, U
             workflows = []
             workflows_by_toolshed = dict()
             for value in rval:
-                tool_ids = []
-                workflow_details = self.workflow_contents_manager.workflow_to_dict(trans, self.__get_stored_workflow(trans, value['id']), style='instance')
-                if 'steps' in workflow_details:
-                    for step in workflow_details['steps']:
-                        tool_id = workflow_details['steps'][step].get('tool_id')
-                        if tool_id and tool_id not in tool_ids and self.app.toolbox.is_missing_shed_tool(tool_id):
-                            tool_ids.append(tool_id)
-                if len(tool_ids) > 0:
-                    value['missing_tools'] = tool_ids
+                tool_ids = self.workflow_contents_manager.get_all_tool_ids(self.__get_stored_workflow(trans, value['id']).latest_workflow)
+                missing_tool_ids = [tool_id for tool_id in tool_ids if self.app.toolbox.is_missing_shed_tool(tool_id)]
+                if len(missing_tool_ids) > 0:
+                    value['missing_tools'] = missing_tool_ids
                     workflows_missing_tools.append(value)
             for workflow in workflows_missing_tools:
                 for tool_id in workflow['missing_tools']:
@@ -828,6 +823,8 @@ class WorkflowsAPIController(BaseGalaxyAPIController, UsesStoredWorkflowMixin, U
 
         .. note:: This method takes the same arguments as
             :func:`galaxy.webapps.galaxy.api.workflows.WorkflowsAPIController.create` above.
+
+        :raises: exceptions.MessageException, exceptions.RequestParameterInvalidException
         """
         # Get workflow + accessibility check.
         stored_workflow = self.__get_stored_accessible_workflow(trans, workflow_id)
@@ -837,6 +834,10 @@ class WorkflowsAPIController(BaseGalaxyAPIController, UsesStoredWorkflowMixin, U
         if not is_batch and len(run_configs) != 1:
             raise exceptions.RequestParameterInvalidException("Must specify 'batch' to use batch parameters.")
 
+        tool_ids = self.workflow_contents_manager.get_all_tool_ids(workflow)
+        if not all([self.app.toolbox.has_tool(tool_id) for tool_id in tool_ids]):
+            raise exceptions.MessageException("Workflow was not invoked; some required tools are not installed.")
+
         invocations = []
         for run_config in run_configs:
             workflow_scheduler_id = payload.get('scheduler', None)

lib/galaxy_test/api/test_workflows.py

@@ -928,6 +928,19 @@ steps:
         self._assert_status_code_is(run_workflow_response, 200)
         self.dataset_populator.wait_for_history(history_id, assert_ok=True)
 
+    def test_run_workflow_with_missing_tool(self):
+        with self.dataset_populator.test_history() as history_id:
+            workflow_id = self._upload_yaml_workflow("""
+class: GalaxyWorkflow
+steps:
+  step1:
+    tool_id: nonexistent_tool
+    tool_version: "0.1"
+""")
+            invocation_response = self.__invoke_workflow(history_id, workflow_id, assert_ok=False)
+            self._assert_status_code_is(invocation_response, 400)
+            self.assertEqual(invocation_response.json().get('err_msg'), "Workflow was not invoked; some required tools are not installed.")
+
     @skip_without_tool("collection_creates_pair")
     def test_workflow_run_output_collections(self):
         with self.dataset_populator.test_history() as history_id: