ariadne/tests/test_hermes_autotriage_repair.py

from __future__ import annotations

from types import SimpleNamespace

from ariadne.services import hermes_autotriage_repair as module


INCIDENT_ID = "hermes-triage-demo/12"


class FakeClock:
    def __init__(self) -> None:
        self.now = 0.0
        self.sleeps: list[float] = []

    def time(self) -> float:
        return self.now

    def sleep(self, seconds: float) -> None:
        self.sleeps.append(seconds)
        self.now += seconds


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


def _cfg(**overrides) -> dict:  # type: ignore[no-untyped-def]
    base = {
        "namespace": "hermes-triage-demo",
        "fixture_pvc": "hermes-triage-demo-fixture",
        "image": "busybox:1.37",
    }
    base.update(overrides)
    return base


def _jenkins_settings(**overrides) -> SimpleNamespace:  # type: ignore[no-untyped-def]
    values = {
        "jenkins_base_url": "https://ci.example",
        "jenkins_api_user": "user",
        "jenkins_api_token": "token",
        "jenkins_api_timeout_sec": 5.0,
    }
    values.update(overrides)
    return SimpleNamespace(**values)


def _install_k8s(monkeypatch, statuses, post_exc=None) -> dict:  # type: ignore[no-untyped-def]
    calls: dict = {"posts": [], "gets": []}
    queue = list(statuses)

    def fake_post(path, payload):  # type: ignore[no-untyped-def]
        calls["posts"].append((path, payload))
        if post_exc is not None:
            raise post_exc
        return {"metadata": {"name": payload["metadata"]["name"]}}

    def fake_get(path):  # type: ignore[no-untyped-def]
        calls["gets"].append(path)
        item = queue.pop(0) if queue else {"status": {"active": 1}}
        if isinstance(item, Exception):
            raise item
        return item

    monkeypatch.setattr(module, "post_json", fake_post)
    monkeypatch.setattr(module, "get_json", fake_get)
    clock = FakeClock()
    monkeypatch.setattr(module, "time", clock)
    calls["clock"] = clock
    return calls


def test_repair_job_payload_contract(monkeypatch) -> None:
    calls = _install_k8s(monkeypatch, [{"status": {"succeeded": 1}}])
    result = module.execute_repair(_cfg(), INCIDENT_ID, 12)

    assert result == {"job_name": "hermes-demo-repair-12", "succeeded": True, "error": None}
    path, payload = calls["posts"][0]
    assert path == "/apis/batch/v1/namespaces/hermes-triage-demo/jobs"
    assert payload["apiVersion"] == "batch/v1"
    assert payload["kind"] == "Job"
    assert payload["metadata"]["name"] == "hermes-demo-repair-12"
    assert payload["metadata"]["namespace"] == "hermes-triage-demo"
    assert payload["metadata"]["labels"] == {
        "atlas.bstein.dev/trigger": "ariadne",
        "app.kubernetes.io/part-of": "hermes-triage-demo",
    }
    spec = payload["spec"]
    assert spec["backoffLimit"] == 0
    assert spec["ttlSecondsAfterFinished"] == 3600
    pod = spec["template"]["spec"]
    assert pod["restartPolicy"] == "Never"
    container = pod["containers"][0]
    assert container["image"] == "busybox:1.37"
    assert container["command"][:2] == ["sh", "-c"]
    script = container["command"][2]
    assert "printf 'healthy' > /fixture/state" in script
    assert '"event":"hermes_demo_repair"' in script
    assert f'"incident_id":"{INCIDENT_ID}"' in script
    assert '"message":"fixture state reset to healthy"' in script
    assert container["volumeMounts"] == [{"name": "fixture", "mountPath": "/fixture"}]
    assert pod["volumes"] == [
        {"name": "fixture", "persistentVolumeClaim": {"claimName": "hermes-triage-demo-fixture"}}
    ]


def test_execute_repair_waits_through_active_polls(monkeypatch) -> None:
    calls = _install_k8s(
        monkeypatch,
        [{"status": {"active": 1}}, {"status": {"succeeded": 1}}],
    )
    result = module.execute_repair(_cfg(), INCIDENT_ID, 12)
    assert result["succeeded"] is True
    assert len(calls["gets"]) == 2
    assert calls["clock"].sleeps == [2.0]


