Merge pull request galaxyproject#21446 from mvdbeek/sample_sheet_column_fix

[25.1] Sample sheet column fix

Author: mvdbeek

lib/galaxy/model/__init__.py

@@ -8162,6 +8162,7 @@ def copy_to_collection(
             collection=collection,
             element_index=self.element_index,
             element_identifier=self.element_identifier,
+            columns=self.columns,
         )
         return new_element
 

lib/galaxy/model/deferred.py

@@ -372,6 +372,7 @@ def _materialize_collection_element(
         element=materialized_object,
         element_index=element.element_index,
         element_identifier=element.element_identifier,
+        columns=element.columns,
     )
     return materialized_element
 

lib/galaxy_test/api/test_dataset_collections.py

@@ -508,6 +508,73 @@ def test_sample_sheet_map_over_preserves_columns(self, history_id):
         ], f"Expected [3, 'control'], got {output_elements[2]['columns']}"
         assert output_elements[2]["element_identifier"] == "sample3"
 
+    def test_copy_sample_sheet_collection(self, history_id):
+        """Test that copying a sample sheet collection preserves columns metadata."""
+        # Create a sample sheet collection with columns metadata
+        contents = [
+            ("sample1", "content1"),
+            ("sample2", "content2"),
+        ]
+        sample_sheet_identifiers = self.dataset_collection_populator.list_identifiers(history_id, contents)
+        payload = dict(
+            name="original sample sheet",
+            instance_type="history",
+            history_id=history_id,
+            element_identifiers=sample_sheet_identifiers,
+            collection_type="sample_sheet",
+            column_definitions=[
+                {"type": "int", "name": "replicate", "optional": False},
+                {"type": "string", "name": "condition", "optional": False},
+            ],
+            rows={
+                "sample1": [1, "control"],
+                "sample2": [2, "treatment"],
+            },
+        )
+        create_response = self._post("dataset_collections", payload, json=True)
+        original_collection = self._check_create_response(create_response)
+        original_hdca_id = original_collection["id"]
+
+        # Verify the original sample sheet has columns metadata
+        original_elements = original_collection["elements"]
+        assert len(original_elements) == 2
+        assert original_elements[0]["columns"] == [1, "control"]
+        assert original_elements[1]["columns"] == [2, "treatment"]
+
+        # Copy the collection using the new copy_collection method
+        copy_response = self.dataset_collection_populator.copy_collection(
+            history_id, original_hdca_id, copy_elements=True, wait=False
+        )
+        copied_collection = copy_response.json()
+
+        # Fetch the full details of the copied collection
+        copied_collection_details = self.dataset_populator.get_history_collection_details(
+            history_id, content_id=copied_collection["id"]
+        )
+
+        # Verify the copied collection has the same columns metadata
+        copied_elements = copied_collection_details["elements"]
+        assert len(copied_elements) == 2, f"Expected 2 elements, got {len(copied_elements)}"
+
+        # Check that columns metadata was preserved for each element
+        self._assert_has_keys(copied_elements[0], "columns")
+        assert copied_elements[0]["columns"] == [
+            1,
+            "control",
+        ], f"Expected [1, 'control'], got {copied_elements[0]['columns']}"
+        assert copied_elements[0]["element_identifier"] == "sample1"
+
+        self._assert_has_keys(copied_elements[1], "columns")
+        assert copied_elements[1]["columns"] == [
+            2,
+            "treatment",
+        ], f"Expected [2, 'treatment'], got {copied_elements[1]['columns']}"
+        assert copied_elements[1]["element_identifier"] == "sample2"
+
+        # Verify column definitions are preserved
+        assert copied_collection_details["column_definitions"] == original_collection["column_definitions"]
+        assert copied_collection_details["collection_type"] == "sample_sheet"
+
     def test_workbook_download(self):
         xlsx_file = self.dataset_collection_populator.download_workbook(
             "sample_sheet",

lib/galaxy_test/base/populators.py

@@ -3401,6 +3401,20 @@ def create_list_in_history(self, history_id: str, wait: bool = False, **kwds):
         payload = self.create_list_payload(history_id, instance_type="history", **kwds)
         return self.__create(payload, wait=wait)
 
+    def copy_collection(self, history_id: str, hdca_id: str, copy_elements: bool = True, wait: bool = False):
+        """Copy an existing dataset collection to a history."""
+        payload = {
+            "source": "hdca",
+            "content": hdca_id,
+            "type": "dataset_collection",
+            "copy_elements": copy_elements,
+        }
+        copy_response = self.dataset_populator._post(
+            f"histories/{history_id}/contents/dataset_collections", payload, json=True
+        )
+        api_asserts.assert_status_code_is_ok(copy_response)
+        return copy_response
+
     def upload_collection(self, history_id: str, collection_type, elements, wait: bool = False, **kwds):
         payload = self.__create_payload_fetch(history_id, collection_type, contents=elements, **kwds)
         return self.__create(payload, wait=wait)