Add test for copying sample sheet

This commit is contained in:
mvdbeek
2025-12-12 19:04:20 +01:00
parent 49d06524d1
commit 2534c3e9eb
3 changed files with 81 additions and 2 deletions
-2
View File
@@ -105,7 +105,6 @@ def derefence_collection_element(
element=child_dataset_collection,
element_identifier=element.identifier,
element_index=element_index,
columns=element.columns,
)
sa_session.add(child_dataset_collection)
for index, child_element in enumerate(element.elements):
@@ -136,7 +135,6 @@ def dereference_collection_dataset_element(
element=hda,
element_identifier=element.identifier,
element_index=element_index,
columns=element.columns,
)
parent_dataset_collection.elements.append(dce)
@@ -508,6 +508,73 @@ class TestDatasetCollectionsApi(ApiTestCase):
], 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",
+14
View File
@@ -3401,6 +3401,20 @@ class BaseDatasetCollectionPopulator:
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)