def test_execute_repair_job_failure(monkeypatch) -> None:
    _install_k8s(monkeypatch, [{"status": {"failed": 1}}])
    result = module.execute_repair(_cfg(), INCIDENT_ID, 12)
    assert result == {"job_name": "hermes-demo-repair-12", "succeeded": False, "error": "repair job failed"}


def test_execute_repair_timeout(monkeypatch) -> None:
    _install_k8s(monkeypatch, [])
    result = module.execute_repair(_cfg(wait_timeout_seconds=6), INCIDENT_ID, 12)
    assert result["succeeded"] is False
    assert result["error"] == "repair job timeout after 6.0s"


def test_execute_repair_duplicate_job(monkeypatch) -> None:
    conflict = RuntimeError("conflict")
    conflict.response = SimpleNamespace(status_code=409)  # type: ignore[attr-defined]
    _install_k8s(monkeypatch, [], post_exc=conflict)
    result = module.execute_repair(_cfg(), INCIDENT_ID, 12)
    assert result == {"job_name": "hermes-demo-repair-12", "succeeded": False, "error": "duplicate job"}


def test_execute_repair_create_error(monkeypatch) -> None:
    _install_k8s(monkeypatch, [], post_exc=ValueError("nope"))
    result = module.execute_repair(_cfg(), INCIDENT_ID, 12)
    assert result["succeeded"] is False
    assert result["error"] == "job create failed: nope"


def test_execute_repair_status_read_error(monkeypatch) -> None:
    _install_k8s(monkeypatch, [RuntimeError("api down")])
    result = module.execute_repair(_cfg(), INCIDENT_ID, 12)
    assert result["succeeded"] is False
    assert result["error"] == "job status read failed: api down"


def _install_http(monkeypatch, response=None, exc=None) -> dict:  # type: ignore[no-untyped-def]
    calls: dict = {"posts": [], "kwargs": None}

    class FakeClient:
        def __init__(self, **kwargs) -> None:  # type: ignore[no-untyped-def]
            calls["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 post(self, url, data=None):  # type: ignore[no-untyped-def]
            calls["posts"].append((url, data))
            if exc is not None:
                raise exc
            return response

    monkeypatch.setattr(module.httpx, "Client", FakeClient)
    return calls


def test_trigger_rebuild_success(monkeypatch) -> None:
    calls = _install_http(monkeypatch, response=FakeResponse(201))
    result = module.trigger_rebuild(_jenkins_settings(), "hermes-triage-demo")
    assert result == {"requested": True, "error": None}
    url, data = calls["posts"][0]
    assert url == "https://ci.example/job/hermes-triage-demo/buildWithParameters"
    assert data == {"SEED_FAILURE": "false"}
    assert calls["kwargs"]["auth"] == ("user", "token")
    assert calls["kwargs"]["timeout"] == 5.0


def test_trigger_rebuild_non_created_status(monkeypatch) -> None:
    _install_http(monkeypatch, response=FakeResponse(500))
    result = module.trigger_rebuild(_jenkins_settings(), "hermes-triage-demo")
    assert result == {"requested": False, "error": "rebuild http 500"}


def test_trigger_rebuild_request_failure(monkeypatch) -> None:
    _install_http(monkeypatch, exc=RuntimeError("connect refused"))
    result = module.trigger_rebuild(_jenkins_settings(), "hermes-triage-demo")
    assert result["requested"] is False
    assert result["error"] == "rebuild request failed: connect refused"


def test_trigger_rebuild_without_base_url(monkeypatch) -> None:
    calls = _install_http(monkeypatch, response=FakeResponse(201))
    result = module.trigger_rebuild(_jenkins_settings(jenkins_base_url=""), "hermes-triage-demo")
    assert result == {"requested": False, "error": "jenkins base url is empty"}
    assert calls["posts"] == []


def test_trigger_rebuild_without_credentials(monkeypatch) -> None:
    calls = _install_http(monkeypatch, response=FakeResponse(201))
    config = _jenkins_settings(jenkins_api_user="", jenkins_api_token="")
    result = module.trigger_rebuild(config, "hermes-triage-demo")
    assert result["requested"] is True
    assert "auth" not in calls["kwargs"]