Fix PUT /api/workflows for user defined tools (03a485e2) · Commits · NDIP / Galaxy

lib/galaxy/workflow/modules.py

+3 −1

Original line number	Diff line number	Diff line
		@@ -2015,7 +2015,9 @@ class ToolModule(WorkflowModule):
		else:
		step.tool_version = self.tool_version
		if tool_uuid := getattr(self, "tool_uuid", None):
		step.dynamic_tool = self.trans.app.dynamic_tool_manager.get_tool_by_uuid(tool_uuid)
		tool = self.trans.app.toolbox.get_tool(tool_uuid=tool_uuid, user=self.trans.user)
		if tool:
		step.dynamic_tool = tool.dynamic_tool
		if not detached:
		for k, v in self.post_job_actions.items():
		pja = self.__to_pja(k, v, step)

lib/galaxy_test/api/test_workflows.py

+29 −0

Original line number	Diff line number	Diff line
		@@ -26,8 +26,10 @@ from requests import (
		)

		from galaxy.exceptions import error_codes
		from galaxy.tool_util_models import UserToolSource
		from galaxy.util import UNKNOWN
		from galaxy.util.unittest_utils import skip_if_github_down
		from galaxy_test.api.test_tools import TOOL_WITH_SHELL_COMMAND
		from galaxy_test.base import rules_test_data
		from galaxy_test.base.populators import (
		DatasetCollectionPopulator,
		@@ -8085,6 +8087,33 @@ outer_input:
		history_id=history_id,
		)

		def test_user_defined_workflow_update(self):
		with self.dataset_populator.user_tool_execute_permissions():
		unprivileged_tool = self.dataset_populator.create_unprivileged_tool(
		UserToolSource(**TOOL_WITH_SHELL_COMMAND)
		)
		# Workflow doesn't matter, we're replacing it in the update
		workflow = self.workflow_populator.load_workflow_from_resource("test_workflow_pause")
		workflow_id = self.workflow_populator.create_workflow(workflow)
		update_response = self._update_workflow(
		workflow_id,
		{
		"steps": {
		"0": {
		"content_id": "cat_user_defined",
		"id": 1,
		"input_connections": {"datasets": []},
		"name": "Concatenate Files",
		"tool_uuid": unprivileged_tool["uuid"],
		"type": "tool",
		}
		},
		},
		)
		assert update_response.status_code == 200, update_response.text
		workflow = self.workflow_populator.download_workflow(workflow_id)
		assert workflow["steps"]["0"]["tool_representation"]["class"] == "GalaxyUserTool"

		def _invoke_paused_workflow(self, history_id):
		workflow = self.workflow_populator.load_workflow_from_resource("test_workflow_pause")
		workflow_id = self.workflow_populator.create_workflow(workflow)