ariadne/tests/test_hermes_autotriage_evidence.py

from __future__ import annotations

from types import SimpleNamespace

from ariadne.services import hermes_autotriage_evidence as module


JOB = "hermes-triage-demo"
INCIDENT_ID = f"{JOB}/12"


def _settings() -> SimpleNamespace:
    return SimpleNamespace(
        jenkins_base_url="https://ci.example",
        jenkins_api_user="user",
        jenkins_api_token="token",
        jenkins_api_timeout_sec=5.0,
        opensearch_url="http://opensearch:9200",
        hermes_demo_namespace="hermes-triage-demo",
    )


class FakeResponse:
    def __init__(self, payload=None, text="") -> None:  # type: ignore[no-untyped-def]
        self.payload = payload
        self.text = text

    def raise_for_status(self) -> None:
        return None

    def json(self):  # type: ignore[no-untyped-def]
        return self.payload


def _install(monkeypatch, routes, log_result=None) -> dict:  # type: ignore[no-untyped-def]
    calls: dict = {"gets": [], "log": [], "client_kwargs": None}

    class FakeClient:
        def __init__(self, **kwargs) -> None:  # type: ignore[no-untyped-def]
            calls["client_kwargs"] = kwargs

        def __enter__(self):  # type: ignore[no-untyped-def]
            return self

        def __exit__(self, *args) -> None:  # type: ignore[no-untyped-def]
            return None

        def get(self, url, params=None):  # type: ignore[no-untyped-def]
            calls["gets"].append((url, params))
            for suffix, item in routes.items():
                if suffix in url:
                    if isinstance(item, Exception):
                        raise item
                    return item
            raise AssertionError(f"unexpected url {url}")

    def fake_logs(config, incident_id, window_start, window_end):  # type: ignore[no-untyped-def]
        calls["log"].append((config, incident_id, window_start, window_end))
        if log_result is not None:
            return log_result
        return {"query_window": {}, "records": [], "truncated": False, "error": None}

    monkeypatch.setattr(module, "settings", _settings())
    monkeypatch.setattr(module.httpx, "Client", FakeClient)
    monkeypatch.setattr(module, "collect_log_evidence", fake_logs)
    return calls


def _last_build(**overrides):  # type: ignore[no-untyped-def]
    payload = {
        "number": 12,
        "result": "FAILURE",
        "building": False,
        "timestamp": 1720000000000,
        "duration": 60000,
        "url": f"https://ci.example/job/{JOB}/12/",
    }
    payload.update(overrides)
    return payload


def _all_failing_routes() -> dict:
    return {
        "/wfapi/describe": RuntimeError("stage fetch failed"),
        "/testReport/api/json": RuntimeError("no test report"),
        "/consoleText": RuntimeError("no console"),
    }


