Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
history_id = self.gi.histories.create_history(name="TestDSListCreate")["id"]
dataset1_id = self._test_dataset(history_id)
dataset2_id = self._test_dataset(history_id)
dataset3_id = self._test_dataset(history_id)
dataset4_id = self._test_dataset(history_id)
collection_response = self.gi.histories.create_dataset_collection(
history_id=history_id,
collection_description=collections.CollectionDescription(
name="MyListOfPairedDatasets",
type="list:paired",
elements=[
collections.CollectionElement(
name="sample1",
type="paired",
elements=[
collections.HistoryDatasetElement(name="forward", id=dataset1_id),
collections.HistoryDatasetElement(name="reverse", id=dataset2_id),
]
),
collections.CollectionElement(
name="sample2",
type="paired",
elements=[
collections.HistoryDatasetElement(name="forward", id=dataset3_id),
collections.HistoryDatasetElement(name="reverse", id=dataset4_id),
]
),
]
)
)
self.assertEqual(collection_response["name"], "MyListOfPairedDatasets")
self.assertEqual(collection_response["collection_type"], "list:paired")
def test_run_workflow_with_dataset_collection(self):
dataset1 = self.hist.paste_content(FOO_DATA)
dataset2 = self.hist.paste_content(FOO_DATA_2)
collection_description = dataset_collections.CollectionDescription(
name="MyDatasetList",
elements=[
dataset_collections.HistoryDatasetElement(name="sample1", id=dataset1.id),
dataset_collections.HistoryDatasetElement(name="sample2", id=dataset2.id),
]
)
dataset_collection = self.hist.create_dataset_collection(collection_description)
input_map = {"Input Dataset Collection": dataset_collection,
"Input 2": dataset1}
outputs, out_hist = self.wf.run(input_map, self.hist, wait=True)
self.assertEqual(len(outputs), 1)
out_hdca = outputs[0]
self.assertIsInstance(out_hdca, wrappers.HistoryDatasetCollectionAssociation)
self.assertEqual(out_hdca.collection_type, 'list')
self.assertEqual(len(out_hdca.elements), 2)
self.assertEqual(out_hist.id, self.hist.id)
def _create_pair_in_history(self, history_id):
dataset1_id = self._test_dataset(history_id)
dataset2_id = self._test_dataset(history_id)
collection_response = self.gi.histories.create_dataset_collection(
history_id=history_id,
collection_description=collections.CollectionDescription(
name="MyTestPair",
type="paired",
elements=[
collections.HistoryDatasetElement(name="forward", id=dataset1_id),
collections.HistoryDatasetElement(name="reverse", id=dataset2_id),
]
)
)
return collection_response
def _create_collection_description(self):
self.dataset1 = self.hist.paste_content(FOO_DATA)
self.dataset2 = self.hist.paste_content(FOO_DATA_2)
self.collection_description = dataset_collections.CollectionDescription(
name="MyDatasetList",
elements=[
dataset_collections.HistoryDatasetElement(name="sample1", id=self.dataset1.id),
dataset_collections.HistoryDatasetElement(name="sample2", id=self.dataset2.id),
]
def _create_collection_description(self):
self.dataset1 = self.hist.paste_content(FOO_DATA)
self.dataset2 = self.hist.paste_content(FOO_DATA_2)
self.collection_description = dataset_collections.CollectionDescription(
name="MyDatasetList",
elements=[
dataset_collections.HistoryDatasetElement(name="sample1", id=self.dataset1.id),
dataset_collections.HistoryDatasetElement(name="sample2", id=self.dataset2.id),
]
okDatasets = [d for d in failedCollection['elements'] if d['object']['state'] == 'ok' and d['object']['file_size'] > 0]
notOkDatasets = [d for d in failedCollection['elements'] if d['object']['state'] != 'ok' or d['object']['file_size'] == 0]
okCollectionName = failedCollection['name'] + " (ok)"
notOkCollectionName = failedCollection['name'] + " (not ok)"
gi.histories.create_dataset_collection(
history_id=historyId,
collection_description=collections.CollectionDescription(
name=okCollectionName,
elements=[collections.HistoryDatasetElement(d['object']['name'], d['object']['id']) for d in okDatasets]))
gi.histories.create_dataset_collection(
history_id=historyId,
collection_description=collections.CollectionDescription(
name=notOkCollectionName,
elements=[collections.HistoryDatasetElement(d['object']['name'], d['object']['id']) for d in notOkDatasets]))