prole/tests/installer/test_monitoring_ops.py

from __future__ import annotations

import json
from subprocess import CompletedProcess

import pytest

from knoe.core.ops import monitoring


def _cp(*, stdout: str = "", stderr: str = "", returncode: int = 0) -> CompletedProcess:
    return CompletedProcess(args=["helm"], returncode=returncode, stdout=stdout, stderr=stderr)


def test_monitoring_update_recovers_stale_prometheus_pending_lock_and_upgrades_once(monkeypatch):
    helm_calls: list[list[str]] = []
    history_poll_count = 0
    status_call_count = 0

    def fake_kubectl(_args, **_kwargs):
        return _cp()

    def fake_helm(args, **_kwargs):
        nonlocal history_poll_count, status_call_count
        helm_calls.append(list(args))

        if args[:2] == ["repo", "add"]:
            return _cp()
        if args[:2] == ["repo", "update"]:
            return _cp()
        if args[:4] == ["status", "prometheus", "-n", "monitoring"]:
            status_call_count += 1
            status = "pending-upgrade" if status_call_count == 1 else "deployed"
            return _cp(stdout=json.dumps({"info": {"status": status}}))
        if args[:4] == ["history", "prometheus", "-n", "monitoring"] and "-o" in args:
            history_poll_count += 1
            history = [
                {"revision": 1, "status": "deployed"},
                {"revision": 2, "status": "pending-upgrade"},
            ]
            return _cp(stdout=json.dumps(history))
        if args[:2] == ["rollback", "prometheus"]:
            return _cp(stdout="rolled back")
        if args[:2] == ["upgrade", "--install"]:
            return _cp(stdout="upgrade ok")
        raise AssertionError(f"Unexpected helm args: {args}")

    monkeypatch.setattr(monitoring, "_kubectl", fake_kubectl)
    monkeypatch.setattr(monitoring, "_helm", fake_helm)
    monkeypatch.setattr(monitoring.time, "sleep", lambda _seconds: None)

    env = {
        "KNOE_MODE": "k8s",
        "APP_CLUSTER_KUBECONTEXT": "ctx-app",
        "KUBECONTEXT": "ctx-app",
    }
    monitoring.update(env=env)

    rollback_idx = next(i for i, call in enumerate(helm_calls) if call[:2] == ["rollback", "prometheus"])
    upgrade_idx = next(i for i, call in enumerate(helm_calls) if call[:2] == ["upgrade", "--install"])
    assert rollback_idx < upgrade_idx
    assert helm_calls[rollback_idx] == ["rollback", "prometheus", "1", "-n", "monitoring"]
    assert sum(1 for call in helm_calls if call[:2] == ["upgrade", "--install"]) == 1
    assert history_poll_count == 3


def test_monitoring_update_blocks_when_stale_pending_lock_has_no_deployed_revision(monkeypatch):
    helm_calls: list[list[str]] = []

    def fake_kubectl(_args, **_kwargs):
        return _cp()

    def fake_helm(args, **_kwargs):
        helm_calls.append(list(args))

        if args[:2] == ["repo", "add"]:
            return _cp()
        if args[:2] == ["repo", "update"]:
            return _cp()
        if args[:4] == ["status", "prometheus", "-n", "monitoring"]:
            return _cp(stdout=json.dumps({"info": {"status": "pending-install"}}))
        if args[:4] == ["history", "prometheus", "-n", "monitoring"] and "-o" in args:
            return _cp(stdout=json.dumps([{"revision": 1, "status": "pending-install"}]))
        if args[:4] == ["history", "prometheus", "-n", "monitoring"]:
            return _cp(stdout="REVISION STATUS\n1 pending-install\n")
        if args[:2] == ["upgrade", "--install"]:
            return _cp(stdout="upgrade should not run")
        raise AssertionError(f"Unexpected helm args: {args}")

    monkeypatch.setattr(monitoring, "_kubectl", fake_kubectl)
    monkeypatch.setattr(monitoring, "_helm", fake_helm)
    monkeypatch.setattr(monitoring.time, "sleep", lambda _seconds: None)

    env = {
        "KNOE_MODE": "k8s",
        "APP_CLUSTER_KUBECONTEXT": "ctx-app",
        "KUBECONTEXT": "ctx-app",
    }

    with pytest.raises(RuntimeError, match="REPAIR_BLOCKED") as exc:
        monitoring.update(env=env)

    message = str(exc.value)
    assert "helm history output" in message
    assert "pending-install" in message
    assert not any(call[:2] == ["upgrade", "--install"] for call in helm_calls)


def test_monitoring_update_skips_stale_lock_recovery_outside_app_context(monkeypatch):
    helm_calls: list[list[str]] = []

    def fake_kubectl(_args, **_kwargs):
        return _cp()

    def fake_helm(args, **_kwargs):
        helm_calls.append(list(args))

        if args[:2] == ["repo", "add"]:
            return _cp()
        if args[:2] == ["repo", "update"]:
            return _cp()
        if args[:2] == ["upgrade", "--install"]:
            return _cp(stdout="upgrade ok")
        if args and args[0] in {"status", "history", "rollback"}:
            raise AssertionError(f"Stale-lock command should not run outside APP context: {args}")
        raise AssertionError(f"Unexpected helm args: {args}")

    monkeypatch.setattr(monitoring, "_kubectl", fake_kubectl)
    monkeypatch.setattr(monitoring, "_helm", fake_helm)

    env = {
        "KNOE_MODE": "k8s",
        "APP_CLUSTER_KUBECONTEXT": "ctx-app",
        "KUBECONTEXT": "ctx-db",
    }
    monitoring.update(env=env)

    assert sum(1 for call in helm_calls if call[:2] == ["upgrade", "--install"]) == 1