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

Test case to clarify workflow API a bit in simple cases.

Browse Source
This commit is contained in:
John Chilton
2018-08-22 10:45:06 -04:00
parent fc167c068c
commit 2d4b717aaf
1 changed files with 30 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test/api/test_workflows.py
+30 -8
View File
@@ -2902,15 +2902,37 @@ steps:
invocation_id = usage["id"]
usage_details = self._invocation_details(workflow_id, invocation_id)
# Assert some high-level things about the structure of data returned.
self._assert_has_keys(usage_details, "inputs", "steps")
self._assert_has_keys(usage_details, "inputs", "steps", "workflow_id")
# stored_workflow vs workflow thing makes following state false.
# assert usage_details["workflow_id"] == workflow_id
# Wait for the invocation to be fully scheduled, so we have details on all steps.
self._wait_for_invocation_state(workflow_id, invocation_id, 'scheduled')
usage_details = self._invocation_details(workflow_id, invocation_id)
invocation_steps = usage_details["steps"]
for step in invocation_steps:
self._assert_has_keys(step, "workflow_step_id", "order_index", "id")
an_invocation_step = invocation_steps[0]
step_id = an_invocation_step["id"]
step_response = self._get("workflows/%s/usage/%s/steps/%s" % (workflow_id, invocation_id, step_id))
self._assert_status_code_is(step_response, 200)
self._assert_has_keys(step_response.json(), "id", "order_index")
invocation_input_step, invocation_tool_step = None, None
for invocation_step in invocation_steps:
self._assert_has_keys(invocation_step, "workflow_step_id", "order_index", "id")
order_index = invocation_step["order_index"]
assert order_index in [0, 1, 2], order_index
if order_index == 0:
invocation_input_step = invocation_step
elif order_index == 2:
invocation_tool_step = invocation_step
# Tool steps have non-null job_ids (deprecated though they may be)
assert invocation_input_step.get("job_id", None) is None
job_id = invocation_tool_step.get("job_id", None)
assert job_id is not None
invocation_tool_step_id = invocation_tool_step["id"]
invocation_tool_step_response = self._get("workflows/%s/invocations/%s/steps/%s" % (workflow_id, invocation_id, invocation_tool_step_id))
self._assert_status_code_is(invocation_tool_step_response, 200)
self._assert_has_keys(invocation_tool_step_response.json(), "id", "order_index", "job_id")
assert invocation_tool_step_response.json()["job_id"] == job_id
@skip_without_tool("cat1")
def test_invocations_accessible_imported_workflow(self):