diff --git a/ariadne/services/hermes_console_evidence.py b/ariadne/services/hermes_console_evidence.py index 4087826..e8eca44 100644 --- a/ariadne/services/hermes_console_evidence.py +++ b/ariadne/services/hermes_console_evidence.py @@ -4,7 +4,9 @@ import re from typing import Any -FAILURE_MARKERS: tuple[str, ...] = ( +# Markers that assert a failure happened. A line carrying one of these is +# evidence on its own. +STRONG_MARKERS: tuple[str, ...] = ( # kubernetes / build-agent signals: long and unambiguous, so they are # matched first and never mislabelled as a generic test failure "ImagePullBackOff", @@ -23,12 +25,19 @@ FAILURE_MARKERS: tuple[str, ...] = ( "BUILD FAILED", "FAILURE:", "non-zero exit", - "exit code", "command not found", "No such file", "Exception", "ERROR:", - # tool gates +) + +# Markers that only say a gate or tool ran. Every green build prints +# "coverage", "SonarQube", "Trivy" and "ruff" dozens of times before any real +# failure appears, so treating these as evidence let successful setup output +# crowd the actual failure out of the byte budget on long pipelines. They are +# retained only after every strong region has had its chance. +WEAK_MARKERS: tuple[str, ...] = ( + "exit code", "Quality gate", "fail-under", "coverage", @@ -40,6 +49,17 @@ FAILURE_MARKERS: tuple[str, ...] = ( "[loc]", ) +FAILURE_MARKERS: tuple[str, ...] = STRONG_MARKERS + WEAK_MARKERS + +_STRONG_TIER = 0 +_WEAK_TIER = 1 + +# A pytest progress line ending in a passing verdict is not a failure even when +# its parametrized id quotes one. Ariadne's own suite parametrizes marker +# detection with strings like "FAILED tests/test_a.py::test_b", and those +# passing lines were being collected as failure evidence. +_PASSING_VERDICT = re.compile(r"\b(PASSED|SKIPPED|XFAIL|XPASS)\b\s*(\[\s*\d+%\])?\s*$") + # Markers that are only meaningful at the start of a (stripped) line; matching # them anywhere would swallow every line that merely says "... failed ...". # All matching is case-insensitive, so these are stored lowercased. @@ -91,7 +111,10 @@ def marker_for_line(line: str) -> str | None: `FAILURE_MARKERS`, or None when the line carries no failure signal. """ - lowered = str(line).lower() + raw = str(line) + if _PASSING_VERDICT.search(raw.rstrip()): + return None + lowered = raw.lower() stripped = lowered.lstrip() for marker in FAILURE_MARKERS: needle = marker.lower() @@ -103,6 +126,17 @@ def marker_for_line(line: str) -> str | None: return None +def marker_tier(marker: str | None) -> int: + """Return the evidence tier of a marker: 0 asserts failure, 1 only hints. + + Inputs: a marker string as returned by `marker_for_line`, or None. + Outputs: `_STRONG_TIER` for markers that assert a failure occurred and + `_WEAK_TIER` for markers that merely name a tool or gate that ran. + """ + + return _STRONG_TIER if marker in STRONG_MARKERS else _WEAK_TIER + + def _lines(console_text: Any) -> list[str]: """Split console text into lines, tolerating None and non-strings.""" @@ -136,14 +170,22 @@ def _regions(lines: list[str], options: dict[str, int]) -> list[dict[str, Any]]: continue start = max(0, index - options["context_before"]) end = min(len(lines), index + options["context_after"] + 1) + tier = marker_tier(marker) if spans and start <= spans[-1]["end"]: spans[-1]["end"] = max(spans[-1]["end"], end) + if tier < spans[-1]["tier"]: + # A merged span inherits the strongest signal inside it, so a + # real failure a few lines below a tool banner is not demoted. + spans[-1].update({"tier": tier, "marker": marker, "line_number": index + 1}) continue - spans.append({"marker": marker, "line_number": index + 1, "start": start, "end": end}) + spans.append( + {"marker": marker, "line_number": index + 1, "start": start, "end": end, "tier": tier} + ) return [ { "marker": span["marker"], "line_number": span["line_number"], + "tier": span["tier"], "text": "\n".join(lines[span["start"] : span["end"]]), } for span in spans @@ -184,25 +226,41 @@ def _with_repeat_note(region: dict[str, Any], count: int) -> dict[str, Any]: def _budgeted( regions: list[dict[str, Any]], options: dict[str, int] ) -> tuple[list[dict[str, Any]], bool]: - """Keep the earliest regions that fit the byte and count budgets.""" + """Keep the strongest, then earliest, regions that fit the budgets. + + Regions are offered strongest-tier-first and earliest-first within a tier, + so on a long pipeline the enforced failure is retained even when hundreds + of successful tool banners precede it. The kept regions are returned in + chronological order, which is the order the reader needs them in. + """ limit = options["max_total_bytes"] kept: list[dict[str, Any]] = [] used = 0 truncated = False - for region in regions: + ordered = sorted(regions, key=lambda region: (region.get("tier", _WEAK_TIER), region["line_number"])) + for region in ordered: if len(kept) >= options["max_regions"]: - return kept, True + truncated = True + break size = _byte_len(region["text"]) if used + size > limit: if kept: - return kept, True + truncated = True + break region = {**region, "text": _clip_head(region["text"], limit)} size = _byte_len(region["text"]) truncated = True kept.append(region) used += size - return kept, truncated + chronological = sorted(kept, key=lambda region: region["line_number"]) + return [_without_tier(region) for region in chronological], truncated + + +def _without_tier(region: dict[str, Any]) -> dict[str, Any]: + """Drop the internal tier key so the emitted region shape stays frozen.""" + + return {key: value for key, value in region.items() if key != "tier"} def _tail(lines: list[str], options: dict[str, int]) -> str: diff --git a/tests/test_hermes_console_evidence.py b/tests/test_hermes_console_evidence.py index aa48d29..8fd8a26 100644 --- a/tests/test_hermes_console_evidence.py +++ b/tests/test_hermes_console_evidence.py @@ -298,3 +298,60 @@ def test_long_pipeline_regions_stay_chronological_and_reach_the_end() -> None: assert numbers == sorted(numbers) assert len(numbers) >= 2 assert "script returned exit code 1" in result["regions"][-1]["text"] + + +def test_strong_marker_survives_a_flood_of_weak_tool_banners() -> None: + """The enforced failure must outrank successful gate output.""" + + lines: list[str] = [] + for index in range(8): + lines.append(f"INFO SonarQube analysing file {index}, coverage cached, ruff clean") + lines.extend(_noise(30, f"gap{index}")) + lines.append("FAILED tests/test_utils.py::test_thing - AssertionError: assert 1 == 2") + lines.extend(_noise(5, "post")) + + result = module.extract_console_evidence(_console(lines), {"max_regions": 2}) + text = " ".join(region["text"] for region in result["regions"]) + + # Eight separate weak banners precede the failure. Earliest-first selection + # kept only those and dropped the failure entirely; tiering must not. + assert "tests/test_utils.py::test_thing" in text + markers = [region["marker"] for region in result["regions"]] + assert any(module.marker_tier(marker) == module._STRONG_TIER for marker in markers) + + +def test_passing_pytest_progress_lines_are_not_failures() -> None: + """A parametrized id may quote a failure without being one.""" + + assert module.marker_for_line( + "tests/test_x.py::test_markers[FAILED tests/a.py::b - boom] PASSED [ 32%]" + ) is None + assert module.marker_for_line("tests/test_x.py::test_y SKIPPED [ 10%]") is None + real = module.marker_for_line("FAILED tests/test_x.py::test_y - AssertionError") + assert module.marker_tier(real) == module._STRONG_TIER + + +def test_marker_tiers_split_assertions_from_tool_names() -> None: + """Tool banners are hints; failure text is evidence.""" + + assert module.marker_tier("AssertionError") == module._STRONG_TIER + assert module.marker_tier("short test summary") == module._STRONG_TIER + assert module.marker_tier("coverage") == module._WEAK_TIER + assert module.marker_tier("SonarQube") == module._WEAK_TIER + assert module.marker_tier(None) == module._WEAK_TIER + + +def test_kept_regions_stay_in_chronological_order() -> None: + """Selection reorders by strength; output must read top to bottom.""" + + lines = ["INFO Trivy scanning base image"] + lines.extend(_noise(30)) + lines.append("AssertionError: boom") + lines.extend(_noise(30, "later")) + lines.append("short test summary info") + + result = module.extract_console_evidence(_console(lines)) + numbers = [region["line_number"] for region in result["regions"]] + + assert numbers == sorted(numbers) + assert all("tier" not in region for region in result["regions"])