prole/tests/test_cluster_screen_observe_only.py

from __future__ import annotations


import json


from knoe.ui.screens.cluster import ClusterScreenMixin


class _Var:
    def __init__(self, value: str = ""):
        self._value = value

    def get(self) -> str:
        return self._value

    def set(self, value: str) -> None:
        self._value = value


class _Root:
    def after(self, _delay_ms: int, fn):
        fn()


class _Canvas:
    def itemconfig(self, *_args, **_kwargs):
        return None


class _Button:
    def configure(self, **_kwargs):
        return None


def test_cluster_status_refresh_is_observe_only(monkeypatch):
    """Regression: simply navigating/refreshing the Cluster Environment screen must not mutate the cluster."""

    class _ImmediateThread:
        def __init__(self, *, target=None, daemon=None):
            self._target = target

        def start(self):
            # Execute synchronously so exceptions surface to the test.
            if self._target:
                self._target()

    monkeypatch.setattr(
        "knoe.ui.screens.cluster.threading.Thread", _ImmediateThread, raising=True
    )

    class Dummy(ClusterScreenMixin):
        def __init__(self):
            self.root = _Root()
            self.bg_canvas = _Canvas()
            self.k3d_status_label = 1
            self.common_services_status_label = 2
            self._init_cluster_button = _Button()
            self._kong_last_repair_at = 0.0
            self._kong_repair_cooldown_s = 0.0
            self._repair_inflight = False
            self._repair_last_run_at = 0.0
            self._repair_cooldown_s = 0.0

        def _set_cluster_env_message(self, *_args, **_kwargs):
            return None

        def _cluster_status_snapshot(self) -> dict:
            return {
                "env": "service",
                "cluster_ok": True,
                "cluster_msg": "ok",
                "cluster_fill": "#34c759",
                "common_ok": True,
                "common_msg": "ok",
                "common_fill": "#34c759",
            }

        def _dashboard_kong_status(self, _env=None):
            # Unhealthy, but observe-only behavior must not attempt repairs.
            return {"ok": False, "reset_pods": ["pod-1"]}

        def _authority_context_missing(self) -> bool:
            return True

        def _maybe_run_repair_pipeline(self, *_args, **_kwargs):
            raise AssertionError("repair pipeline must not run during status refresh")

    Dummy().check_cluster_status_async()


def test_service_namespace_infers_registry_namespace(monkeypatch):
    """If cluster is reachable and `registry:2` exists, infer that namespace instead of defaulting to `default`."""

    class Dummy(ClusterScreenMixin):
        def __init__(self):
            self.service_namespace = _Var("")
            self.cluster_env = _Var("prod")
            self.prole_cfg_data = {"Global": {}}

        def _ensure_k3s_kubeconfig_merged(self):
            return None

        def _deployment_mode(self) -> str:
            return "k8s"

        def _kubectl_base_cmd(self, _mode=None):
            return ["kubectl"]

    payload = {
        "items": [
            {
                "metadata": {"namespace": "knoe-system"},
                "spec": {
                    "template": {
                        "spec": {"containers": [{"image": "registry:2"}]}
                    }
                },
            }
        ]
    }

    def fake_run(cmd, capture_output, text, env, timeout):
        assert cmd[:3] == ["kubectl", "get", "deploy"]

        class _Res:
            returncode = 0
            stdout = json.dumps(payload)
            stderr = ""

        return _Res()

    monkeypatch.setattr(
        "knoe.ui.screens.cluster.subprocess.run", fake_run, raising=True
    )
    # Avoid reliance on an actual kubeconfig in test environment.
    monkeypatch.setattr(
        "knoe.ui.screens.cluster._find_kubeconfig_file", lambda *_a, **_k: "", raising=True
    )

    assert Dummy()._get_service_namespace() == "knoe-system"


def test_service_namespace_rejects_context_like_value_and_falls_back():
    """Regression: kube context names must not be treated as Kubernetes namespaces."""

    class Dummy(ClusterScreenMixin):
        def __init__(self):
            self.service_namespace = _Var("knoe.dev.prole.org")
            self.cluster_env = _Var("service")
            self.prole_cfg_data = {"Global": {}}

        def _infer_common_services_namespace(self) -> str:
            return ""

    assert Dummy()._get_service_namespace() == "knoe-system"