prole/tests/installer/test_database_context_selection.py

from __future__ import annotations

from knoe.ui.screens.database import DatabaseScreenMixin


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

    def get(self):
        return self._value

    def set(self, value):
        self._value = value


class _DummyDbScreen(DatabaseScreenMixin):
    def __init__(self, *, inputs: dict | None = None, knoe_cfg_data: dict | None = None):
        self.inputs = inputs or {}
        self.knoe_cfg_data = knoe_cfg_data or {}
        self.selected_kubectx = _Var(
            (self.inputs.get("init_cluster.selected_kubectx") or "")
        )

    def _get_input(self, key: str, default=""):
        return self.inputs.get(key, default)

    def _deployment_mode(self):
        return "k8s"

    def _kubectl_base_cmd(self, mode: str | None = None):
        _ = mode
        return ["kubectl", "--context", "gke_proj_us-west3_knoe-dev-0"]


def test_db_kubectl_base_cmd_prefers_db_cluster_context_over_dev_context():
    screen = _DummyDbScreen(
        knoe_cfg_data={
            "Global": {
                "KUBECONTEXT": "gke_proj_us-west3_knoe-dev-0",
                "DB_CLUSTER_KUBECONTEXT": "gke_proj_us-west3_knoe-cnpg-0",
            }
        }
    )

    cmd = screen._db_kubectl_base_cmd()

    assert cmd == ["kubectl", "--context", "gke_proj_us-west3_knoe-cnpg-0"]


def test_resolve_db_kube_context_derives_cnpg_context_when_only_app_context_exists():
    screen = _DummyDbScreen(
        knoe_cfg_data={
            "Global": {
                "KUBECONTEXT": "gke_proj_us-west3_knoe-dev-0",
                "DB_CLUSTER_NAME": "knoe-cnpg-0",
            }
        }
    )

    assert screen._resolve_db_kube_context() == "gke_proj_us-west3_knoe-cnpg-0"


def test_ensure_db_screen_context_keeps_explicit_user_override():
    screen = _DummyDbScreen(
        inputs={"init_cluster.selected_kubectx": "gke_proj_us-west3_custom-ops"},
        knoe_cfg_data={
            "Global": {
                "KUBECONTEXT": "gke_proj_us-west3_knoe-dev-0",
                "DB_CLUSTER_KUBECONTEXT": "gke_proj_us-west3_knoe-cnpg-0",
            }
        },
    )

    screen._ensure_db_screen_context()

    assert screen.selected_kubectx.get() == "gke_proj_us-west3_custom-ops"


def test_db_target_note_includes_context_and_namespace_confirmation():
    screen = _DummyDbScreen(
        knoe_cfg_data={
            "Global": {
                "DB_CLUSTER_KUBECONTEXT": "gke_proj_us-west3_knoe-cnpg-0",
            }
        }
    )

    note = screen._db_target_note("knoe-db")

    assert "Target context: gke_proj_us-west3_knoe-cnpg-0" in note
    assert "Target namespace: knoe-db" in note