def test_collect_evidence_full_bundle(monkeypatch) -> None:
    routes = {
        "/wfapi/describe": FakeResponse(
            {"stages": [{"name": "Build", "status": "SUCCESS"}, {"name": "Test", "status": "FAILED"}]}
        ),
        "/testReport/api/json": FakeResponse(
            {
                "suites": [
                    {
                        "cases": [
                            {
                                "name": "fixture-state-check",
                                "className": "demo.Fixture",
                                "status": "FAILED",
                                "errorDetails": "boom",
                            },
                            {"name": "ok-test", "className": "demo.Ok", "status": "PASSED"},
                            {
                                "name": "flaky",
                                "className": "demo.Flaky",
                                "status": "REGRESSION",
                                "errorDetails": 5,
                            },
                        ]
                    }
                ]
            }
        ),
        "/consoleText": FakeResponse(text="line1\nline2\nhermes_demo_test_failure seen"),
    }
    calls = _install(monkeypatch, routes)
    bundle = module.collect_evidence(INCIDENT_ID, JOB, _last_build())

    assert bundle["incident_id"] == INCIDENT_ID
    assert bundle["generated_at"]
    jenkins = bundle["jenkins"]
    assert jenkins["job"] == JOB
    assert jenkins["build_number"] == 12
    assert jenkins["result"] == "FAILURE"
    assert jenkins["url"] == f"https://ci.example/job/{JOB}/12/"
    assert jenkins["duration_seconds"] == 60.0
    assert jenkins["timestamps"] == {
        "start": "2024-07-03T09:46:40+00:00",
        "end": "2024-07-03T09:47:40+00:00",
    }
    assert jenkins["first_failed_stage"] == "Test"
    assert jenkins["failed_tests"] == [
        {
            "name": "fixture-state-check",
            "className": "demo.Fixture",
            "errorDetails": "boom",
            "errorStackTrace": None,
        },
        {"name": "flaky", "className": "demo.Flaky", "errorDetails": None, "errorStackTrace": None},
    ]
    assert jenkins["console_tail"] == "line1\nline2\nhermes_demo_test_failure seen"
    assert jenkins["console_failures"] == []
    assert jenkins["console_truncated"] is False
    assert bundle["log_evidence"]["records"] == []
    config, incident_id, window_start, window_end = calls["log"][0]
    assert incident_id == INCIDENT_ID
    assert (window_start, window_end) == (jenkins["timestamps"]["start"], jenkins["timestamps"]["end"])
    assert config == {
        "opensearch_url": "http://opensearch:9200",
        "namespace": "hermes-triage-demo",
        "extra_namespaces": ["jenkins"],
    }
    assert calls["client_kwargs"]["auth"] == ("user", "token")


def test_collect_evidence_tolerates_all_fetch_failures(monkeypatch) -> None:
    _install(monkeypatch, _all_failing_routes())
    bundle = module.collect_evidence(INCIDENT_ID, JOB, _last_build())
    jenkins = bundle["jenkins"]
    assert jenkins["first_failed_stage"] is None
    assert jenkins["failed_tests"] == []
    assert jenkins["console_tail"] is None
    assert bundle["log_evidence"]["error"] is None


def test_collect_evidence_tolerates_client_failure(monkeypatch) -> None:
    _install(monkeypatch, {})

    def boom(**kwargs):  # type: ignore[no-untyped-def]
        raise RuntimeError("no client")

    monkeypatch.setattr(module.httpx, "Client", boom)
    bundle = module.collect_evidence(INCIDENT_ID, JOB, _last_build())
    assert bundle["jenkins"]["failed_tests"] == []
    assert bundle["jenkins"]["console_tail"] is None
    assert "log_evidence" in bundle


def test_console_tail_caps_lines_and_bytes(monkeypatch) -> None:
    routes = _all_failing_routes()
    routes["/consoleText"] = FakeResponse(text="\n".join(f"line-{i}" for i in range(300)))
    _install(monkeypatch, routes)
    tail = module.collect_evidence(INCIDENT_ID, JOB, _last_build())["jenkins"]["console_tail"]
    lines = tail.splitlines()
    assert len(lines) == 40
    assert lines[0] == "line-260"
    assert lines[-1] == "line-299"

    routes["/consoleText"] = FakeResponse(text="x" * 20000)
    _install(monkeypatch, routes)
    tail = module.collect_evidence(INCIDENT_ID, JOB, _last_build())["jenkins"]["console_tail"]
    assert len(tail) == 4000


def _long_console() -> str:
    head = [
        "[Pipeline] stage: Unit Tests",
        "+ python -m pytest tests -q",
        "=================================== FAILURES ===================================",
        "E       assert 0 == 100",
        "FAILED tests/test_ledger.py::test_balance",
    ]
    noise = [f"[Pipeline] teardown {i} completed ok" for i in range(2000)]
    tail = [
        "java.io.IOException: Failed to archive artifacts",
        "ERROR: script returned exit code 1",
        "Finished: FAILURE",
    ]
    return "\n".join(head + noise + tail)


