Merge pull request #17276 from mvdbeek/fix_nested_collection_serializ…

…ation_in_to_cwl

[23.1] Fix ``to_cwl`` for nested collections

lib/galaxy/workflow/modules.py

@@ -123,6 +123,8 @@ class ConditionalStepWhen(BooleanToolParameter):
 
 def to_cwl(value, hda_references, step):
     element_identifier = None
+    if isinstance(value, model.HistoryDatasetCollectionAssociation):
+        value = value.collection
     if isinstance(value, model.DatasetCollectionElement) and value.hda:
         element_identifier = value.element_identifier
         value = value.hda
@@ -152,14 +154,13 @@ def to_cwl(value, hda_references, step):
                 properties, value.dataset.created_from_basename or element_identifier or value.name
             )
             return properties
-    elif hasattr(value, "collection"):
-        collection = value.collection
-        if collection.collection_type == "list":
-            return [to_cwl(dce, hda_references=hda_references, step=step) for dce in collection.dataset_elements]
+    elif isinstance(value, model.DatasetCollection):
+        if value.collection_type == "list":
+            return [to_cwl(dce, hda_references=hda_references, step=step) for dce in value.dataset_elements]
         else:
             # Could be record or nested lists
             rval = {}
-            for element in collection.elements:
+            for element in value.elements:
                 rval[element.element_identifier] = to_cwl(
                     element.element_object, hda_references=hda_references, step=step
                 )

test/unit/workflows/test_modules.py

@@ -261,6 +261,19 @@ def test_to_cwl():
     assert hda_references == hdas
 
 
+def test_to_cwl_nested_collection():
+    hda = model.HistoryDatasetAssociation(create_dataset=True, flush=False)
+    hda.dataset.state = model.Dataset.states.OK
+    dc_inner = model.DatasetCollection(collection_type="list")
+    model.DatasetCollectionElement(collection=dc_inner, element_identifier="inner", element=hda)
+    dc_outer = model.DatasetCollection(collection_type="list:list")
+    model.DatasetCollectionElement(collection=dc_outer, element_identifier="outer", element=dc_inner)
+    hdca = model.HistoryDatasetCollectionAssociation(name="the collection", collection=dc_outer)
+    result = modules.to_cwl(hdca, [], model.WorkflowStep())
+    assert result["outer"][0]["class"] == "File"
+    assert result["outer"][0]["basename"] == "inner"
+
+
 class MapOverTestCase(NamedTuple):
     data_input: str
     step_input_def: Union[str, List[str]]