cleveragents-core/scripts/behave_pass_suppress_formatter.py

"""PassSuppressFormatter — a custom Behave formatter that suppresses output
for passing and skipped scenarios, reducing CI log noise from ~100,000 lines
to ≤ 10 lines for an all-passing suite.

This module is embedded in the same directory as ``run_behave_parallel.py``
so that ``_install_behave_parallel()`` in ``noxfile.py`` can copy both files
into the temporary ``behave_parallel`` package.
"""

from __future__ import annotations

import io
from typing import Any

from behave.formatter.base import (
    Formatter as _BehaveFormatter,  # type: ignore[import-untyped]
)

# Sentinel set of statuses whose output should be suppressed.
# Every status not in this set (e.g. "failed", "undefined") causes the
# buffered scenario output to be flushed to the real output stream.
_SUPPRESS_STATUSES: frozenset[str] = frozenset({"passed", "skipped"})


class PassSuppressFormatter(_BehaveFormatter):
    """Behave formatter that suppresses output for passing scenarios.

    All per-scenario output is buffered in a :class:`io.StringIO`.  When a
    scenario ends (signalled by the next :meth:`scenario` call or by
    :meth:`eof`):

    * **Failed / errored scenario** — the buffer is flushed to the real
      output stream so developers and CI tools see the full step-level
      details.
    * **Passed / skipped scenario** — the buffer is silently discarded.

    The result for an all-passing suite is zero formatter output, leaving
    only the ``_print_overall_summary`` block emitted by the runner as
    visible output (~5-10 lines regardless of suite size).

    Coverage mode (``BEHAVE_PARALLEL_COVERAGE=1``) bypasses this formatter
    entirely via ``_make_runner()``, which falls back to behave's default
    format so that slipcover can instrument a single sequential process
    without output interference.
    """

    name: str = "pass_suppress"
    description: str = "Suppress passing scenario output; show failures fully"

    def __init__(self, stream_opener: Any, config: Any) -> None:
        super().__init__(stream_opener, config)
        # Ensure the real output stream is open (opens it if stream_opener
        # was constructed with only a filename, not a pre-opened stream).
        self.stream = self.open()

        # Per-scenario buffering state.
        self._current_scenario: Any = None
        self._scenario_buf: io.StringIO = io.StringIO()

    # ------------------------------------------------------------------
    # Internal helpers
    # ------------------------------------------------------------------

    def _finalize_previous_scenario(self) -> None:
        """Flush the scenario buffer to the real stream if scenario failed.

        Called at the start of each new scenario and at end-of-feature so
        the previous scenario's buffered output is either committed or
        discarded based on the scenario's final status.
        """
        if self._current_scenario is None:
            return
        status: Any = getattr(self._current_scenario, "status", None)
        status_name: str = (
            getattr(status, "name", str(status)) if status is not None else ""
        )
        if status_name not in _SUPPRESS_STATUSES:
            output = self._scenario_buf.getvalue()
            if output:
                self.stream.write(output)
                self.stream.flush()
        # Reset the buffer regardless — the next scenario starts fresh.
        self._scenario_buf = io.StringIO()

    # ------------------------------------------------------------------
    # Formatter interface (behave lifecycle hooks)
    # ------------------------------------------------------------------

    def uri(self, uri: str) -> None:
        """Called before each feature file; no output needed."""

    def feature(self, feature: Any) -> None:
        """Called at feature start; no output needed."""

    def background(self, background: Any) -> None:
        """Called when a feature background is declared; no output needed."""

    def scenario(self, scenario: Any) -> None:
        """Start buffering output for *scenario*, finalising the previous one."""
        self._finalize_previous_scenario()
        self._current_scenario = scenario
        # Write the scenario header into the new buffer.
        self._scenario_buf.write(f"\n  {scenario.keyword}: {scenario.name}\n")

    def step(self, step: Any) -> None:
        """Step announced before execution; no output needed at this point."""

    def match(self, match: Any) -> None:
        """Step matched; no output needed."""

    def result(self, step: Any) -> None:
        """Record a step result in the per-scenario buffer."""
        status: Any = getattr(step, "status", None)
        status_name: str = (
            getattr(status, "name", "unknown") if status is not None else "unknown"
        )
        self._scenario_buf.write(f"    {step.keyword} {step.name} ... {status_name}\n")
        error_msg: str | None = getattr(step, "error_message", None)
        if error_msg:
            self._scenario_buf.write(f"{error_msg}\n")

    def eof(self) -> None:
        """End of feature file: finalise the current (last) scenario."""
        self._finalize_previous_scenario()
        self._current_scenario = None