prole/tests/test_deployment_mode_isolation.py

"""Regression tests: k3d / k3s / gke mode-specific changes must not bleed into each other."""
from __future__ import annotations

import pathlib
import tempfile

import pytest


# ---------------------------------------------------------------------------
# 7a — milestone._get_script_env SERVICE_NAMESPACE must not fall back to DB ns
# ---------------------------------------------------------------------------

def test_get_script_env_service_namespace_does_not_fall_back_to_db_ns(monkeypatch):
    """Regression: SERVICE_NAMESPACE must not default to the DB namespace.

    milestone._get_script_env formerly fell back to env["NAMESPACE"] (the DB
    namespace) when SERVICE_NAMESPACE was absent from config and OS env.
    This caused Kong to deploy into the DB namespace and collide with the
    already-deployed knoe-system/svc-knoe-ingress.
    """
    monkeypatch.delenv("SERVICE_NAMESPACE", raising=False)

    class _FakeState:
        inputs = {"init_password.db_namespace": "knoe-db", "init_cluster.cluster_env": "dev"}
        config_data = {"Global": {}}  # no SERVICE_NAMESPACE

    from knoe.milestone import Milestone

    class _M(Milestone):
        def execute(self, state, progress=None):
            pass

    m = _M.__new__(_M)
    env = m._get_script_env(_FakeState())
    assert env["SERVICE_NAMESPACE"] != "knoe-db", (
        "SERVICE_NAMESPACE must not fall back to the DB namespace"
    )
    assert env["SERVICE_NAMESPACE"] == "knoe-system"


# ---------------------------------------------------------------------------
# 7b — monitoring values YAML is mode-specific
# ---------------------------------------------------------------------------

def test_monitoring_k3d_values_use_local_path_storage():
    from knoe.core.ops.monitoring import _values_yaml_k3d

    yaml_str = _values_yaml_k3d("", {})
    assert "local-path" in yaml_str
    assert "merlin-local-iscsi" not in yaml_str
    assert "pi.prole.org" not in yaml_str  # no k3s-specific node affinity


def test_monitoring_k3s_values_do_not_use_local_path():
    from knoe.core.ops.monitoring import _values_yaml_k3s

    yaml_str = _values_yaml_k3s("", {})
    assert "merlin-local-iscsi" in yaml_str
    assert "local-path" not in yaml_str


# ---------------------------------------------------------------------------
# 7c — init_kong.sh gitea host disabled for k3d and k8s, enabled for k3s
# ---------------------------------------------------------------------------

def test_kong_gitea_host_excluded_for_k3d_and_k8s():
    text = pathlib.Path("etc/init_kong.sh").read_text(encoding="utf-8")
    # k3d and k8s branches set include_gitea_host=0
    assert "include_gitea_host=0" in text
    # k3s starts with include_gitea_host=1 (default)
    assert "include_gitea_host=1" in text
    # The mode guard must mention both k8s and k3d explicitly
    assert '"k8s"' in text or "k8s" in text
    assert '"k3d"' in text or "k3d" in text


# ---------------------------------------------------------------------------
# 7d — _service_namespace() always returns knoe-system regardless of mode
# ---------------------------------------------------------------------------

@pytest.mark.parametrize("cluster_env,mode", [
    ("dev", "k3d"),
    ("service", "k3s"),
    ("prod", "k8s"),
])
def test_service_namespace_defaults_to_knoe_system_for_all_modes(
    cluster_env, mode, monkeypatch
):
    monkeypatch.delenv("SERVICE_NAMESPACE", raising=False)

    from knoe.core.actions import KnoeConsoleInstaller
    from knoe.core.controller import KnoeController

    with tempfile.TemporaryDirectory() as td:
        c = KnoeController(pathlib.Path(td))
        inst = KnoeConsoleInstaller(c)
        inst.knoe_cfg_data = {"Global": {}}
        inst.inputs["init_cluster.cluster_env"] = cluster_env
        assert inst._service_namespace() == "knoe-system", (
            f"_service_namespace() returned wrong value for mode={mode}"
        )


# ---------------------------------------------------------------------------
# 7e — port-forward builder: argocd/dashboard conditional; release-aware names
# ---------------------------------------------------------------------------

def test_port_forward_argocd_excluded_when_disabled():
    from knoe.core.env import _build_required_port_forwards

    mappings = _build_required_port_forwards(
        mode="k3d",
        service_ns="knoe-system",
        argocd_ns="argocd",
        db_ns="knoe-db",
        db_host_port="5432",
        supabase_enabled=False,
        supabase_namespace="supabase",
        argocd_enabled=False,
        dashboard_enabled=False,
    )
    ids = [m.split("id=", 1)[1].split(";", 1)[0] for m in mappings if "id=" in m]
    assert "argocd" not in ids, "argocd must be excluded when argocd_enabled=False"
    assert "dashboard" not in ids, "dashboard must be excluded when dashboard_enabled=False"


# ---------------------------------------------------------------------------
# 7f — CNPG initialize() fast-path: skip when cluster already healthy
# ---------------------------------------------------------------------------

def test_cnpg_initialize_skips_when_cluster_healthy(monkeypatch):
    """Regression: initialize() must return immediately when the cluster is healthy.

    Formerly it always ran install_barman_plugin() which fetches from GitHub
    (up to 6 × 120 s), stalling the installer even when CNPG was already running.
    """
    import types
    from knoe.core.ops import cloudnative_pg as cnpg

    calls: list[str] = []

    def _fake_kubectl(args, env=None, timeout=None):
        cmd = " ".join(str(a) for a in args)
        calls.append(cmd)
        result = types.SimpleNamespace(returncode=0, stdout="", stderr="")
        if "jsonpath={.status.phase}" in cmd:
            result.stdout = "Cluster in healthy state"
        elif "get pods" in cmd:
            result.stdout = "knoe-db-1  2/2  Running  0  5d\nknoe-db-2  2/2  Running  0  5d\n"
        return result

    monkeypatch.setattr(cnpg, "_kubectl", _fake_kubectl)

    logged: list[str] = []
    cnpg.initialize("knoe-db", "knoe-db", log=lambda msg: logged.append(msg))

    assert any("skipping reinitialize" in m for m in logged), (
        "Expected fast-path log message not found"
    )
    # ensure_operator / install_barman_plugin / _apply_manifest must NOT be called
    assert not any("apply" in c and "barman" in c.lower() for c in calls), (
        "install_barman_plugin must not be called when cluster is already healthy"
    )


def test_port_forward_grafana_uses_monitoring_release():
    from knoe.core.env import _build_required_port_forwards

    mappings = _build_required_port_forwards(
        mode="k3d",
        service_ns="knoe-system",
        argocd_ns="argocd",
        db_ns="knoe-db",
        db_host_port="5432",
        supabase_enabled=False,
        supabase_namespace="supabase",
        monitoring_release="prometheus",
    )
    grafana = next((m for m in mappings if "id=grafana;" in m), None)
    assert grafana is not None, "grafana mapping must be present"
    assert "svc/prometheus-grafana" in grafana, (
        "grafana target must use the monitoring release name"
    )
    assert "svc/kps-grafana" not in grafana