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

@@ -129,7 +129,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

@@ -109,7 +109,43 @@ 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 +158,7 @@ def run_workflow_simulation(yaml_str: str, display_configuration: Optional[Displ
             invocation_api,
             polling_tracker,
             display,
+            fail_fast=fail_fast,
         )