diff --git a/lib/galaxy/webapps/galaxy/services/wes.py b/lib/galaxy/webapps/galaxy/services/wes.py index 3d3ac701847..02837ed8702 100644 --- a/lib/galaxy/webapps/galaxy/services/wes.py +++ b/lib/galaxy/webapps/galaxy/services/wes.py @@ -493,7 +493,11 @@ class WesService(ServiceBase): ) # Use the existing workflow directly - no need to create a new one - # Skip to step 5 (engine parameters and history) + if instance: + # The URI named a specific version - invoke that one, not the latest. + invoke_workflow_id = trans.security.decode_id(encoded_workflow_id) + else: + invoke_workflow_id = stored_workflow.id else: # Step 2: Determine/validate workflow type detected_type = _determine_workflow_type(workflow_dict) @@ -520,6 +524,8 @@ class WesService(ServiceBase): source="WES API", ) stored_workflow = created_workflow.stored_workflow + invoke_workflow_id = stored_workflow.id + instance = False # Step 5: Parse engine parameters and create/select history engine_params = {} @@ -532,9 +538,10 @@ class WesService(ServiceBase): history = _get_or_create_history(trans, engine_params) # Step 6: Parse workflow parameters - invoke_params = { + invoke_params: dict[str, Any] = { "history_id": trans.security.encode_id(history.id), "inputs_by": "name", + "instance": instance, } if workflow_params: @@ -554,7 +561,7 @@ class WesService(ServiceBase): invoke_payload = InvokeWorkflowPayload(**invoke_params) workflow_invocation_response = self._workflows_service.invoke_workflow( trans, - trans.security.encode_id(stored_workflow.id), + invoke_workflow_id, invoke_payload, ) diff --git a/lib/galaxy_test/api/test_wes.py b/lib/galaxy_test/api/test_wes.py index 4eba1a99748..af35bc7651d 100644 --- a/lib/galaxy_test/api/test_wes.py +++ b/lib/galaxy_test/api/test_wes.py @@ -520,16 +520,23 @@ steps: self._assert_status_code_is(response, 200) def test_wes_submit_run_with_gxworkflow_uri_with_instance_param(self): - """Test gxworkflow:// URI with instance=true parameter.""" + """Test gxworkflow:// URI with instance=true runs that version, not the latest.""" with self.dataset_populator.test_history() as history_id: dataset_id = self._get_test_dataset_id(history_id) # Upload a workflow to get its ID workflow_id = self._upload_yaml_workflow(WORKFLOW_SIMPLE) - latest_instance_id = self._latest_instance_id(workflow_id, history_id) + first_instance_id = self._latest_instance_id(workflow_id, history_id) + + # Create a second version so the referenced instance is no longer the latest + workflow_object = self._download_workflow(workflow_id) + workflow_object["steps"]["1"]["annotation"] = "second version" + update_response = self.workflow_populator.update_workflow(workflow_id, workflow_object) + self._assert_status_code_is(update_response, 200) + assert self._latest_instance_id(workflow_id, history_id) != first_instance_id # Construct gxworkflow:// URI with instance=true - workflow_uri = f"gxworkflow://{latest_instance_id}?instance=true" + workflow_uri = f"gxworkflow://{first_instance_id}?instance=true" # Submit workflow using the gxworkflow:// URI data = { @@ -545,6 +552,8 @@ steps: # Validate response assert run_id is not None + invocation = self.workflow_populator.get_invocation(run_id) + assert invocation["workflow_id"] == first_instance_id def test_wes_submit_run_with_gxworkflow_uri_published_workflow(self): """A published workflow owned by another user can be run via gxworkflow://."""