ariadne/tests/test_hermes_console_evidence.py

from __future__ import annotations

import pytest

from ariadne.services import hermes_console_evidence as module


def _console(lines) -> str:  # type: ignore[no-untyped-def]
    return "\n".join(lines)


def _noise(count: int, prefix: str = "step") -> list[str]:
    return [f"[Pipeline] {prefix} {index} completed ok" for index in range(count)]


@pytest.mark.parametrize(
    ("line", "marker"),
    [
        ("=== FAILURES ===", "=== FAILURES ==="),
        ("=========== short test summary info ==========", "short test summary"),
        ("ERROR at setup of test_thing", "ERROR at setup"),
        ("Traceback (most recent call last):", "Traceback (most recent call last)"),
        ("    raise AssertionError(x)", "AssertionError"),
        ("FAILED tests/test_a.py::test_b - boom", "FAILED "),
        ("E       assert 1 == 2", "E   "),
        ("BUILD FAILED in 3s", "BUILD FAILED"),
        ("FAILURE: gradle task :check reported problems", "FAILURE:"),
        ("process exited with non-zero exit status", "non-zero exit"),
        ("script returned exit code 1", "exit code"),
        ("bash: pytest: command not found", "command not found"),
        ("cp: No such file or directory", "No such file"),
        ("java.io.IOException: Exception while archiving", "Exception"),
        ("ERROR: something broke", "ERROR:"),
        ("error: lowercase also matches", "ERROR:"),
        ("Quality gate failed for project", "Quality gate"),
        ("Required test coverage of 80% not reached", "coverage"),
        ("FAIL Required test --fail-under=80", "fail-under"),
        ("missing docstring on public function", "docstring"),
        ("Semgrep found 3 findings", "Semgrep"),
        ("SonarQube analysis failed", "SonarQube"),
        ("Trivy detected HIGH severity", "Trivy"),
        ("ruff check found 2 problems", "ruff"),
        ("[loc] file exceeds 500 lines", "[loc]"),
        ("pod status ImagePullBackOff", "ImagePullBackOff"),
        ("container was OOMKilled", "OOMKilled"),
        (
            "Failed to establish a new connection: [Errno 111]",
            "Failed to establish a new connection",
        ),
        (
            "Temporary failure in name resolution for jenkins",
            "Temporary failure in name resolution",
        ),
    ],
)
def test_marker_families_are_detected(line: str, marker: str) -> None:
    assert module.marker_for_line(line) == marker
    result = module.extract_console_evidence(line)
    assert result["regions"][0]["marker"] == marker
    assert result["regions"][0]["line_number"] == 1


def test_clean_lines_produce_no_regions() -> None:
    result = module.extract_console_evidence(_console(_noise(20)))
    assert result["regions"] == []
    assert result["truncated"] is False
    assert result["total_lines"] == 20


def test_prefix_only_marker_does_not_match_mid_line() -> None:
    assert module.marker_for_line("phase   done   nothing wrong here") is None
    assert module.marker_for_line("   E   assert False") == "E   "


def test_prefix_only_failed_marker_ignores_prose() -> None:
    assert module.marker_for_line("Quality gate FAILED for project") == "Quality gate"
    assert module.marker_for_line("  FAILED tests/test_a.py::test_b") == "FAILED "


def test_priority_order_prefers_pytest_signal() -> None:
    assert module.marker_for_line("ERROR: raised AssertionError while loading") == "AssertionError"


def test_context_window_before_and_after() -> None:
    lines = [f"line-{index}" for index in range(60)]
    lines[30] = "ERROR: boom"
    result = module.extract_console_evidence(_console(lines))
    region = result["regions"][0]
    assert region["line_number"] == 31
    text = region["text"].splitlines()
    assert text[0] == "line-24"
    assert text[-1] == "line-42"
    assert len(text) == 19


def test_context_window_is_configurable_and_clamped_at_edges() -> None:
    lines = ["ERROR: boom"] + [f"line-{index}" for index in range(3)]
    result = module.extract_console_evidence(
        _console(lines), {"context_before": 2, "context_after": 1}
    )
    assert result["regions"][0]["text"] == "ERROR: boom\nline-0"


