diff --git a/CHANGELOG.md b/CHANGELOG.md index 2b6cd4c8b..ab831fe09 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,13 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). ### Fixed +- **Parallel Behave Runner Log Noise Reduction** (#8351): The parallel behave + runner now suppresses captured stdout/stderr for passing worker chunks and + only replays diagnostics for failed, errored, or crashed chunks. This makes + failure output significantly easier to spot in CI and local runs. A worker + crash (unhandled exception) is detected via an all-zero summary and the + captured traceback is always surfaced. + - **Automation Profile Silent Fallback** (#8232): `_resolve_profile_for_plan` in `PlanLifecycleService` now raises a clear `ValidationError` when a plan's automation profile name is not a known built-in profile, instead of silently diff --git a/features/behave_parallel_log_filtering.feature b/features/behave_parallel_log_filtering.feature new file mode 100644 index 000000000..547cf1a97 --- /dev/null +++ b/features/behave_parallel_log_filtering.feature @@ -0,0 +1,101 @@ +@mock_only +Feature: Behave-parallel conditional log replay + The parallel behave runner should only replay captured stdout/stderr + for worker chunks that failed, errored, or crashed. Passing chunks + must suppress their captured output to reduce CI noise. + + # ---- _chunk_has_failures helper ---- + + Scenario: chunk_has_failures returns False for all-passing summary + Given behave_parallel a summary with 3 passed features and 5 passed scenarios + When behave_parallel I check chunk_has_failures + Then behave_parallel chunk_has_failures should return false + + Scenario: chunk_has_failures returns True when features have failures + Given behave_parallel a summary with 1 failed feature + When behave_parallel I check chunk_has_failures + Then behave_parallel chunk_has_failures should return true + + Scenario: chunk_has_failures returns True when features have errors + Given behave_parallel a summary with 1 errored feature + When behave_parallel I check chunk_has_failures + Then behave_parallel chunk_has_failures should return true + + Scenario: chunk_has_failures returns True when scenarios have failures + Given behave_parallel a summary with 1 failed scenario + When behave_parallel I check chunk_has_failures + Then behave_parallel chunk_has_failures should return true + + Scenario: chunk_has_failures returns True when scenarios have errors + Given behave_parallel a summary with 1 errored scenario + When behave_parallel I check chunk_has_failures + Then behave_parallel chunk_has_failures should return true + + # ---- _chunk_no_scenarios_ran helper ---- + + Scenario: chunk_no_scenarios_ran returns True for empty summary + Given behave_parallel an empty summary + When behave_parallel I check chunk_no_scenarios_ran + Then behave_parallel chunk_no_scenarios_ran should return true + + Scenario: chunk_no_scenarios_ran returns False when scenarios passed + Given behave_parallel a summary with 3 passed features and 5 passed scenarios + When behave_parallel I check chunk_no_scenarios_ran + Then behave_parallel chunk_no_scenarios_ran should return false + + Scenario: chunk_no_scenarios_ran returns False when all scenarios are skipped + Given behave_parallel a summary with all scenarios skipped + When behave_parallel I check chunk_no_scenarios_ran + Then behave_parallel chunk_no_scenarios_ran should return false + + # ---- Conditional log replay integration ---- + + Scenario: Passing chunk output is suppressed in parallel aggregation + Given behave_parallel a worker result with passing summary and captured stdout "PASS OUTPUT" + When behave_parallel I aggregate the worker results + Then behave_parallel the aggregated stdout should not contain "PASS OUTPUT" + + Scenario: Failed chunk output is replayed in parallel aggregation + Given behave_parallel a worker result with failed summary and captured stdout "FAIL OUTPUT" + When behave_parallel I aggregate the worker results + Then behave_parallel the aggregated stdout should contain "FAIL OUTPUT" + + Scenario: Crashed worker output is replayed in parallel aggregation + Given behave_parallel a worker result with crashed summary and captured stderr "WORKER CRASH" + When behave_parallel I aggregate the worker results + Then behave_parallel the aggregated stderr should contain "WORKER CRASH" + + Scenario: Pure empty summary chunk triggers log replay via no-scenarios-ran path + Given behave_parallel a worker result with empty summary and captured stderr "EMPTY CRASH" + When behave_parallel I aggregate the worker results + Then behave_parallel the aggregated stderr should contain "EMPTY CRASH" + + Scenario: Failed chunk stderr is replayed in parallel aggregation + Given behave_parallel a worker result with failed summary and captured stderr "FAIL STDERR" + When behave_parallel I aggregate the worker results + Then behave_parallel the aggregated stderr should contain "FAIL STDERR" + + Scenario: Mixed results replay only failed chunks + Given behave_parallel worker results with one passing and one failing chunk + When behave_parallel I aggregate the worker results + Then behave_parallel the aggregated stdout should contain "FAILING CHUNK" + And behave_parallel the aggregated stdout should not contain "PASSING CHUNK" + + Scenario: TDD-inverted chunk suppresses output when worker_failed but summary is passing + Given behave_parallel a worker result where worker_failed is true but summary is passing with stdout "TDD OUTPUT" + When behave_parallel I aggregate the worker results + Then behave_parallel the aggregated stdout should not contain "TDD OUTPUT" + + # ---- Worker exception handling ---- + + Scenario: Worker exception returns failed summary with traceback + Given behave_parallel a worker payload that will raise an exception + When behave_parallel I invoke the worker + Then behave_parallel the worker result should indicate failure + And behave_parallel the worker stderr should contain "WORKER CRASH:" + And behave_parallel the worker summary should have features errors equal to 1 + + Scenario: Partial worker crash alongside passing workers produces summary with failures + Given behave_parallel worker results with one crashed chunk and one passing chunk + When behave_parallel I aggregate the worker results + Then behave_parallel the aggregated summary should have failures diff --git a/features/steps/behave_parallel_log_filtering_steps.py b/features/steps/behave_parallel_log_filtering_steps.py new file mode 100644 index 000000000..6eeef6a92 --- /dev/null +++ b/features/steps/behave_parallel_log_filtering_steps.py @@ -0,0 +1,359 @@ +"""Step definitions for behave_parallel_log_filtering.feature. + +Tests the conditional log replay and worker exception handling in +``scripts/run_behave_parallel.py``. +""" + +from __future__ import annotations + +import importlib.util +import io +import sys +from contextlib import redirect_stderr, redirect_stdout +from pathlib import Path +from types import ModuleType + +from behave import given, then, when # type: ignore[import-untyped] +from behave.runner import Context # type: ignore[import-untyped] + + +def _load_runner_module() -> ModuleType: + """Load scripts/run_behave_parallel.py as a module via importlib. + + **CWD requirement**: This function resolves the script path relative to + the current working directory (``Path("scripts") / "run_behave_parallel.py"``). + It must be called from the repository root (i.e. the directory that + contains the ``scripts/`` folder). Behave sets the working directory to + the project root when running feature tests, so this is satisfied + automatically in the normal test invocation. If you invoke step + definitions from an unexpected directory, either adjust your working + directory or anchor the path relative to ``__file__``. + """ + script_path = Path("scripts") / "run_behave_parallel.py" + spec = importlib.util.spec_from_file_location( + "run_behave_parallel", str(script_path) + ) + if spec is None or spec.loader is None: + msg = f"Cannot load {script_path}" + raise ImportError(msg) + mod = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = mod # Prevent re-execution on subsequent calls + spec.loader.exec_module(mod) + return mod + + +_runner_mod = _load_runner_module() + +# Bind module-level references for convenience. +Summary = _runner_mod.Summary +_chunk_has_failures = _runner_mod._chunk_has_failures +_chunk_no_scenarios_ran = _runner_mod._chunk_no_scenarios_ran +_empty_summary = _runner_mod._empty_summary +_worker_run_features = _runner_mod._worker_run_features +_aggregate_worker_results = _runner_mod._aggregate_worker_results +_has_failures = _runner_mod._has_failures + + +def _passing_summary(features: int = 1, scenarios: int = 1) -> Summary: + """Build a summary with all-passing counts.""" + return { + "features": { + "passed": features, + "failed": 0, + "errors": 0, + "skipped": 0, + }, + "scenarios": { + "passed": scenarios, + "failed": 0, + "errors": 0, + "skipped": 0, + }, + "steps": {"passed": scenarios * 3, "failed": 0, "errors": 0, "skipped": 0}, + "duration": 1.0, + } + + +def _failed_feature_summary() -> Summary: + s = _empty_summary() + s["features"]["failed"] = 1 + return s + + +def _errored_feature_summary() -> Summary: + s = _empty_summary() + s["features"]["errors"] = 1 + return s + + +def _failed_scenario_summary() -> Summary: + s = _empty_summary() + s["scenarios"]["failed"] = 1 + return s + + +def _errored_scenario_summary() -> Summary: + s = _empty_summary() + s["scenarios"]["errors"] = 1 + return s + + +def _all_skipped_summary() -> Summary: + """Build a summary where all scenarios are skipped (but some ran).""" + s = _empty_summary() + s["features"]["skipped"] = 1 + s["scenarios"]["skipped"] = 3 + return s + + +def _crash_summary() -> Summary: + """Build a summary representing a worker crash (features.errors = 1).""" + s = _empty_summary() + s["features"]["errors"] = 1 + return s + + +# ---- _chunk_has_failures helper steps ---- + + +@given( + "behave_parallel a summary with {n:d} passed features and {m:d} passed scenarios" +) +def step_passing_summary(context: Context, n: int, m: int) -> None: + context.bp_summary = _passing_summary(features=n, scenarios=m) + + +@given("behave_parallel a summary with 1 failed feature") +def step_failed_feature_summary(context: Context) -> None: + context.bp_summary = _failed_feature_summary() + + +@given("behave_parallel a summary with 1 errored feature") +def step_errored_feature_summary(context: Context) -> None: + context.bp_summary = _errored_feature_summary() + + +@given("behave_parallel a summary with 1 failed scenario") +def step_failed_scenario_summary(context: Context) -> None: + context.bp_summary = _failed_scenario_summary() + + +@given("behave_parallel a summary with 1 errored scenario") +def step_errored_scenario_summary(context: Context) -> None: + context.bp_summary = _errored_scenario_summary() + + +@given("behave_parallel an empty summary") +def step_empty_summary(context: Context) -> None: + context.bp_summary = _empty_summary() + + +@given("behave_parallel a summary with all scenarios skipped") +def step_all_skipped_summary(context: Context) -> None: + context.bp_summary = _all_skipped_summary() + + +@when("behave_parallel I check chunk_has_failures") +def step_check_chunk_has_failures(context: Context) -> None: + context.bp_result = _chunk_has_failures(context.bp_summary) + + +@when("behave_parallel I check chunk_no_scenarios_ran") +def step_check_chunk_no_scenarios_ran(context: Context) -> None: + context.bp_result = _chunk_no_scenarios_ran(context.bp_summary) + + +@then("behave_parallel chunk_has_failures should return false") +def step_chunk_has_failures_false(context: Context) -> None: + assert context.bp_result is False, f"Expected False, got {context.bp_result}" + + +@then("behave_parallel chunk_has_failures should return true") +def step_chunk_has_failures_true(context: Context) -> None: + assert context.bp_result is True, f"Expected True, got {context.bp_result}" + + +@then("behave_parallel chunk_no_scenarios_ran should return true") +def step_chunk_no_scenarios_ran_true(context: Context) -> None: + assert context.bp_result is True, f"Expected True, got {context.bp_result}" + + +@then("behave_parallel chunk_no_scenarios_ran should return false") +def step_chunk_no_scenarios_ran_false(context: Context) -> None: + assert context.bp_result is False, f"Expected False, got {context.bp_result}" + + +# ---- Conditional log replay integration steps ---- + + +@given( + 'behave_parallel a worker result with passing summary and captured stdout "{text}"' +) +def step_passing_worker_result(context: Context, text: str) -> None: + context.bp_worker_results = [ + (False, text, "", _passing_summary()), + ] + + +@given( + 'behave_parallel a worker result with failed summary and captured stdout "{text}"' +) +def step_failed_worker_result(context: Context, text: str) -> None: + context.bp_worker_results = [ + (True, text, "", _failed_scenario_summary()), + ] + + +@given( + 'behave_parallel a worker result with crashed summary and captured stderr "{text}"' +) +def step_crashed_worker_result(context: Context, text: str) -> None: + context.bp_worker_results = [ + (True, "", text, _crash_summary()), + ] + + +@given( + 'behave_parallel a worker result with empty summary and captured stderr "{text}"' +) +def step_empty_summary_worker_result(context: Context, text: str) -> None: + # Pure all-zeros summary: _chunk_has_failures() returns False but + # _chunk_no_scenarios_ran() returns True, exercising the no-scenarios-ran + # path exclusively. + context.bp_worker_results = [ + (False, "", text, _empty_summary()), + ] + + +@given( + 'behave_parallel a worker result with failed summary and captured stderr "{text}"' +) +def step_failed_worker_result_stderr(context: Context, text: str) -> None: + context.bp_worker_results = [ + (True, "", text, _failed_scenario_summary()), + ] + + +@given("behave_parallel worker results with one passing and one failing chunk") +def step_mixed_worker_results(context: Context) -> None: + context.bp_worker_results = [ + (False, "PASSING CHUNK", "", _passing_summary()), + (True, "FAILING CHUNK", "", _failed_scenario_summary()), + ] + + +@given( + "behave_parallel a worker result where worker_failed is true but summary is passing" + ' with stdout "{text}"' +) +def step_tdd_inverted_worker_result(context: Context, text: str) -> None: + # TDD-inverted chunk: runner.run() returned True (raw worker_failed=True) + # but the after_scenario hook corrected the scenario status to passing. + # The summary therefore shows no failures. Output must be suppressed. + context.bp_worker_results = [ + (True, text, "", _passing_summary()), + ] + + +@given("behave_parallel worker results with one crashed chunk and one passing chunk") +def step_crash_and_pass_worker_results(context: Context) -> None: + # Partial crash: one worker sets features.errors=1, the other passes. + # The merged total must have _has_failures() == True. + context.bp_worker_results = [ + (True, "", "CRASH TRACEBACK", _crash_summary()), + (False, "PASS OUT", "", _passing_summary()), + ] + + +@when("behave_parallel I aggregate the worker results") +def step_aggregate_worker_results(context: Context) -> None: + """Invoke _aggregate_worker_results() and capture its stdout/stderr output.""" + results: list[tuple[bool, str, str, Summary]] = context.bp_worker_results + + stdout_capture = io.StringIO() + stderr_capture = io.StringIO() + + with redirect_stdout(stdout_capture), redirect_stderr(stderr_capture): + context.bp_aggregated_summary = _aggregate_worker_results(results) + + context.bp_aggregated_stdout = stdout_capture.getvalue() + context.bp_aggregated_stderr = stderr_capture.getvalue() + + +@then('behave_parallel the aggregated stdout should not contain "{text}"') +def step_aggregated_stdout_not_contains(context: Context, text: str) -> None: + assert text not in context.bp_aggregated_stdout, ( + f"Expected '{text}' NOT in stdout, but found it: " + f"{context.bp_aggregated_stdout!r}" + ) + + +@then('behave_parallel the aggregated stdout should contain "{text}"') +def step_aggregated_stdout_contains(context: Context, text: str) -> None: + assert text in context.bp_aggregated_stdout, ( + f"Expected '{text}' in stdout, but not found: {context.bp_aggregated_stdout!r}" + ) + + +@then('behave_parallel the aggregated stderr should contain "{text}"') +def step_aggregated_stderr_contains(context: Context, text: str) -> None: + assert text in context.bp_aggregated_stderr, ( + f"Expected '{text}' in stderr, but not found: {context.bp_aggregated_stderr!r}" + ) + + +@then("behave_parallel the aggregated summary should have failures") +def step_aggregated_summary_has_failures(context: Context) -> None: + summary = context.bp_aggregated_summary + assert _has_failures(summary), ( + f"Expected _has_failures() to return True for summary: {summary}" + ) + + +# ---- Worker exception handling steps ---- + + +@given("behave_parallel a worker payload that will raise an exception") +def step_worker_payload_raises(context: Context) -> None: + # This step relies on behave raising a real exception when it cannot + # locate the feature file — specifically, behave's feature loader raises + # an exception (e.g. ConfigError or IOError) for paths that do not exist + # on disk. This is verified, documented behave behavior: passing a + # non-existent path to behave.configuration.Configuration causes behave + # to fail during feature loading rather than silently returning zero + # scenarios. If this test starts failing, check whether behave changed + # its file-not-found handling (look at behave.runner.Runner.load_step_modules + # and behave.runner_util.collect_feature_locations). + context.bp_worker_payload = ( + ["__nonexistent_feature_path_that_will_crash__.feature"], + [], + ) + + +@when("behave_parallel I invoke the worker") +def step_invoke_worker(context: Context) -> None: + """Invoke _worker_run_features with a real crashing payload.""" + result = _worker_run_features(context.bp_worker_payload) + context.bp_worker_result = result + + +@then("behave_parallel the worker result should indicate failure") +def step_worker_result_failure(context: Context) -> None: + failed, _stdout, _stderr, _summary = context.bp_worker_result + assert failed is True, f"Expected failed=True, got {failed}" + + +@then('behave_parallel the worker stderr should contain "{text}"') +def step_worker_stderr_contains(context: Context, text: str) -> None: + _failed, _stdout, stderr, _summary = context.bp_worker_result + assert text in stderr, f"Expected '{text}' in stderr, got: {stderr!r}" + + +@then("behave_parallel the worker summary should have features errors equal to 1") +def step_worker_summary_features_errors_1(context: Context) -> None: + _failed, _stdout, _stderr, summary = context.bp_worker_result + errors = summary["features"]["errors"] + assert errors == 1, ( + f"Expected features.errors=1 so _has_failures() detects partial crash, " + f"got: {errors}" + ) diff --git a/robot/behave_parallel_log_filtering.robot b/robot/behave_parallel_log_filtering.robot new file mode 100644 index 000000000..3b3a0d9ed --- /dev/null +++ b/robot/behave_parallel_log_filtering.robot @@ -0,0 +1,77 @@ +*** Settings *** +Documentation Integration tests for conditional log replay in the parallel +... behave runner (``scripts/run_behave_parallel.py``). +... +... Verifies that passing chunks suppress output while failed, +... errored, or crashed chunks surface diagnostics. +Resource ${CURDIR}/common.resource +Suite Setup Setup Test Environment +Suite Teardown Cleanup Test Environment + +*** Variables *** +${HELPER} ${CURDIR}/helper_behave_parallel_log_filtering.py + +*** Test Cases *** +Passing Chunk Output Is Suppressed + [Documentation] A fully passing worker chunk must not trigger log replay. + [Tags] testing parallel log_filtering + ${result}= Run Process ${PYTHON} ${HELPER} passing-chunk-suppressed + ... cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} passing-chunk-suppressed-ok + +Failed Chunk Output Is Replayed + [Documentation] A worker chunk with scenario failures must trigger log replay. + [Tags] testing parallel log_filtering + ${result}= Run Process ${PYTHON} ${HELPER} failed-chunk-replayed + ... cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} failed-chunk-replayed-ok + +Crashed Chunk Is Detected + [Documentation] A worker chunk that crashes (empty summary) must be detected. + [Tags] testing parallel log_filtering + ${result}= Run Process ${PYTHON} ${HELPER} crashed-chunk-detected + ... cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} crashed-chunk-detected-ok + +Worker Crash Returns Traceback And Empty Summary + [Documentation] A worker that raises an unhandled exception must return a + ... traceback in stderr and an empty summary so the parent can + ... detect the crash. + [Tags] testing parallel log_filtering + ${result}= Run Process ${PYTHON} ${HELPER} worker-crash-handling + ... cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} worker-crash-handling-ok + +Mixed Results Only Replay Failed Chunks + [Documentation] When aggregating results from multiple workers, only + ... failed chunks should have their output replayed. + [Tags] testing parallel log_filtering + ${result}= Run Process ${PYTHON} ${HELPER} mixed-results-filtering + ... cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} mixed-results-filtering-ok + +Summary Merge Semantics Are Preserved + [Documentation] The summary merge logic must correctly combine counts + ... from multiple worker chunks regardless of filtering. + [Tags] testing parallel log_filtering + ${result}= Run Process ${PYTHON} ${HELPER} summary-merge-preserved + ... cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} summary-merge-preserved-ok diff --git a/robot/helper_behave_parallel_log_filtering.py b/robot/helper_behave_parallel_log_filtering.py new file mode 100644 index 000000000..6b657ae9a --- /dev/null +++ b/robot/helper_behave_parallel_log_filtering.py @@ -0,0 +1,180 @@ +"""Helper for behave_parallel_log_filtering.robot integration tests. + +Each function exercises a specific aspect of the parallel runner's +conditional log replay logic and prints a sentinel on success. +""" + +from __future__ import annotations + +import importlib.util +import io +import sys +from contextlib import redirect_stdout +from pathlib import Path +from types import ModuleType + + +def _load_runner_module() -> ModuleType: + """Load scripts/run_behave_parallel.py as a module via importlib. + + **CWD requirement**: This function resolves the script path relative to + the current working directory (``Path("scripts") / "run_behave_parallel.py"``). + It must be called from the repository root (i.e. the directory that + contains the ``scripts/`` folder). The Robot integration tests satisfy + this by passing ``cwd=${WORKSPACE}`` to ``Run Process``. If you call + this helper from a different directory, either ``cd`` to the repo root + first or modify the path to be anchored relative to ``__file__``. + """ + script_path = Path("scripts") / "run_behave_parallel.py" + spec = importlib.util.spec_from_file_location( + "run_behave_parallel", str(script_path) + ) + if spec is None or spec.loader is None: + msg = f"Cannot load {script_path}" + raise ImportError(msg) + mod = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = mod # Prevent re-execution on subsequent calls + spec.loader.exec_module(mod) + return mod + + +_mod = _load_runner_module() + +# Re-export Summary type for annotation use below. +Summary = _mod.Summary + + +def _passing_summary(features: int = 1, scenarios: int = 1) -> Summary: + return { + "features": { + "passed": features, + "failed": 0, + "errors": 0, + "skipped": 0, + }, + "scenarios": { + "passed": scenarios, + "failed": 0, + "errors": 0, + "skipped": 0, + }, + "steps": { + "passed": scenarios * 3, + "failed": 0, + "errors": 0, + "skipped": 0, + }, + "duration": 1.0, + } + + +def _failed_summary() -> Summary: + s = _mod._empty_summary() + s["scenarios"]["failed"] = 1 + return s + + +# ---- Test functions ---- + + +def test_passing_chunk_suppressed() -> None: + """Verify passing chunk output is not replayed.""" + summary = _passing_summary() + assert not _mod._chunk_has_failures(summary) + assert not _mod._chunk_no_scenarios_ran(summary) + print("passing-chunk-suppressed-ok") + + +def test_failed_chunk_replayed() -> None: + """Verify failed chunk triggers log replay.""" + summary = _failed_summary() + assert _mod._chunk_has_failures(summary) + print("failed-chunk-replayed-ok") + + +def test_crashed_chunk_detected() -> None: + """Verify crashed chunk (empty summary) is detected.""" + summary = _mod._empty_summary() + assert _mod._chunk_no_scenarios_ran(summary) + print("crashed-chunk-detected-ok") + + +def test_worker_crash_handling() -> None: + """Verify worker crash returns a traceback in stderr and a failed summary. + + Uses a real crash: passing a path that does not exist causes behave's + feature loader to raise a real exception inside _worker_run_features, + exercising the actual exception handler without any mocking. The crash + summary must have features.errors == 1 so _has_failures() propagates + the failure through the merged total even when other workers passed. + + This test relies on behave raising an exception for non-existent feature + file paths — specifically, behave's feature loader raises an exception + (e.g. ConfigError or IOError) rather than returning an empty run result. + This is verified, documented behave behavior. If this test starts + failing, check whether behave changed its file-not-found behavior + (look at behave.runner.Runner.load_step_modules and + behave.runner_util.collect_feature_locations). + """ + failed, _stdout, stderr, summary = _mod._worker_run_features( + (["__nonexistent_feature_path_that_will_crash__.feature"], []) + ) + + assert failed is True, f"Expected failed=True, got {failed}" + assert "WORKER CRASH:" in stderr, f"Expected 'WORKER CRASH:' in stderr: {stderr!r}" + # features.errors must be 1 so _has_failures() catches partial crashes. + assert summary["features"]["errors"] == 1, ( + f"Expected features.errors=1, got: {summary['features']['errors']}" + ) + print("worker-crash-handling-ok") + + +def test_mixed_results_filtering() -> None: + """Verify only failed chunks contribute output in aggregation. + + Calls _aggregate_worker_results() directly so the test exercises the + real production code rather than a reimplementation of the loop. + """ + results: list[tuple[bool, str, str, Summary]] = [ + (False, "PASS_OUT", "", _passing_summary()), + (True, "FAIL_OUT", "", _failed_summary()), + ] + + # Capture stdout to inspect what _aggregate_worker_results replays. + captured = io.StringIO() + with redirect_stdout(captured): + _mod._aggregate_worker_results(results) + + replayed = captured.getvalue() + assert "FAIL_OUT" in replayed, f"Expected FAIL_OUT in output: {replayed!r}" + assert "PASS_OUT" not in replayed, f"Expected PASS_OUT NOT in output: {replayed!r}" + print("mixed-results-filtering-ok") + + +def test_summary_merge_preserved() -> None: + """Verify summary merge semantics are unchanged.""" + s1 = _passing_summary(features=2, scenarios=4) + s2 = _failed_summary() + merged = _mod._merge_summaries([s1, s2]) + assert merged["features"]["passed"] == 2 + assert merged["scenarios"]["passed"] == 4 + assert merged["scenarios"]["failed"] == 1 + print("summary-merge-preserved-ok") + + +if __name__ == "__main__": + cmd = sys.argv[1] if len(sys.argv) > 1 else "" + dispatch = { + "passing-chunk-suppressed": test_passing_chunk_suppressed, + "failed-chunk-replayed": test_failed_chunk_replayed, + "crashed-chunk-detected": test_crashed_chunk_detected, + "worker-crash-handling": test_worker_crash_handling, + "mixed-results-filtering": test_mixed_results_filtering, + "summary-merge-preserved": test_summary_merge_preserved, + } + fn = dispatch.get(cmd) + if fn is None: + print(f"Unknown command: {cmd!r}", file=sys.stderr) + print(f"Available: {', '.join(sorted(dispatch))}", file=sys.stderr) + sys.exit(1) + fn() diff --git a/scripts/run_behave_parallel.py b/scripts/run_behave_parallel.py index 3a8bd5d94..14721d779 100644 --- a/scripts/run_behave_parallel.py +++ b/scripts/run_behave_parallel.py @@ -25,6 +25,7 @@ import multiprocessing import os import sys import time +import traceback from contextlib import redirect_stderr, redirect_stdout, suppress from pathlib import Path from typing import Any @@ -143,6 +144,12 @@ def _no_scenarios_ran(total: Summary) -> bool: return s["passed"] + s["failed"] + s["errors"] + s["skipped"] == 0 +# Aliases for chunk-level checks — identical semantics to the overall-summary +# helpers; exposed under distinct names for readability at the call site. +_chunk_has_failures = _has_failures +_chunk_no_scenarios_ran = _no_scenarios_ran + + # --------------------------------------------------------------------------- # Feature discovery # --------------------------------------------------------------------------- @@ -221,19 +228,80 @@ def _worker_run_features( ``load_step_definitions()`` and ``load_hooks()`` still execute inside each worker (they ``exec()`` the step .py files and environment.py), but every ``import`` they trigger is a cache hit. + + If the worker crashes with an unhandled exception, the traceback is + captured into stderr and a crash summary with ``features.errors = 1`` + is returned so that the parent can detect the crash via + ``_chunk_has_failures`` (and also ``_chunk_no_scenarios_ran``, since no + scenarios reached a terminal state) and replay the diagnostics. """ feature_paths, behave_args = payload stdout_buf = io.StringIO() stderr_buf = io.StringIO() - runner = _make_runner(feature_paths, behave_args) + try: + runner = _make_runner(feature_paths, behave_args) - with redirect_stdout(stdout_buf), redirect_stderr(stderr_buf): - failed = runner.run() + with redirect_stdout(stdout_buf), redirect_stderr(stderr_buf): + failed = runner.run() - summary = _extract_summary(runner) - return failed, stdout_buf.getvalue(), stderr_buf.getvalue(), summary + summary = _extract_summary(runner) + return failed, stdout_buf.getvalue(), stderr_buf.getvalue(), summary + except Exception: + # Surface the full traceback so the parent can replay it for + # diagnostics. Set features.errors = 1 so that _has_failures() + # detects the crash in the merged total even when other workers + # passed (preventing a silent CI green when a partial worker crash + # occurs alongside passing chunks). + tb_text = traceback.format_exc() + stderr_buf.write(f"\nWORKER CRASH:\n{tb_text}") + crash_summary = _empty_summary() + crash_summary["features"]["errors"] = 1 + return True, stdout_buf.getvalue(), stderr_buf.getvalue(), crash_summary + + +# --------------------------------------------------------------------------- +# Parallel result aggregation +# --------------------------------------------------------------------------- + + +def _aggregate_worker_results( + results: list[tuple[bool, str, str, Summary]], +) -> Summary: + """Aggregate worker results, replaying logs only for failed chunks. + + Iterates over the list of ``(worker_failed, stdout, stderr, summary)`` + tuples returned by the multiprocessing pool. For each chunk whose + summary-based check indicates failure (failed/errored scenarios or + features, or a crash that produced no terminal scenario results), the + captured stdout and stderr are replayed to the parent's streams. + Passing chunks have their output suppressed to reduce CI noise. + + Uses summary-based checks rather than the raw ``worker_failed`` boolean + so that TDD-inverted chunks (``@tdd_expected_fail`` scenarios whose + status has been corrected to "passed" by the after_scenario hook) do + not trigger spurious log replay. + """ + summaries: list[Summary] = [] + for idx, (_, stdout, stderr, summary) in enumerate(results): + # Rely on summary-based checks: the @tdd_expected_fail handler in + # environment.py inverts scenario statuses but cannot update the + # runner.run() local variable, so the raw worker_failed flag is + # not trustworthy for TDD-inverted chunks. + chunk_failed = _chunk_has_failures(summary) or _chunk_no_scenarios_ran(summary) + if chunk_failed: + if stdout: + print(f"\n--- Worker {idx} stdout (chunk failed) ---") + print(stdout, end="") + if stderr: + print( + f"\n--- Worker {idx} stderr (chunk failed) ---", + file=sys.stderr, + ) + print(stderr, end="", file=sys.stderr) + summaries.append(summary) + return _merge_summaries(summaries) # --------------------------------------------------------------------------- @@ -294,14 +362,7 @@ def main(argv: list[str] | None = None) -> None: [(chunk, other_args) for chunk in chunks], ) - summaries = [] - for _worker_failed, stdout, stderr, summary in results: - if stdout: - print(stdout, end="") - if stderr: - print(stderr, end="", file=sys.stderr) - summaries.append(summary) - total = _merge_summaries(summaries) + total = _aggregate_worker_results(results) wall = time.monotonic() - start _print_overall_summary(total, wall_seconds=wall)