mirror of
https://github.com/galaxyproject/galaxy.git
synced 2026-09-24 16:30:27 +08:00
Refactor populators for easier nested collection creation.
This commit is contained in:
@@ -64,19 +64,13 @@ class DatasetCollectionApiTestCase(api.ApiTestCase):
|
||||
assert len(returned_collections) == 1, dataset_collection
|
||||
|
||||
def test_create_list_of_new_pairs(self):
|
||||
pair_identifiers = self.dataset_collection_populator.pair_identifiers(self.history_id)
|
||||
element_identifiers = [dict(
|
||||
src="new_collection",
|
||||
name="test_pair",
|
||||
collection_type="paired",
|
||||
element_identifiers=pair_identifiers,
|
||||
)]
|
||||
identifiers = self.dataset_collection_populator.nested_collection_identifiers(self.history_id, "list:paired")
|
||||
payload = dict(
|
||||
collection_type="list:paired",
|
||||
instance_type="history",
|
||||
history_id=self.history_id,
|
||||
name="a nested collection",
|
||||
element_identifiers=json.dumps(element_identifiers),
|
||||
element_identifiers=json.dumps(identifiers),
|
||||
)
|
||||
create_response = self._post("dataset_collections", payload)
|
||||
dataset_collection = self._check_create_response(create_response)
|
||||
@@ -85,7 +79,9 @@ class DatasetCollectionApiTestCase(api.ApiTestCase):
|
||||
returned_collections = dataset_collection["elements"]
|
||||
assert len(returned_collections) == 1, dataset_collection
|
||||
pair_1_element = returned_collections[0]
|
||||
self._assert_has_keys(pair_1_element, "element_index")
|
||||
self._assert_has_keys(pair_1_element, "element_identifier", "element_index", "object")
|
||||
assert pair_1_element["element_identifier"] == "test_level_1", pair_1_element
|
||||
assert pair_1_element["element_index"] == 0, pair_1_element
|
||||
pair_1_object = pair_1_element["object"]
|
||||
self._assert_has_keys(pair_1_object, "collection_type", "elements", "element_count")
|
||||
self.assertEquals(pair_1_object["collection_type"], "paired")
|
||||
|
||||
+40
-10
@@ -560,14 +560,42 @@ class BaseDatasetCollectionPopulator(object):
|
||||
collection_type='list:paired',
|
||||
name=name)
|
||||
|
||||
def create_nested_collection(self, history_id, collection, collection_type, name):
|
||||
element_identifiers = []
|
||||
for i, pair in enumerate(collection):
|
||||
element_identifiers.append(dict(
|
||||
name="test%d" % i,
|
||||
src="hdca",
|
||||
id=pair
|
||||
))
|
||||
def nested_collection_identifiers(self, history_id, collection_type):
|
||||
rank_types = list(reversed(collection_type.split(":")))
|
||||
assert len(rank_types) > 0
|
||||
rank_type_0 = rank_types[0]
|
||||
if rank_type_0 == "list":
|
||||
identifiers = self.list_identifiers(history_id)
|
||||
else:
|
||||
identifiers = self.pair_identifiers(history_id)
|
||||
nested_collection_type = rank_type_0
|
||||
|
||||
for i, rank_type in enumerate(reversed(rank_types[1:])):
|
||||
name = "test_level_%d" % (i + 1) if rank_type == "list" else "paired"
|
||||
identifiers = [dict(
|
||||
src="new_collection",
|
||||
name=name,
|
||||
collection_type=nested_collection_type,
|
||||
element_identifiers=identifiers,
|
||||
)]
|
||||
nested_collection_type = "%s:%s" % (rank_type, nested_collection_type)
|
||||
return identifiers
|
||||
|
||||
def create_nested_collection(self, history_id, collection_type, name=None, collection=None, element_identifiers=None):
|
||||
"""Create a nested collection either from collection or using collection_type)."""
|
||||
assert collection_type is not None
|
||||
name = name or "Test %s" % collection_type
|
||||
if collection is not None:
|
||||
assert element_identifiers is None
|
||||
element_identifiers = []
|
||||
for i, pair in enumerate(collection):
|
||||
element_identifiers.append(dict(
|
||||
name="test%d" % i,
|
||||
src="hdca",
|
||||
id=pair
|
||||
))
|
||||
if element_identifiers is None:
|
||||
element_identifiers = self.nested_collection_identifiers(history_id, collection_type)
|
||||
|
||||
payload = dict(
|
||||
instance_type="history",
|
||||
@@ -583,6 +611,8 @@ class BaseDatasetCollectionPopulator(object):
|
||||
return self.create_list_from_pairs(history_id, [pair1])
|
||||
|
||||
def create_list_of_list_in_history(self, history_id, **kwds):
|
||||
# create_nested_collection will generate nested collection from just datasets,
|
||||
# this function uses recursive generation of history hdcas.
|
||||
collection_type = kwds.pop('collection_type', 'list:list')
|
||||
collection_types = collection_type.split(':')
|
||||
list = self.create_list_in_history(history_id, **kwds).json()['id']
|
||||
@@ -590,9 +620,9 @@ class BaseDatasetCollectionPopulator(object):
|
||||
for collection_type in collection_types[1:]:
|
||||
current_collection_type = "%s:%s" % (current_collection_type, collection_type)
|
||||
response = self.create_nested_collection(history_id=history_id,
|
||||
collection=[list],
|
||||
collection_type=current_collection_type,
|
||||
name=current_collection_type)
|
||||
name=current_collection_type,
|
||||
collection=[list])
|
||||
list = response.json()['id']
|
||||
return response
|
||||
|
||||
|
||||
Reference in New Issue
Block a user