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"])