Refactor workflow testing toward reuse via "populators".

Should allow more functionality to be shared between vanilla API tests and integration tests for various workflow scheduling options.
galaxyproject · Nov 30, 2017 · c076b3b · c076b3b
1 parent 8c1658f
commit c076b3b
Show file tree

Hide file tree

Showing 2 changed files with 17 additions and 14 deletions.
diff --git a/test/api/test_workflows.py b/test/api/test_workflows.py
@@ -2040,19 +2040,8 @@ def __assert_lines_hid_line_count_is(self, history, hid, lines):
         self._assert_status_code_is(hda_info_response, 200)
         self.assertEqual(hda_info_response.json()["metadata_data_lines"], lines)
 
-    def __invoke_workflow(self, history_id, workflow_id, inputs={}, request={}, assert_ok=True):
-        request["history"] = "hist_id=%s" % history_id,
-        if inputs:
-            request["inputs"] = dumps(inputs)
-            request["inputs_by"] = 'step_index'
-        url = "workflows/%s/usage" % (workflow_id)
-        invocation_response = self._post(url, data=request)
-        if assert_ok:
-            self._assert_status_code_is(invocation_response, 200)
-            invocation_id = invocation_response.json()["id"]
-            return invocation_id
-        else:
-            return invocation_response
+    def __invoke_workflow(self, *args, **kwds):
+        return self.workflow_populator.invoke_workflow(*args, **kwds)
 
     def __import_workflow(self, workflow_id, deprecated_route=False):
         if deprecated_route:

diff --git a/test/base/populators.py b/test/base/populators.py
@@ -383,7 +383,21 @@ def wait_for_workflow(self, workflow_id, invocation_id, history_id, assert_ok=Tr
         """ Wait for a workflow invocation to completely schedule and then history
         to be complete. """
         self.wait_for_invocation(workflow_id, invocation_id, timeout=timeout)
-        self.dataset_populator.wait_for_history(history_id, assert_ok=assert_ok, timeout=timeout)
+        self.dataset_populator.wait_for_history_jobs(history_id, assert_ok=assert_ok, timeout=timeout)
+
+    def invoke_workflow(self, history_id, workflow_id, inputs={}, request={}, assert_ok=True):
+        request["history"] = "hist_id=%s" % history_id,
+        if inputs:
+            request["inputs"] = json.dumps(inputs)
+            request["inputs_by"] = 'step_index'
+        url = "workflows/%s/usage" % (workflow_id)
+        invocation_response = self._post(url, data=request)
+        if assert_ok:
+            api_asserts.assert_status_code_is(invocation_response, 200)
+            invocation_id = invocation_response.json()["id"]
+            return invocation_id
+        else:
+            return invocation_response
 
 
 class WorkflowPopulator(BaseWorkflowPopulator, ImporterGalaxyInterface):