Handle $graph packed documents in guess_artifact_type()

lib/galaxy/tool_util/cwl/util.py

@@ -8,13 +8,17 @@ import json
 import os
 import tarfile
 import tempfile
+import urllib.parse
 from collections import namedtuple
 from typing import Any, List, Optional
 
 import yaml
 from typing_extensions import TypedDict
 
-from galaxy.util import unicodify
+from galaxy.util import (
+    str_removeprefix,
+    unicodify,
+)
 
 STORE_SECONDARY_FILES_WITH_BASENAME = True
 SECONDARY_FILES_EXTRA_PREFIX = "__secondary_files__"
@@ -594,15 +598,28 @@ def download_output(galaxy_output, get_metadata, get_dataset, get_extra_files, o
 
 
 def guess_artifact_type(path):
-    # TODO: Handle IDs within files.
     tool_or_workflow = "workflow"
-    try:
-        with open(path) as f:
-            artifact = yaml.safe_load(f)
+    path, object_id = urllib.parse.urldefrag(path)
+    with open(path) as f:
+        document = yaml.safe_load(f)
 
-        tool_or_workflow = "tool" if artifact["class"] != "Workflow" else "workflow"
+    if '$graph' in document:
+        # Packed document without a process object at the root
+        objects = document['$graph']
+        if not object_id:
+            object_id = 'main'  # default object id
 
-    except Exception as e:
-        print(e)
+        # Have to use str_removeprefix() instead of rstrip() because only the
+        # first '#' should be removed from the object id
+        matching_objects = [o for o in objects if str_removeprefix(o['id'], '#') == object_id]
+        if len(matching_objects) == 0:
+            raise Exception(f"No process object with id [{object_id}]")
+        if len(matching_objects) > 1:
+            raise Exception(f"Multiple process objects with id [{object_id}]")
+        object_ = matching_objects[0]
+    else:
+        object_ = document
+
+    tool_or_workflow = "tool" if object_["class"] != "Workflow" else "workflow"
 
     return tool_or_workflow

lib/galaxy/util/__init__.py

@@ -102,6 +102,17 @@ XML = etree.XML
 defaultdict = collections.defaultdict
 
 
+def str_removeprefix(s: str, prefix: str):
+    """
+    str.removeprefix() equivalent for Python < 3.9
+    """
+    if sys.version_info >= (3, 9):
+        return s.removeprefix(prefix)
+    if s.startswith(prefix):
+        return s[len(prefix):]
+    return s
+
+
 def remove_protocol_from_url(url):
     """ Supplied URL may be null, if not ensure http:// or https://
     etc... is stripped off.

lib/galaxy_test/base/populators.py

@@ -44,6 +44,7 @@ import random
 import string
 import time
 import unittest
+import urllib.parse
 from abc import ABCMeta, abstractmethod
 from functools import wraps
 from io import StringIO
@@ -330,9 +331,7 @@ class CwlPopulator:
         history_id: str,
         assert_ok: bool = True,
     ):
-        object_id = None
-        if "#" in workflow_path:
-            workflow_path, object_id = workflow_path.split("#", 1)
+        workflow_path, object_id = urllib.parse.urldefrag(workflow_path)
         workflow_id = self.workflow_populator.import_workflow_from_path(workflow_path, object_id)
 
         request = {