def test_overlapping_regions_are_merged() -> None:
    lines = [f"line-{index}" for index in range(40)]
    lines[10] = "ERROR: first"
    lines[15] = "ERROR: second"
    result = module.extract_console_evidence(_console(lines))
    assert len(result["regions"]) == 1
    region = result["regions"][0]
    assert region["line_number"] == 11
    assert "ERROR: first" in region["text"]
    assert "ERROR: second" in region["text"]
    assert region["text"].splitlines()[-1] == "line-27"


def test_distant_regions_are_kept_separate() -> None:
    lines = [f"line-{index}" for index in range(200)]
    lines[10] = "ERROR: first"
    lines[150] = "ERROR: second"
    result = module.extract_console_evidence(_console(lines))
    assert [region["line_number"] for region in result["regions"]] == [11, 151]


def test_repeated_regions_are_deduplicated_with_a_note() -> None:
    block = ["connecting to registry", "ERROR: connection reset by peer (attempt 1)", "retrying"]
    lines: list[str] = []
    for attempt in range(5):
        lines.extend(_noise(30, "pad"))
        lines.extend([block[0], block[1].replace("1", str(attempt + 1)), block[2]])
    lines.extend(_noise(30, "pad"))
    result = module.extract_console_evidence(_console(lines))
    assert len(result["regions"]) == 1
    assert result["regions"][0]["text"].endswith("(repeated 5 times)")
    assert result["regions"][0]["line_number"] == 32


def test_distinct_regions_are_not_deduplicated() -> None:
    lines = _noise(30) + ["ERROR: alpha broke"] + _noise(30, "mid") + ["ERROR: beta broke"]
    result = module.extract_console_evidence(_console(lines))
    assert len(result["regions"]) == 2
    assert "repeated" not in result["regions"][1]["text"]


def test_byte_budget_prefers_earliest_regions() -> None:
    lines: list[str] = []
    for index in range(6):
        lines.extend(_noise(40, "pad"))
        lines.append(f"ERROR: distinct failure {chr(ord('a') + index)} " + "x" * 400)
    result = module.extract_console_evidence(_console(lines), {"max_total_bytes": 2000})
    assert result["truncated"] is True
    markers = [region["line_number"] for region in result["regions"]]
    assert markers == sorted(markers)
    assert result["regions"][0]["line_number"] == 41
    assert "distinct failure a" in result["regions"][0]["text"]
    assert len(result["regions"]) < 6
    assert sum(len(region["text"]) for region in result["regions"]) <= 2000


def test_first_region_is_kept_and_clipped_when_it_alone_exceeds_budget() -> None:
    lines = ["ERROR: huge " + "y" * 5000, "ERROR: later " + "z" * 5000]
    result = module.extract_console_evidence(_console(lines), {"max_total_bytes": 100})
    assert len(result["regions"]) == 1
    assert len(result["regions"][0]["text"]) == 100
    assert result["truncated"] is True


def test_zero_byte_budget_still_reports_one_empty_region() -> None:
    result = module.extract_console_evidence("ERROR: boom", {"max_total_bytes": 0})
    assert result["regions"] == [{"marker": "ERROR:", "line_number": 1, "text": ""}]
    assert result["truncated"] is True


def test_max_regions_caps_the_region_count() -> None:
    lines: list[str] = []
    for index in range(8):
        lines.extend(_noise(40, "pad"))
        lines.append(f"ERROR: unique failure {chr(ord('a') + index)}")
    result = module.extract_console_evidence(_console(lines), {"max_regions": 3})
    assert len(result["regions"]) == 3
    assert result["truncated"] is True
    assert "ERROR: unique failure a" in result["regions"][0]["text"]


def test_all_regions_within_budget_are_not_truncated() -> None:
    lines = _noise(30) + ["ERROR: only one"] + _noise(30, "post")
    result = module.extract_console_evidence(_console(lines))
    assert len(result["regions"]) == 1
    assert result["truncated"] is False


