Use wait_for_invocation_and_completion instead of asking for state to be scheduled or completed

Author: mvdbeek

lib/galaxy_test/api/test_workflow_completion.py

@@ -73,11 +73,7 @@ def test_monitor_transitions_to_completed(self):
             )
 
             # Wait for monitor to detect completion and transition state
-            invocation = self.workflow_populator.wait_for_invocation_and_completion(summary.invocation_id, timeout=30)
-
-            # Verify the monitor transitioned to completed state
-            assert invocation["state"] == "completed"
-
+            self.workflow_populator.wait_for_invocation_and_completion(summary.invocation_id, timeout=30)
             # Verify completion record exists
             completion = self.workflow_populator.get_invocation_completion(summary.invocation_id)
             assert completion is not None, "Expected completion record to exist"
@@ -153,7 +149,4 @@ def test_completion_export_config_accepted(self):
 
             # Wait for monitor to process (completion transition)
             # (export hook will be skipped since no file source is configured)
-            invocation = self.workflow_populator.wait_for_invocation_and_completion(summary.invocation_id, timeout=30)
-
-            # Verify state transitioned to completed
-            assert invocation["state"] == "completed"
+            self.workflow_populator.wait_for_invocation_and_completion(summary.invocation_id, timeout=30)

lib/galaxy_test/api/test_workflows.py