def test_console_failure_regions_capture_the_early_failure(monkeypatch) -> None:
    routes = _all_failing_routes()
    routes["/consoleText"] = FakeResponse(text=_long_console())
    _install(monkeypatch, routes)
    jenkins = module.collect_evidence(INCIDENT_ID, JOB, _last_build())["jenkins"]

    regions = jenkins["console_failures"]
    assert regions
    assert "FAILED tests/test_ledger.py::test_balance" in regions[0]["text"]
    assert regions[0]["line_number"] <= 5
    assert set(regions[0]) == {"marker", "line_number", "text"}
    assert "FAILED tests/test_ledger.py::test_balance" not in jenkins["console_tail"]
    assert "ERROR: script returned exit code 1" in jenkins["console_tail"]
    assert jenkins["console_truncated"] is False


def test_console_read_is_bounded_but_keeps_head_and_tail(monkeypatch) -> None:
    text = "\n".join(
        ["ERROR: early boom"] + ["filler line that is long enough to add up" * 3] * 60000 + ["ERROR: late boom"]
    )
    assert len(text) > module._CONSOLE_MAX_CHARS
    routes = _all_failing_routes()
    routes["/consoleText"] = FakeResponse(text=text)
    _install(monkeypatch, routes)
    jenkins = module.collect_evidence(INCIDENT_ID, JOB, _last_build())["jenkins"]
    assert "ERROR: early boom" in jenkins["console_failures"][0]["text"]
    assert "ERROR: late boom" in jenkins["console_tail"]


def test_failed_tests_are_capped_and_error_details_truncated(monkeypatch) -> None:
    cases = [
        {"name": f"t{i}", "className": "demo.Case", "status": "FAILED", "errorDetails": "x" * 5000}
        for i in range(15)
    ]
    routes = _all_failing_routes()
    routes["/testReport/api/json"] = FakeResponse({"suites": [{"cases": cases}]})
    _install(monkeypatch, routes)
    failed = module.collect_evidence(INCIDENT_ID, JOB, _last_build())["jenkins"]["failed_tests"]
    assert len(failed) == 10
    assert all(len(test["errorDetails"]) == 2000 for test in failed)


def test_window_falls_back_when_timestamp_missing(monkeypatch) -> None:
    _install(monkeypatch, _all_failing_routes())
    bundle = module.collect_evidence(INCIDENT_ID, JOB, _last_build(timestamp=None, duration=None))
    timestamps = bundle["jenkins"]["timestamps"]
    assert timestamps["start"] == timestamps["end"]
    assert bundle["jenkins"]["duration_seconds"] == 0.0


def test_window_end_uses_now_when_duration_missing(monkeypatch) -> None:
    _install(monkeypatch, _all_failing_routes())
    bundle = module.collect_evidence(INCIDENT_ID, JOB, _last_build(duration=0))
    timestamps = bundle["jenkins"]["timestamps"]
    assert timestamps["start"] == "2024-07-03T09:46:40+00:00"
    assert timestamps["end"] > timestamps["start"]


def test_evidence_tolerates_odd_jenkins_payloads(monkeypatch) -> None:
    routes = {
        "/wfapi/describe": FakeResponse({"stages": [{"name": "Build", "status": "SUCCESS"}]}),
        "/testReport/api/json": FakeResponse(
            {"suites": ["bad", {"cases": [123, {"name": "t", "className": "c", "status": "FAILED"}]}]}
        ),
        "/consoleText": RuntimeError("no console"),
    }
    _install(monkeypatch, routes)
    bundle = module.collect_evidence(INCIDENT_ID, JOB, _last_build(number="not-a-number"))
    jenkins = bundle["jenkins"]
    assert jenkins["build_number"] == 0
    assert jenkins["first_failed_stage"] is None
    assert jenkins["failed_tests"] == [
        {"name": "t", "className": "c", "errorDetails": None, "errorStackTrace": None}
    ]


def _bundle(failed_tests=(), console_tail="", records=(), console_failures=()):  # type: ignore[no-untyped-def]
    return {
        "jenkins": {
            "failed_tests": list(failed_tests),
            "console_tail": console_tail,
            "console_failures": list(console_failures),
        },
        "log_evidence": {"records": list(records)},
    }