def test_tail_is_capped_by_lines_and_bytes() -> None:
    lines = [f"line-{index}" for index in range(200)]
    result = module.extract_console_evidence(_console(lines))
    tail = result["tail"].splitlines()
    assert len(tail) == 40
    assert tail[0] == "line-160"
    assert tail[-1] == "line-199"

    wide = module.extract_console_evidence(_console(["q" * 500] * 50), {"tail_lines": 20})
    assert len(wide["tail"]) == 4000
    assert wide["tail"].endswith("q")

    assert module.extract_console_evidence("a\nb\nc", {"tail_lines": 0})["tail"] == ""
    assert module.extract_console_evidence("a\nb\nc", {"max_tail_bytes": 0})["tail"] == ""


def test_tail_shorter_than_the_window_is_returned_whole() -> None:
    result = module.extract_console_evidence("a\nb\nc")
    assert result["tail"] == "a\nb\nc"
    assert result["total_lines"] == 3


def test_empty_and_none_input() -> None:
    for value in ("", None, 12345, b"bytes"):
        result = module.extract_console_evidence(value)  # type: ignore[arg-type]
        assert result == {"regions": [], "tail": "", "total_lines": 0, "truncated": False}


def test_invalid_cfg_values_fall_back_to_defaults() -> None:
    lines = [f"line-{index}" for index in range(200)]
    cfg = {"tail_lines": "not-an-int", "context_before": -5, "max_regions": None}
    result = module.extract_console_evidence(_console(lines), cfg)
    assert len(result["tail"].splitlines()) == 40

    numeric = module.extract_console_evidence(_console(lines), {"tail_lines": "7"})
    assert len(numeric["tail"].splitlines()) == 7

    assert module.extract_console_evidence(_console(lines), "not-a-dict")["total_lines"] == 200  # type: ignore[arg-type]


def test_extractor_never_raises(monkeypatch) -> None:  # type: ignore[no-untyped-def]
    def boom(lines, options):  # type: ignore[no-untyped-def]
        raise RuntimeError("kaboom")

    monkeypatch.setattr(module, "_regions", boom)
    assert module.extract_console_evidence("ERROR: boom") == {
        "regions": [],
        "tail": "",
        "total_lines": 0,
        "truncated": False,
    }


def _long_pipeline_console() -> str:
    lines: list[str] = ["Started by user jenkins", "[Pipeline] node", "Running on agent-1"]
    lines.extend(_noise(120, "checkout"))
    lines.extend(
        [
            "[Pipeline] stage: Unit Tests",
            "+ python -m pytest tests -q",
            "tests/test_wallet.py .........",
            "tests/test_ledger.py ..F",
            "=================================== FAILURES ===================================",
            "____________________ test_ledger_balance_after_migration _______________________",
            "    def test_ledger_balance_after_migration():",
            "        balance = ledger.balance('acct-7')",
            ">       assert balance == 100",
            "E       assert 0 == 100",
            "tests/test_ledger.py:42: AssertionError",
            "FAILED tests/test_ledger.py::test_ledger_balance_after_migration",
        ]
    )
    lines.extend(_noise(3000, "teardown"))
    lines.extend(
        [
            "[Pipeline] stage: Archive",
            "java.io.IOException: Failed to archive artifacts",
            "    at hudson.FilePath.copyRecursiveTo(FilePath.java:2810)",
            "[Pipeline] End of Pipeline",
            "ERROR: script returned exit code 1",
            "Finished: FAILURE",
        ]
    )
    return _console(lines)


def test_long_pipeline_keeps_the_early_failure_that_the_tail_loses() -> None:
    console = _long_pipeline_console()
    result = module.extract_console_evidence(console)

    assert result["total_lines"] > 3000
    assert "FAILED tests/test_ledger.py::test_ledger_balance_after_migration" not in result["tail"]
    assert "script returned exit code 1" in result["tail"]

    first = result["regions"][0]
    assert first["marker"] == "=== FAILURES ==="
    assert "test_ledger_balance_after_migration" in first["text"]
    assert "assert 0 == 100" in first["text"]
    assert "FAILED tests/test_ledger.py::test_ledger_balance_after_migration" in first["text"]
    assert first["line_number"] < 200
    assert sum(len(region["text"]) for region in result["regions"]) <= 12000


def test_long_pipeline_regions_stay_chronological_and_reach_the_end() -> None:
    result = module.extract_console_evidence(_long_pipeline_console())
    numbers = [region["line_number"] for region in result["regions"]]
    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"])