Fix workflow download when using instance id

planemo/commands/cmd_workflow_test_on_invocation.py

@@ -6,6 +6,7 @@
 from planemo.engine.factory import engine_context
 from planemo.galaxy.activity import invocation_to_run_response
 from planemo.galaxy.test.actions import handle_reports_and_summary
+from planemo.galaxy.workflows import GALAXY_WORKFLOW_INSTANCE_PREFIX
 from planemo.runnable import definition_to_test_case
 from planemo.runnable_resolve import for_runnable_identifier
 from planemo.test.results import StructuredData
@@ -24,7 +25,7 @@ def cli(ctx, path, workflow_identifier, test_index, **kwds):
     with engine_context(ctx, engine="external_galaxy", **kwds) as engine, engine.ensure_runnables_served([]) as config:
         user_gi = config.user_gi
         invocation = user_gi.invocations.show_invocation(workflow_identifier)
-        runnable = for_runnable_identifier(ctx, invocation["workflow_id"], kwds)
+        runnable = for_runnable_identifier(ctx, f"{GALAXY_WORKFLOW_INSTANCE_PREFIX}{invocation['workflow_id']}", kwds)
         test_cases = definition_to_test_case(path, runnable)
         assert (
             len(test_cases) >= test_index

planemo/galaxy/activity.py

@@ -763,8 +763,8 @@ def _history_id(gi, **kwds) -> str:
     return history_id
 
 
-def get_dict_from_workflow(gi, workflow_id):
-    return gi.workflows.export_workflow_dict(workflow_id)
+def get_dict_from_workflow(gi, workflow_id, instance=False):
+    return gi.workflows._get(workflow_id, params={"instance": instance})
 
 
 def wait_for_invocation_and_jobs(

planemo/galaxy/api.py

@@ -7,6 +7,10 @@
 DEFAULT_ADMIN_API_KEY = "test_key"
 
 
+def get_dict_from_workflow(gi: GalaxyInstance, workflow_id: str, instance: bool = False):
+    return gi.workflows._get(workflow_id, params={"instance": instance})
+
+
 def gi(port: Optional[int] = None, url: Optional[str] = None, key: Optional[str] = None) -> GalaxyInstance:
     """Return a bioblend ``GalaxyInstance`` for Galaxy on this port."""
     if key is None:

planemo/galaxy/workflows.py

@@ -25,11 +25,15 @@
     outputs_normalized,
 )
 
-from planemo.galaxy.api import gi
+from planemo.galaxy.api import (
+    get_dict_from_workflow,
+    gi,
+)
 from planemo.io import warn
 
 FAILED_REPOSITORIES_MESSAGE = "Failed to install one or more repositories."
 GALAXY_WORKFLOWS_PREFIX = "gxid://workflows/"
+GALAXY_WORKFLOW_INSTANCE_PREFIX = "gxid://workflow-instance/"
 
 
 def load_shed_repos(runnable):
@@ -156,10 +160,11 @@ def remote_runnable_to_workflow_id(runnable):
 
 def describe_outputs(runnable, gi=None):
     """Return a list of :class:`WorkflowOutput` objects for target workflow."""
-    if runnable.uri.startswith(GALAXY_WORKFLOWS_PREFIX):
+    if runnable.uri.startswith("gxid://"):
         workflow_id = remote_runnable_to_workflow_id(runnable)
         assert gi is not None
-        workflow = get_dict_from_workflow(gi, workflow_id)
+        instance = runnable.uri.startswith(GALAXY_WORKFLOW_INSTANCE_PREFIX)
+        workflow = get_dict_from_workflow(gi, workflow_id, instance)
     else:
         workflow = _raw_dict(runnable.path)
 
@@ -291,10 +296,6 @@ def new_workflow_associated_path(workflow_path, suffix="tests"):
     return base + sep + suffix + "." + ext
 
 
-def get_dict_from_workflow(gi, workflow_id):
-    return gi.workflows.export_workflow_dict(workflow_id)
-
-
 def rewrite_job_file(input_file, output_file, job):
     """Rewrite a job file with galaxy_ids for upload_data subcommand"""
     with open(input_file) as f:
@@ -310,7 +311,7 @@ def rewrite_job_file(input_file, output_file, job):
 def get_workflow_from_invocation_id(invocation_id, galaxy_url, galaxy_api_key):
     user_gi = gi(url=galaxy_url, key=galaxy_api_key)
     workflow_id = user_gi.invocations.show_invocation(invocation_id)["workflow_id"]
-    workflow = user_gi.workflows._get(workflow_id, params={"instance": "true"})
+    workflow = get_dict_from_workflow(user_gi, workflow_id, instance=True)
     workflow_name = "-".join(workflow["name"].split())
     export_dict = user_gi.workflows.export_workflow_dict(workflow_id=workflow["id"], version=workflow["version"])
     with open(f"{workflow_name}.ga", "w") as workflow_out:

planemo/runnable.py

@@ -37,7 +37,6 @@
 )
 from planemo.galaxy.workflows import (
     describe_outputs,
-    GALAXY_WORKFLOWS_PREFIX,
     WorkflowOutput,
 )
 from planemo.io import error
@@ -115,7 +114,7 @@ def has_path(self):
 
     @property
     def is_remote_workflow_uri(self) -> bool:
-        return self.uri.startswith(GALAXY_WORKFLOWS_PREFIX)
+        return self.uri.startswith("gxid://")
 
     @property
     def test_data_search_path(self) -> str:

planemo/runnable_resolve.py

@@ -17,14 +17,14 @@ def for_runnable_identifier(ctx, runnable_identifier, kwds):
     # could be a URI, path, or alias
     current_profile = kwds.get("profile")
     runnable_identifier = translate_alias(ctx, runnable_identifier, current_profile)
-    if not runnable_identifier.startswith(GALAXY_WORKFLOWS_PREFIX):
+    if not runnable_identifier.startswith("gxid://"):
         runnable_identifier = uri_to_path(ctx, runnable_identifier)
     if os.path.exists(runnable_identifier):
         runnable = for_path(runnable_identifier)
     else:  # assume galaxy workflow or tool id
         if "/repos/" in runnable_identifier:
             runnable_identifier = f"{GALAXY_TOOLS_PREFIX}{runnable_identifier}"
-        elif not runnable_identifier.startswith(GALAXY_WORKFLOWS_PREFIX):
+        elif not runnable_identifier.startswith("gxid://"):
             runnable_identifier = f"{GALAXY_WORKFLOWS_PREFIX}{runnable_identifier}"
         runnable = for_uri(runnable_identifier)
     return runnable