def test_signature_from_console_failure_region() -> None:
    bundle = _bundle(
        console_tail="ERROR: script returned exit code 1",
        console_failures=[
            {"marker": "FAILED ", "line_number": 12, "text": "FAILED hermes_demo_test_failure check"},
            "not-a-region",
        ],
    )
    assert module.evidence_has_signature(bundle, INCIDENT_ID) is True


def test_signature_from_failed_test_name() -> None:
    bundle = _bundle(failed_tests=[{"name": "fixture-state-check", "className": "c", "errorDetails": None}])
    assert module.evidence_has_signature(bundle, INCIDENT_ID) is True


def test_signature_from_console_marker() -> None:
    bundle = _bundle(console_tail="... hermes_demo_test_failure ...")
    assert module.evidence_has_signature(bundle, INCIDENT_ID) is True


def test_signature_from_log_records_with_incident_id() -> None:
    bundle = _bundle(
        records=[
            {"message": "seed hermes_demo_test_failure marker"},
            {"message": f"incident {INCIDENT_ID} correlated"},
        ]
    )
    assert module.evidence_has_signature(bundle, INCIDENT_ID) is True


def test_signature_marker_without_incident_id_is_not_enough() -> None:
    bundle = _bundle(records=[{"message": "hermes_demo_test_failure only"}])
    assert module.evidence_has_signature(bundle, INCIDENT_ID) is False


def test_signature_absent() -> None:
    bundle = _bundle(
        failed_tests=[{"name": "other-test", "className": "c", "errorDetails": None}],
        console_tail="all quiet",
        records=[{"message": "normal log"}],
    )
    assert module.evidence_has_signature(bundle, INCIDENT_ID) is False


def test_signature_tolerates_malformed_bundle() -> None:
    assert module.evidence_has_signature({}, INCIDENT_ID) is False


def _settings_with(namespaces):  # type: ignore[no-untyped-def]
    cfg = _settings()
    cfg.hermes_job_namespaces = namespaces
    return cfg


def test_log_namespaces_always_include_jenkins(monkeypatch) -> None:
    """CI failures happen in Jenkins agent pods, whatever the service."""

    monkeypatch.setattr(module, "settings", _settings_with({}))
    assert module._log_config("lesavka")["extra_namespaces"] == ["jenkins"]


def test_a_mapped_job_also_searches_its_own_namespace(monkeypatch) -> None:
    """A build failure can correlate with the service itself being unhealthy."""

    monkeypatch.setattr(module, "settings", _settings_with({"bstein-dev-home": "bstein-dev-home"}))
    spaces = module._log_config("bstein-dev-home")["extra_namespaces"]
    assert "bstein-dev-home" in spaces
    assert "jenkins" in spaces


def test_an_unmapped_job_adds_nothing(monkeypatch) -> None:
    monkeypatch.setattr(module, "settings", _settings_with({"other": "other-ns"}))
    assert module._log_config("lesavka")["extra_namespaces"] == ["jenkins"]


def test_a_namespace_is_never_duplicated(monkeypatch) -> None:
    monkeypatch.setattr(module, "settings", _settings_with({"j": "jenkins"}))
    assert module._log_config("j")["extra_namespaces"].count("jenkins") == 1



def test_the_stack_trace_is_collected_and_bounded(monkeypatch) -> None:
    """The traceback names the line; the assertion alone does not."""

    case = {
        "name": "t",
        "className": "c",
        "status": "FAILED",
        "errorDetails": "AssertionError",
        "errorStackTrace": "T" * 9000,
    }
    routes = _all_failing_routes()
    routes["/testReport/api/json"] = FakeResponse({"suites": [{"cases": [case]}]})
    calls = _install(monkeypatch, routes)

    failed = module.collect_evidence(INCIDENT_ID, JOB, _last_build())["jenkins"]["failed_tests"]

    assert len(failed[0]["errorStackTrace"]) == module._MAX_STACK_TRACE_CHARS
    trees = [str(params.get("tree", "")) for _url, params in calls["gets"] if params]
    assert any("errorStackTrace" in tree for tree in trees)