ariadne/tests/test_hermes_autotriage_actions.py

"""Tests for the actions Ariadne executes on its own authority."""

from __future__ import annotations

from types import SimpleNamespace

import pytest

from ariadne.services import hermes_autotriage_actions as module
from ariadne.services import hermes_autotriage_repair as repair


class _Storage:
    def __init__(self) -> None:
        self.events: list = []

    def record_event(self, event_type, detail):  # type: ignore[no-untyped-def]
        self.events.append((event_type, detail))

    def list_events(self, **_kwargs):  # type: ignore[no-untyped-def]
        return []


BASE = {"incident_id": "lesavka/9", "job": "lesavka", "build_number": 9}


def test_storage_reclaim_runs_the_cleanup_then_rebuilds(monkeypatch) -> None:
    """A rebuild alone would land on the same full volume."""

    calls = []
    monkeypatch.setattr(
        module.jenkins_workspace_cleanup, "cleanup_jenkins_workspace_storage",
        lambda: calls.append("cleanup") or SimpleNamespace(removed_pvcs=3),
    )
    monkeypatch.setattr(
        module.hermes_autotriage_repair, "trigger_rebuild",
        lambda cfg, job: calls.append(f"rebuild:{job}") or {"requested": True, "error": None},
    )
    storage = _Storage()
    result = module._reclaim_workspace_storage(storage, dict(BASE), "no space left on device")

    assert calls == ["cleanup", "rebuild:lesavka"]
    assert result["status"] == "awaiting_rebuild"
    assert result["reclaimed"] == 3


def test_a_failed_reclaim_never_rebuilds(monkeypatch) -> None:
    """Rebuilding onto a still-full volume would just fail again."""

    rebuilt = []
    monkeypatch.setattr(
        module.jenkins_workspace_cleanup, "cleanup_jenkins_workspace_storage",
        lambda: (_ for _ in ()).throw(RuntimeError("boom")),
    )
    monkeypatch.setattr(
        module.hermes_autotriage_repair, "trigger_rebuild",
        lambda cfg, job: rebuilt.append(job) or {"requested": True},
    )
    result = module._reclaim_workspace_storage(_Storage(), dict(BASE), "no space left")

    assert rebuilt == []
    assert result["status"] == "failed"


def test_a_failed_rebuild_after_reclaim_is_escalated(monkeypatch) -> None:
    monkeypatch.setattr(
        module.jenkins_workspace_cleanup, "cleanup_jenkins_workspace_storage",
        lambda: SimpleNamespace(removed_pvcs=1),
    )
    monkeypatch.setattr(
        module.hermes_autotriage_repair, "trigger_rebuild",
        lambda cfg, job: {"requested": False, "error": "http 500"},
    )
    result = module._reclaim_workspace_storage(_Storage(), dict(BASE), "no space left")

    assert result["status"] == "failed"


class _Response:
    def __init__(self, status_code: int) -> None:
        self.status_code = status_code


@pytest.mark.parametrize(("status", "requested"), [(302, True), (200, True), (404, False), (500, False)])
def test_abort_treats_any_non_error_status_as_success(monkeypatch, status, requested) -> None:
    """Jenkins answers the stop endpoint with a redirect, not 201."""

    class _Client:
        def __init__(self, **_kwargs):  # type: ignore[no-untyped-def]
            pass

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

        def __exit__(self, *_exc):  # type: ignore[no-untyped-def]
            return False

        def post(self, url, **_kwargs):  # type: ignore[no-untyped-def]
            assert url.endswith("/job/lesavka/9/stop")
            return _Response(status)

    monkeypatch.setattr(repair.httpx, "Client", _Client)
    cfg = SimpleNamespace(jenkins_base_url="https://ci.example", jenkins_api_user="u",
                          jenkins_api_token="t", jenkins_api_timeout_sec=5)
    assert repair.abort_build(cfg, "lesavka", 9)["requested"] is requested


def test_abort_without_a_base_url_reports_rather_than_raises() -> None:
    cfg = SimpleNamespace(jenkins_base_url="", jenkins_api_user="", jenkins_api_token="",
                          jenkins_api_timeout_sec=5)
    assert repair.abort_build(cfg, "j", 1) == {"requested": False, "error": "jenkins base url is empty"}


def test_abort_survives_a_transport_failure(monkeypatch) -> None:
    def _boom(**_kwargs):  # type: ignore[no-untyped-def]
        raise RuntimeError("connection reset")

    monkeypatch.setattr(repair.httpx, "Client", _boom)
    cfg = SimpleNamespace(jenkins_base_url="https://ci.example", jenkins_api_user="u",
                          jenkins_api_token="t", jenkins_api_timeout_sec=5)
    assert repair.abort_build(cfg, "j", 1)["requested"] is False


def test_clearing_stuck_agent_pods_then_rebuilds(monkeypatch) -> None:
    """A retry queued behind stuck pods fails the same way, so clear first."""

    calls = []
    monkeypatch.setattr(
        module.pod_cleaner, "clean_finished_pods",
        lambda: calls.append("clean") or SimpleNamespace(deleted=4),
    )
    monkeypatch.setattr(
        module.hermes_autotriage_repair, "trigger_rebuild",
        lambda cfg, job: calls.append(f"rebuild:{job}") or {"requested": True, "error": None},
    )
    result = module._clear_stuck_agent_pods(_Storage(), dict(BASE), "are offline")

    assert calls == ["clean", "rebuild:lesavka"]
    assert result["status"] == "awaiting_rebuild"
    assert result["cleared"] == 4


def test_a_failed_pod_clear_never_rebuilds(monkeypatch) -> None:
    rebuilt = []
    monkeypatch.setattr(
        module.pod_cleaner, "clean_finished_pods",
        lambda: (_ for _ in ()).throw(RuntimeError("boom")),
    )
    monkeypatch.setattr(
        module.hermes_autotriage_repair, "trigger_rebuild",
        lambda cfg, job: rebuilt.append(job) or {"requested": True},
    )
    assert module._clear_stuck_agent_pods(_Storage(), dict(BASE), "m")["status"] == "failed"
    assert rebuilt == []


def test_a_failed_rebuild_after_clearing_is_escalated(monkeypatch) -> None:
    monkeypatch.setattr(module.pod_cleaner, "clean_finished_pods", lambda: SimpleNamespace(deleted=1))
    monkeypatch.setattr(
        module.hermes_autotriage_repair, "trigger_rebuild",
        lambda cfg, job: {"requested": False, "error": "http 500"},
    )
    assert module._clear_stuck_agent_pods(_Storage(), dict(BASE), "m")["status"] == "failed"