opensource/galaxy
3
0
Fork 0
mirror of https://github.com/galaxyproject/galaxy.git synced 2026-09-24 16:30:27 +08:00
Code Issues Packages Projects Releases Wiki Activity

Workflow test utility build ds_map more easily.

Browse Source 
Probably should do something like this server side so workflow API can be used more deterministically (wouldn't need to import a workflow and then hit the API to know how to use it).
This commit is contained in:
John Chilton
2014-07-25 14:47:34 -05:00
parent 336aa4c31a
commit 2b78724d9b
1 changed files with 14 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test/api/test_workflows.py
+14 -12
View File
@@ -412,28 +412,30 @@ class WorkflowsApiTestCase( api.ApiTestCase ):
def _setup_workflow_run( self, workflow, history_id=None ):
uploaded_workflow_id = self.workflow_populator.create_workflow( workflow )
workflow_inputs = self._workflow_inputs( uploaded_workflow_id )
step_1 = step_2 = None
for key, value in workflow_inputs.iteritems():
label = value[ "label" ]
if label == "WorkflowInput1":
step_1 = key
if label == "WorkflowInput2":
step_2 = key
if not history_id:
history_id = self.dataset_populator.new_history()
hda1 = self.dataset_populator.new_dataset( history_id, content="1 2 3" )
hda2 = self.dataset_populator.new_dataset( history_id, content="4 5 6" )
label_map = {
'WorkflowInput1': self._ds_entry(hda1),
'WorkflowInput2': self._ds_entry(hda2)
}
workflow_request = dict(
history="hist_id=%s" % history_id,
workflow_id=uploaded_workflow_id,
ds_map=dumps( {
step_1: self._ds_entry(hda1),
step_2: self._ds_entry(hda2),
} ),
ds_map=self._build_ds_map( uploaded_workflow_id, label_map ),
)
return workflow_request, history_id
def _build_ds_map( self, workflow_id, label_map ):
workflow_inputs = self._workflow_inputs( workflow_id )
ds_map = {}
for key, value in workflow_inputs.iteritems():
label = value[ "label" ]
if label in label_map:
ds_map[ key ] = label_map[ label ]
return dumps( ds_map )
def _setup_random_x2_workflow( self, name ):
workflow = self.workflow_populator.load_random_x2_workflow( name )
uploaded_workflow_id = self.workflow_populator.create_workflow( workflow )