Add test for fail-fast flag

Author: mvdbeek

planemo/commands/cmd_workflow_test_on_invocation.py

@@ -33,7 +33,9 @@ def cli(ctx, path, invocation_id, test_index, **kwds):
         ), f"Selected test case {test_index}, but only found {len(test_cases)} test case(s)."
         test_case = test_cases[test_index - 1]
         # Hardcode fail_fast, no need to expose the option to the user IMO.
-        run_response = invocation_to_run_response(ctx, user_gi=config.user_gi, runnable=runnable, invocation=invocation, fail_fast=True)
+        run_response = invocation_to_run_response(
+            ctx, user_gi=config.user_gi, runnable=runnable, invocation=invocation, fail_fast=True
+        )
         structured_data = test_case.structured_test_data(run_response)
         test_data = {
             "version": "0.1",

planemo/galaxy/invocations/polling.py

@@ -36,6 +36,75 @@ def sleep(self):
         self.delta += self.polling_backoff
 
 
+def _summarize_invocation(invocation_api: InvocationApi, invocation_id: str):
+    invocation = invocation_api.get_invocation(invocation_id)
+    assert invocation
+    invocation_jobs = invocation_api.get_invocation_summary(invocation_id)
+    return invocation, invocation_jobs
+
+
+def _poll_main_workflow(
+    ctx,
+    invocation_id: str,
+    invocation_api: InvocationApi,
+    workflow_progress_display: WorkflowProgressDisplay,
+    fail_fast: bool,
+):
+    if workflow_progress_display.workflow_progress.terminal:
+        return None, None, None
+
+    try:
+        invocation, invocation_jobs = _summarize_invocation(invocation_api, invocation_id)
+        workflow_progress_display.handle_invocation(invocation, invocation_jobs)
+        return invocation, invocation_jobs, None
+    except Exception as e:
+        print(e)
+        return None, None, e
+
+
+def _poll_subworkflow(
+    ctx,
+    invocation_id: str,
+    invocation_api: InvocationApi,
+    workflow_progress_display: WorkflowProgressDisplay,
+    fail_fast: bool,
+):
+    if workflow_progress_display.all_subworkflows_complete():
+        return None, None, None
+
+    try:
+        subworkflow_id = workflow_progress_display.an_incomplete_subworkflow_id()
+        invocation, invocation_jobs = _summarize_invocation(invocation_api, subworkflow_id)
+        workflow_progress_display.handle_subworkflow_invocation(invocation, invocation_jobs)
+        return invocation, invocation_jobs, None
+    except Exception as e:
+        return None, None, e
+
+
+def _check_for_errors(
+    ctx,
+    invocation_id: str,
+    exception: Optional[Exception],
+    invocation,
+    invocation_jobs,
+    fail_fast: bool,
+):
+    error_message = workflow_in_error_message(
+        ctx, invocation_id, exception, invocation, invocation_jobs, fail_fast=fail_fast
+    )
+    if error_message:
+        final_state = "new" if not invocation else invocation["state"]
+        job_state = summary_job_state(invocation_jobs)
+        return final_state, job_state, error_message
+    return None
+
+
+def _is_polling_complete(workflow_progress_display: WorkflowProgressDisplay) -> bool:
+    return (
+        workflow_progress_display.workflow_progress.terminal and workflow_progress_display.all_subworkflows_complete()
+    )
+
+
 def wait_for_invocation_and_jobs(
     ctx,
     invocation_id: str,
@@ -46,80 +115,34 @@ def wait_for_invocation_and_jobs(
 ):
     ctx.vlog("Waiting for invocation [%s]" % invocation_id)
 
-    def summarize(invocation_id: str):
-        invocation = invocation_api.get_invocation(invocation_id)
-        assert invocation
-        invocation_jobs = invocation_api.get_invocation_summary(invocation_id)
-        return invocation, invocation_jobs
-
     last_invocation = None
     last_invocation_jobs = None
-    last_subworkflow_invocation = None
-    last_subworkflow_invocation_jobs = None
-    last_exception = None
     error_message: Optional[str] = None
 
-    done_polling = False
-    while not done_polling:
-        # loop over the main workflow and one subworkflow each iteration for display,
-
-        # skip the main workflow if it is already tracked as complete - if all steps have been
-        # scheduled there are no new subworkflow invocations to track.
-        if not workflow_progress_display.workflow_progress.terminal:
-            try:
-                last_invocation, last_invocation_jobs = summarize(invocation_id)
-                workflow_progress_display.handle_invocation(last_invocation, last_invocation_jobs)
-            except Exception as e:
-                print(e)
-                last_exception = e
-
-            error_message = workflow_in_error_message(
-                ctx,
-                invocation_id,
-                last_exception,
-                last_invocation,
-                last_invocation_jobs,
-                fail_fast=fail_fast,
-            )
-            if error_message:
-                final_invocation_state = "new" if not last_invocation else last_invocation["state"]
-                job_state = summary_job_state(last_invocation_jobs)
-                return final_invocation_state, job_state, error_message
-
-        assert last_invocation  # if we got here... the first check has passed and we have an invocation
-
-        # grab a subworkflow that isn't complete and check it, also register its subworkflow
-        # invocations so we catch all the children and children of children...
-        if not workflow_progress_display.all_subworkflows_complete():
-            try:
-                a_subworkflow_invocation_id = workflow_progress_display.an_incomplete_subworkflow_id()
-                last_subworkflow_invocation, last_subworkflow_invocation_jobs = summarize(a_subworkflow_invocation_id)
-                workflow_progress_display.handle_subworkflow_invocation(
-                    last_subworkflow_invocation, last_subworkflow_invocation_jobs
-                )
-            except Exception as e:
-                last_exception = e
-
-            error_message = workflow_in_error_message(
-                ctx,
-                invocation_id,
-                last_exception,
-                last_subworkflow_invocation,
-                last_subworkflow_invocation_jobs,
-                fail_fast=fail_fast,
-            )
-            if error_message:
-                final_invocation_state = (
-                    "new" if not last_subworkflow_invocation else last_subworkflow_invocation["state"]
-                )
-                job_state = summary_job_state(last_subworkflow_invocation_jobs)
-                return final_invocation_state, job_state, error_message
-
-        done_polling = (
-            workflow_progress_display.workflow_progress.terminal
-            and workflow_progress_display.all_subworkflows_complete()
+    while not _is_polling_complete(workflow_progress_display):
+        # Poll main workflow
+        main_invocation, main_jobs, main_exception = _poll_main_workflow(
+            ctx, invocation_id, invocation_api, workflow_progress_display, fail_fast
+        )
+
+        if main_invocation:
+            last_invocation = main_invocation
+            last_invocation_jobs = main_jobs
+
+        error_result = _check_for_errors(ctx, invocation_id, main_exception, main_invocation, main_jobs, fail_fast)
+        if error_result:
+            return error_result
+
+        # Poll subworkflow
+        sub_invocation, sub_jobs, sub_exception = _poll_subworkflow(
+            ctx, invocation_id, invocation_api, workflow_progress_display, fail_fast
         )
-        if not done_polling:
+
+        error_result = _check_for_errors(ctx, invocation_id, sub_exception, sub_invocation, sub_jobs, fail_fast)
+        if error_result:
+            return error_result
+
+        if not _is_polling_complete(workflow_progress_display):
             polling_tracker.sleep()
 
     ctx.vlog(f"The final state of all jobs and subworkflow invocations for invocation [{invocation_id}] is 'ok'")
@@ -129,7 +152,12 @@ def summarize(invocation_id: str):
 
 
 def workflow_in_error_message(
-    ctx, invocation_id, last_exception, last_invocation, last_invocation_jobs, fail_fast=False,
+    ctx,
+    invocation_id,
+    last_exception,
+    last_invocation,
+    last_invocation_jobs,
+    fail_fast=False,
 ) -> Optional[str]:
     """Return an error message if workflow is in an error state."""
 

planemo/options.py

@@ -1830,11 +1830,7 @@ def test_output_options():
 
 
 def test_options():
-    return _compose(
-        paste_test_data_paths_option(),
-        test_output_options(),
-        fail_fast_option()
-    )
+    return _compose(paste_test_data_paths_option(), test_output_options(), fail_fast_option())
 
 
 def _compose(*functions):

tests/test_invocation_polling.py

@@ -38,7 +38,7 @@ def sleep(self) -> None:
 
 
 def test_polling_scenario_1():
-    final_invocation_state, job_state, error_message = run_workflow_simulation(SCENARIO_1)
+    final_invocation_state, job_state, error_message = run_workflow_simulation(SCENARIO_1, fail_fast=True)
     assert final_invocation_state == "scheduled"
     assert job_state == "failed"
     assert error_message
@@ -75,6 +75,7 @@ def test_polling_without_display():
         invocation_api,
         polling_tracker,
         display,
+        fail_fast=True,
     )
     assert final_invocation_state == "scheduled"
     assert job_state == "failed"
@@ -109,7 +110,41 @@ def test_polling_without_invocation_as_full_subpanel():
     assert not error_message
 
 
-def run_workflow_simulation(yaml_str: str, display_configuration: Optional[DisplayConfiguration] = None):
+def test_fail_fast_enabled_with_job_failure():
+    """Test that fail_fast=True returns error when a job fails."""
+    final_invocation_state, job_state, error_message = run_workflow_simulation(SCENARIO_1, fail_fast=True)
+    # Invocation should still be scheduled (workflow scheduling succeeded)
+    assert final_invocation_state == "scheduled"
+    assert job_state == "failed"
+    # fail_fast should detect the failed job and return error message
+    assert error_message
+    assert "Failed to run workflow, at least one job is in [failed] state." in error_message
+
+
+def test_fail_fast_disabled_with_job_failure():
+    """Test that fail_fast=False does not report job failures as errors."""
+    final_invocation_state, job_state, error_message = run_workflow_simulation(SCENARIO_1, fail_fast=False)
+    # Invocation should be scheduled (workflow scheduling succeeded)
+    assert final_invocation_state == "scheduled"
+    assert job_state == "failed"
+    # Without fail_fast, job failures shouldn't cause error messages
+    # (unless invocation itself fails, which it doesn't in this case)
+    assert error_message is None
+
+
+def test_fail_fast_enabled_with_successful_workflow():
+    """Test that fail_fast=True works normally when no jobs fail."""
+    final_invocation_state, job_state, error_message = run_workflow_simulation(
+        SCENARIO_MULTIPLE_OK_SUBWORKFLOWS, fail_fast=True
+    )
+    assert final_invocation_state == "scheduled"
+    assert job_state == "ok"
+    assert not error_message
+
+
+def run_workflow_simulation(
+    yaml_str: str, display_configuration: Optional[DisplayConfiguration] = None, fail_fast: bool = False
+):
     simulation = parse_workflow_simulation_from_string(yaml_str)
     invocation_id = simulation.id
     invocation_api = SimulatedApi(simulation)
@@ -122,6 +157,7 @@ def run_workflow_simulation(yaml_str: str, display_configuration: Optional[Displ
             invocation_api,
             polling_tracker,
             display,
+            fail_fast=fail_fast,
         )