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

Add more complicated workflow run test case.

Browse Source 
Re-use test_data specification used by many of the API workflow tests and a workflow used by another test. Add abstractions and DOM element for mapping labeled test data to tool form inputs.
This commit is contained in:
John Chilton
2018-04-13 13:23:22 -04:00
parent 8f529d177a
commit aafd9de76b
6 changed files with 118 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test/api/test_workflows.py
+4 -61
View File
@@ -14,6 +14,7 @@ from base.populators import ( # noqa: I100
DatasetCollectionPopulator,
DatasetPopulator,
flakey,
load_data_dict,
skip_without_tool,
wait_on,
WorkflowPopulator
@@ -102,11 +103,8 @@ class BaseWorkflowsApiTestCase(api.ApiTestCase):
ds_map[key] = label_map[label]
return dumps(ds_map)
def _ds_entry(self, hda):
src = 'hda'
if 'history_content_type' in hda and hda['history_content_type'] == "dataset_collection":
src = 'hdca'
return dict(src=src, id=hda["id"])
def _ds_entry(self, history_content):
return self.dataset_populator.ds_entry(history_content)
def _workflow_inputs(self, uploaded_workflow_id):
workflow_show_resposne = self._get("workflows/%s" % uploaded_workflow_id)
@@ -137,62 +135,7 @@ class BaseWorkflowsApiTestCase(api.ApiTestCase):
jobs_descriptions = yaml.safe_load(has_workflow)
test_data = jobs_descriptions.get("test_data", {})
label_map = {}
inputs = {}
has_uploads = False
for key, value in test_data.items():
is_dict = isinstance(value, dict)
if is_dict and ("elements" in value or value.get("type", None) in ["list:paired", "list", "paired"]):
elements_data = value.get("elements", [])
elements = []
for element_data in elements_data:
identifier = element_data["identifier"]
input_type = element_data.get("type", "raw")
if input_type == "File":
content = read_test_data(element_data)
else:
content = element_data["content"]
elements.append((identifier, content))
# TODO: make this collection_type
collection_type = value["type"]
new_collection_kwds = {}
if "name" in value:
new_collection_kwds["name"] = value["name"]
if collection_type == "list:paired":
hdca = self.dataset_collection_populator.create_list_of_pairs_in_history(history_id, **new_collection_kwds).json()
elif collection_type == "list":
hdca = self.dataset_collection_populator.create_list_in_history(history_id, contents=elements, **new_collection_kwds).json()
else:
hdca = self.dataset_collection_populator.create_pair_in_history(history_id, contents=elements, **new_collection_kwds).json()
label_map[key] = self._ds_entry(hdca)
inputs[key] = hdca
has_uploads = True
elif is_dict and "type" in value:
input_type = value["type"]
if input_type == "File":
content = read_test_data(value)
new_dataset_kwds = {
"content": content
}
if "name" in value:
new_dataset_kwds["name"] = value["name"]
if "file_type" in value:
new_dataset_kwds["file_type"] = value["file_type"]
hda = self.dataset_populator.new_dataset(history_id, **new_dataset_kwds)
label_map[key] = self._ds_entry(hda)
has_uploads = True
elif input_type == "raw":
label_map[key] = value["value"]
inputs[key] = value["value"]
elif not is_dict:
has_uploads = True
hda = self.dataset_populator.new_dataset(history_id, content=value)
label_map[key] = self._ds_entry(hda)
inputs[key] = hda
else:
raise ValueError("Invalid test_data def %" % test_data)
inputs, label_map, has_uploads = load_data_dict(history_id, test_data, self.dataset_populator, self.dataset_collection_populator)
workflow_request = dict(
history="hist_id=%s" % history_id,
workflow_id=workflow_id,