@@ -1760,8 +1760,7 @@ def run_workflow_with_valid_hashes(
         invocation_id = self.workflow_populator.invoke_workflow_and_wait(workflow_id, request=workflow_request).json()[
             "id"
         ]
-        invocation = self._invocation_details(workflow_id, invocation_id)
-        assert invocation["state"] in ("scheduled", "completed"), invocation
+        self.workflow_populator.wait_for_invocation_and_completion(invocation_id)
         invocation_jobs = self.workflow_populator.get_invocation_jobs(invocation_id)
         for job in invocation_jobs:
             assert job["state"] == "ok"
@@ -1844,7 +1843,7 @@ def test_run_workflow_with_invalid_url(self):
     def test_run_workflow_with_url_collection(self):
         with self.dataset_populator.test_history() as history_id:
             invocation = self._run_multi_data_workflow(history_id)
-            assert invocation["state"] in ("scheduled", "completed"), invocation
+            assert invocation["state"] == "completed", invocation
             invocation_jobs = self.workflow_populator.get_invocation_jobs(invocation["id"])
             assert len(invocation_jobs) == 1
             job = invocation_jobs[0]
@@ -1911,7 +1910,9 @@ def _run_multi_data_workflow(self, history_id, invalid_hash=False):
         invocation_id = self.workflow_populator.invoke_workflow_and_wait(
             workflow_id, request=workflow_request, assert_ok=not invalid_hash
         ).json()["id"]
-        return self._invocation_details(workflow_id, invocation_id)
+        if invalid_hash:
+            return self._invocation_details(workflow_id, invocation_id)
+        return self.workflow_populator.wait_for_invocation_and_completion(invocation_id)
 
     @skip_without_tool("collection_paired_default")
     def test_run_workflow_with_url_paired_collection(self):
@@ -1974,8 +1975,7 @@ def test_run_workflow_with_url_paired_collection(self):
             invocation_id = self.workflow_populator.invoke_workflow_and_wait(
                 workflow_id, request=workflow_request
             ).json()["id"]
-            invocation = self._invocation_details(workflow_id, invocation_id)
-            assert invocation["state"] in ("scheduled", "completed"), invocation
+            invocation = self.workflow_populator.wait_for_invocation_and_completion(invocation_id)
             assert invocation["inputs"]["0"]["src"] == "hdca"
             input_hdca = self.dataset_populator.get_history_collection_details(
                 history_id=history_id, content_id=invocation["inputs"]["0"]["id"]
@@ -2131,8 +2131,7 @@ def __run_cat_workflow(self, inputs_by, history_id: Optional[str] = None):
         invocation_id = self.workflow_populator.invoke_workflow_and_wait(workflow_id, request=workflow_request).json()[
             "id"
         ]
-        invocation = self.workflow_populator.wait_for_invocation_and_completion(invocation_id)
-        assert invocation["state"] == "completed", invocation
+        self.workflow_populator.wait_for_invocation_and_completion(invocation_id)
 
     @skip_without_tool("collection_creates_pair")
     def test_workflow_run_output_collections(self) -> None:
@@ -3466,12 +3465,9 @@ def test_run_runtime_parameters_after_pause(self):
             # Review the paused steps to allow the workflow to continue.
             self.__review_paused_steps(uploaded_workflow_id, invocation_id, order_index=1, action=True)
 
-            # Wait for the workflow to finish scheduling and ensure both the invocation
+            # Wait for the workflow to finish and ensure both the invocation
             # and the history are in valid states.
-            invocation_scheduled = self._wait_for_invocation_state(
-                uploaded_workflow_id, invocation_id, ("scheduled", "completed")
-            )
-            assert invocation_scheduled, "Workflow state is not scheduled or completed..."
+            self.workflow_populator.wait_for_invocation_and_completion(invocation_id)
             self.dataset_populator.wait_for_history(history_id, assert_ok=True)
 
             content = self.dataset_populator.get_history_dataset_content(history_id)
@@ -5029,12 +5025,9 @@ def test_workflow_pause(self):
             # Review the paused steps to allow the workflow to continue.
             self.__review_paused_steps(uploaded_workflow_id, invocation_id, order_index=2, action=True)
 
-            # Wait for the workflow to finish scheduling and ensure both the invocation
+            # Wait for the workflow to finish and ensure both the invocation
             # and the history are in valid states.
-            invocation_scheduled = self._wait_for_invocation_state(
-                uploaded_workflow_id, invocation_id, ("scheduled", "completed")
-            )
-            assert invocation_scheduled, "Workflow state is not scheduled or completed..."
+            self.workflow_populator.wait_for_invocation_and_completion(invocation_id)
             self.dataset_populator.wait_for_history(history_id, assert_ok=True)
 
     @skip_without_tool("cat")
@@ -5092,9 +5085,7 @@ def test_workflow_map_reduce_pause(self):
             self._assert_invocation_non_terminal(uploaded_workflow_id, invocation_id)
 
             self.__review_paused_steps(uploaded_workflow_id, invocation_id, order_index=4, action=True)
-            self.workflow_populator.wait_for_invocation_and_jobs(history_id, uploaded_workflow_id, invocation_id)
-            invocation = self._invocation_details(uploaded_workflow_id, invocation_id)
-            assert invocation["state"] in ("scheduled", "completed")
+            self.workflow_populator.wait_for_invocation_and_completion(invocation_id)
             assert "reviewed\n1\nreviewed\n4\n" == self.dataset_populator.get_history_dataset_content(history_id)
 
     @skip_without_tool("cat")
@@ -5171,10 +5162,10 @@ def test_cancel_workflow_invocation_deletes_jobs(self):
                 history_id=history_id,
                 wait=False,
             )
-            # wait_for_invocation just waits until scheduling complete, not jobs or subworkflow invocations
-            self.workflow_populator.wait_for_invocation("null", summary.invocation_id, assert_ok=True)
-            invocation_before_cancellation = self.workflow_populator.get_invocation(summary.invocation_id)
-            assert invocation_before_cancellation["state"] in ("scheduled", "completed")
+            # Wait for workflow to complete before cancellation
+            invocation_before_cancellation = self.workflow_populator.wait_for_invocation_and_completion(
+                summary.invocation_id
+            )
             subworkflow_invocation_id = invocation_before_cancellation["steps"][2]["subworkflow_invocation_id"]
             self.workflow_populator.cancel_invocation(summary.invocation_id)
             self.workflow_populator.wait_for_invocation_and_jobs(
@@ -5245,8 +5236,7 @@ def test_workflow_warning_workflow_output_not_found(self, history_id):
             assert_ok=False,
             wait=True,
         )
-        invocation = self.workflow_populator.get_invocation(summary.invocation_id)
-        assert invocation["state"] in ("scheduled", "completed")
+        invocation = self.workflow_populator.wait_for_invocation_and_completion(summary.invocation_id)
         assert len(invocation["messages"]) == 1
         message = invocation["messages"][0]
         assert message["reason"] == "workflow_output_not_found"
@@ -5915,8 +5905,8 @@ def test_run_subworkflow_with_boolean_parameter_in_when_condition(self):
             summary = self._run_workflow(workflow, history_id=history_id, wait=True, assert_ok=True)
 
             # Verify parent workflow executed successfully
+            self.workflow_populator.wait_for_invocation_and_completion(summary.invocation_id)
             parent_invocation = self.workflow_populator.get_invocation(summary.invocation_id, step_details=True)
-            assert parent_invocation["state"] in ("scheduled", "completed")
 
             # Find the subworkflow step and get its invocation
             subworkflow_step = None

test/integration/test_workflow_completion_hooks.py

@@ -92,8 +92,7 @@ def test_export_to_file_source_on_completion(self):
             )
 
             # Wait for invocation to reach completed state
-            invocation = self.workflow_populator.wait_for_invocation_and_completion(summary.invocation_id, timeout=60)
-            assert invocation["state"] == "completed"
+            self.workflow_populator.wait_for_invocation_and_completion(summary.invocation_id, timeout=60)
 
             # Wait for the export file to appear (hook execution is async via Celery)
             export_path = os.path.join(self.root_dir, export_filename)