opensource/galaxy
3
0
Fork 0
mirror of https://github.com/galaxyproject/galaxy.git synced 2026-09-24 16:30:27 +08:00
Code Issues Packages Projects Releases Wiki Activity

WES: gxworkflow://?instance=true ran latest version, not referenced one

Browse Source 
Resolved the Workflow instance for the access check but then invoked by
stored workflow id with instance unset. Pass the instance id and flag
through to invoke_workflow; strengthen the test to assert the invocation
used the referenced version.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
John Chilton
2026-07-21 08:26:06 -04:00
co-authored by Claude Opus 4.8
parent e4e29c74e1
commit 90e17de3fe
2 changed files with 22 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/galaxy/webapps/galaxy/services/wes.py
+10 -3
View File
@@ -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,
)
lib/galaxy_test/api/test_wes.py
+12 -3
View File
@@ -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://."""