opensource/galaxy
3
0
Fork 0
mirror of https://github.com/galaxyproject/galaxy.git synced 2026-10-30 18:13:23 +08:00
Code Issues Packages Projects Releases Wiki Activity

Tests for preserving column definitions in different metadata collection strategies.

Browse Source
This commit is contained in:
John Chilton
2025-08-14 09:57:45 -04:00
parent d07165e8e8
commit ebd48c7516
3 changed files with 88 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/galaxy_test/api/test_dataset_collections.py
+53 -51
View File
@@ -676,7 +676,7 @@ class TestDatasetCollectionsApi(ApiTestCase):
"__files": {"files_0|file_data": open(self.test_data_resolver.get_filename("4.bed"))},
}
self.dataset_populator.fetch(payload)
hdca = self._assert_one_collection_created_in_history(history_id)
hdca = assert_one_collection_created_in_history(self.dataset_populator, history_id)
assert hdca["name"] == "Test upload"
hdca_tags = hdca["tags"]
assert len(hdca_tags) == 1
@@ -706,7 +706,7 @@ class TestDatasetCollectionsApi(ApiTestCase):
"__files": {"files_0|file_data": open(self.test_data_resolver.get_filename("4.bed"))},
}
self.dataset_populator.fetch(payload)
hdca = self._assert_one_collection_created_in_history(history_id)
hdca = assert_one_collection_created_in_history(self.dataset_populator, history_id)
assert hdca["name"] == "Test upload"
assert len(hdca["elements"]) == 1, hdca
element0 = hdca["elements"][0]
@@ -734,7 +734,7 @@ class TestDatasetCollectionsApi(ApiTestCase):
"targets": targets,
}
self.dataset_populator.fetch(payload)
hdca = self._assert_one_collection_created_in_history(history_id)
hdca = assert_one_collection_created_in_history(self.dataset_populator, history_id)
assert len(hdca["elements"]) == 1, hdca
element0 = hdca["elements"][0]
assert element0["element_identifier"] == "hello.txt"
@@ -762,7 +762,7 @@ class TestDatasetCollectionsApi(ApiTestCase):
"targets": targets,
}
self.dataset_populator.fetch(payload)
hdca = self._assert_one_collection_created_in_history(history_id)
hdca = assert_one_collection_created_in_history(self.dataset_populator, history_id)
assert len(hdca["elements"]) == 1, hdca
element0 = hdca["elements"][0]
assert element0["element_identifier"] == "4.bed"
@@ -786,50 +786,12 @@ class TestDatasetCollectionsApi(ApiTestCase):
"targets": targets,
}
self.dataset_populator.fetch(payload, assert_ok=False, wait=True)
hdca = self._assert_one_collection_created_in_history(history_id)
hdca = assert_one_collection_created_in_history(self.dataset_populator, history_id)
assert hdca["populated"] is False
assert "bagit.txt" in hdca["populated_state_message"], hdca
def test_upload_flat_sample_sheet(self):
column_definitions = [{"type": "int", "name": "replicate", "optional": False, "default_value": 0}]
with self.dataset_populator.test_history(require_new=False) as history_id:
elements = [
{
"src": "url",
"url": self.dataset_populator.base64_url_for_string("hello world"),
"info": "my cool hello world",
"name": "sample1",
"row": [42],
}
]
targets = [
{
"destination": {"type": "hdca"},
"elements": elements,
"collection_type": "sample_sheet",
"column_definitions": column_definitions,
}
]
payload = {
"history_id": history_id,
"targets": targets,
}
self.dataset_populator.fetch(payload)
hdca = self._assert_one_collection_created_in_history(history_id)
assert len(hdca["elements"]) == 1, hdca
element0 = hdca["elements"][0]
assert element0["element_identifier"] == "sample1"
assert element0["columns"][0] == 42
object0 = element0["object"]
assert object0["state"] == "ok"
assert hdca["column_definitions"] is not None
hdca_id = hdca["id"]
dataset_collection_response = self._get(f"dataset_collections/{hdca_id}")
self._assert_status_code_is(dataset_collection_response, 200)
collection = dataset_collection_response.json()
assert collection["collection_type"] == "sample_sheet"
assert collection["column_definitions"] is not None
upload_flat_sample_sheet(self.dataset_populator)
def test_upload_sample_sheet_paired(self):
column_definitions = [{"type": "int", "name": "replicate", "optional": False, "default_value": 0}]
@@ -867,23 +829,19 @@ class TestDatasetCollectionsApi(ApiTestCase):
"targets": targets,
}
self.dataset_populator.fetch(payload)
hdca = self._assert_one_collection_created_in_history(history_id)
hdca = assert_one_collection_created_in_history(self.dataset_populator, history_id)
assert len(hdca["elements"]) == 1, hdca
element0 = hdca["elements"][0]
assert element0["element_identifier"] == "sample1"
assert element0["columns"][0] == 42
def _assert_one_collection_created_in_history(self, history_id: str):
contents_response = self._get(f"histories/{history_id}/contents/dataset_collections")
self._assert_status_code_is(contents_response, 200)
contents = contents_response.json()
contents = self.dataset_populator.get_history_contents_of_type(history_id, "dataset_collections")
assert len(contents) == 1
hdca = contents[0]
assert hdca["history_content_type"] == "dataset_collection"
hdca_id = hdca["id"]
collection_response = self._get(f"histories/{history_id}/contents/dataset_collections/{hdca_id}")
self._assert_status_code_is(collection_response, 200)
return collection_response.json()
return self.dataset_populator.get_history_collection_details(history_id, content_id=hdca_id)
def _check_create_response(self, create_response):
self._assert_status_code_is(create_response, 200)
@@ -1156,3 +1114,47 @@ class TestDatasetCollectionsApi(ApiTestCase):
assert "contents_url" in matches[0]
return matches[0]["contents_url"]
def upload_flat_sample_sheet(dataset_populator: DatasetPopulator):
column_definitions = [{"type": "int", "name": "replicate", "optional": False, "default_value": 0}]
with dataset_populator.test_history(require_new=False) as history_id:
elements = [
{
"src": "url",
"url": dataset_populator.base64_url_for_string("hello world"),
"info": "my cool hello world",
"name": "sample1",
"row": [42],
}
]
targets = [
{
"destination": {"type": "hdca"},
"elements": elements,
"collection_type": "sample_sheet",
"column_definitions": column_definitions,
}
]
payload = {
"history_id": history_id,
"targets": targets,
}
dataset_populator.fetch(payload)
hdca = assert_one_collection_created_in_history(dataset_populator, history_id)
assert len(hdca["elements"]) == 1, hdca
element0 = hdca["elements"][0]
assert element0["element_identifier"] == "sample1"
assert element0["columns"][0] == 42
object0 = element0["object"]
assert object0["state"] == "ok"
assert hdca["column_definitions"] is not None
def assert_one_collection_created_in_history(dataset_populator: DatasetPopulator, history_id: str):
contents = dataset_populator.get_history_contents_of_type(history_id, "dataset_collections")
assert len(contents) == 1
hdca = contents[0]
assert hdca["history_content_type"] == "dataset_collection"
hdca_id = hdca["id"]
return dataset_populator.get_history_collection_details(history_id, content_id=hdca_id)
lib/galaxy_test/base/populators.py
+7
View File
@@ -147,6 +147,8 @@ SKIP_FLAKEY_TESTS_ON_ERROR = os.environ.get("GALAXY_TEST_SKIP_FLAKEY_TESTS_ON_ER
PRIVATE_ROLE_TYPE = "private"
HistoryContentsType = Literal["dataset_collections", "datasets"]
def flakey(method):
@wraps(method)
@@ -1358,6 +1360,11 @@ class BaseDatasetPopulator(BasePopulator):
contents_response.raise_for_status()
return contents_response.json()
def get_history_contents_of_type(self, history_id: str, content_type: str) -> list[dict[str, Any]]:
contents_response = self._get_contents_request(history_id, f"/{content_type}")
api_asserts.assert_status_code_is_ok(contents_response)
return contents_response.json()
def _get_contents_request(self, history_id: str, suffix: str = "", data=None) -> Response:
if data is None:
data = {}
test/integration/test_upload_configuration_options.py
+28
View File
@@ -31,6 +31,9 @@ from requests import Response
from galaxy.tool_util.verify.test_data import TestDataResolver
from galaxy.util.unittest import TestCase
from galaxy_test.api.test_dataset_collections import (
upload_flat_sample_sheet,
)
from galaxy_test.base.api_util import TEST_USER
from galaxy_test.base.constants import (
ONE_TO_SIX_ON_WINDOWS,
@@ -983,3 +986,28 @@ class TestLinkDataUploadExtendedMetadata(BaseUploadContentConfigurationTestCase)
def test_link_data_only(self) -> None:
link_data_only(self.server_dir(), self.library_populator)
# With the API tests all using celery metadata - I think subtle differences in older
# but perfectly valid configurations for running metadata are not being tested by API
# tests anymore.
class TestUploadWithDirectoryMetadata(BaseUploadContentConfigurationTestCase):
@classmethod
def handle_galaxy_config_kwds(cls, config) -> None:
super().handle_galaxy_config_kwds(config)
config["metadata_strategy"] = "directory"
def test_upload_flat_sample_sheet(self):
upload_flat_sample_sheet(self.dataset_populator)
class TestUploadWithExtendedMetadata(BaseUploadContentConfigurationTestCase):
@classmethod
def handle_galaxy_config_kwds(cls, config) -> None:
super().handle_galaxy_config_kwds(config)
config["metadata_strategy"] = "extended"
def test_upload_flat_sample_sheet(self):
upload_flat_sample_sheet(self.dataset_populator)