prole/tests/installer/test_services_init_scripts.py

from __future__ import annotations


from unittest.mock import MagicMock


def test_tk_init_scripts_runner_invokes_init_certmgr(tmp_path, monkeypatch):
    import knoe.ui.screens.services as services
    from knoe.ui.screens.services import ServicesScreenMixin

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

        def get(self):
            return self._value

    class _Console:
        def __init__(self):
            self.buf: list[str] = []

        def clear(self):
            self.buf.clear()

        def write(self, content: str):
            self.buf.append(content)

    class _Widget:
        def winfo_exists(self):
            return False

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

    class _Canvas:
        def winfo_exists(self):
            return False

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

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

        def start(self):
            self._target()

    monkeypatch.setattr(services.threading, "Thread", _ImmediateThread)
    # CNPG is now Python-owned; patch to avoid real kubectl calls in this UI test
    monkeypatch.setattr(services, "cnpg_initialize", lambda **_kw: None)

    class _DummyScreen(ServicesScreenMixin):
        KUBECTL_STATUS_TAB = "__kubectl_status__"

        def __init__(self):
            self._action_flags = {}
            self.controller = MagicMock()
            self.controller.run_script.return_value = 0

            self._init_scripts_button = _Widget()
            self.bg_canvas = _Canvas()
            self._init_scripts_status_label = 0

            self._script_tab_index = {}
            self.script_consoles = {
                "init_common_services.sh": _Console(),
                "common_services_log": _Console(),
                "init_certmgr.sh": _Console(),
                "init_cloudnative_pg.sh": _Console(),
                "init_cnpg_backup.sh": _Console(),
                "init_kong.sh": _Console(),
                "init_monitoring.sh": _Console(),
                "init_nginx_ingress.sh": _Console(),
                "init_port_forwards.sh": _Console(),
                self.KUBECTL_STATUS_TAB: _Console(),
            }

            self.db_username = _Var("prole")
            self.db_password = _Var("pw")
            self.db_namespace = _Var("default")

            self.kerberos_enabled = _Var(False)
            self.kerberos_realm = _Var("")
            self.kerberos_kdc = _Var("")
            self.kerberos_user = _Var("")
            self.kerberos_password = _Var("")
            self.cluster_env = _Var("prole-service-cluster")

            self.prole_cfg_data = {"Initialization Scripts": {"STATUS": ""}}
            self._scripts_success = False

        def safe_after(self, fn, delay: int = 0):
            return fn()

        def _deployment_mode(self):
            return "k3s"

        def _resolve_prole_logs_dir(self):
            return tmp_path

        def _get_service_namespace(self):
            return "default"

        def _sync_port_forward_mappings(self):
            return None

        def _update_legacy_port_mapping(self):
            return None

        def _record_install_log(self, _path):
            return None

        def _process_script_output_line(self, _line: str):
            return None

        def _run_cmd_capture(self, _args):
            return 0, ""

        def _save_prole_cfg(self):
            return None

        def update_footer(self):
            return None

        def check_services_status_async(self):
            return None

    screen = _DummyScreen()
    screen.run_init_scripts()

    called_scripts = [c.args[0] for c in screen.controller.run_script.call_args_list]
    assert "init_certmgr.sh" in called_scripts
    # CNPG is now Python-owned; shell script must never be dispatched
    assert "init_cloudnative_pg.sh" not in called_scripts
    assert screen.prole_cfg_data["Initialization Scripts"]["STATUS"] == "Completed"


def test_init_monitoring_grafana_password_capture_does_not_wipe_monitoring_data_dir(
    tmp_path, monkeypatch
):
    """Regression: capturing Grafana admin password must not wipe other Monitoring values."""

    import knoe.ui.screens.services as services
    from knoe.ui.screens.services import ServicesScreenMixin

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

        def get(self):
            return self._value

    class _Console:
        def __init__(self):
            self.buf: list[str] = []

        def clear(self):
            self.buf.clear()

        def write(self, content: str):
            self.buf.append(content)

    class _Widget:
        def winfo_exists(self):
            return False

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

    class _Canvas:
        def winfo_exists(self):
            return False

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

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

        def start(self):
            self._target()

    monkeypatch.setattr(services.threading, "Thread", _ImmediateThread)
    # CNPG is now Python-owned; patch to avoid real kubectl calls in this UI test
    monkeypatch.setattr(services, "cnpg_initialize", lambda **_kw: None)

    class _DummyScreen(ServicesScreenMixin):
        KUBECTL_STATUS_TAB = "__kubectl_status__"

        def __init__(self):
            self._action_flags = {}
            self.controller = MagicMock()

            def _run_script_side_effect(script, *args, **kwargs):
                on_line = kwargs.get("on_line")
                if script == "init_monitoring.sh" and on_line is not None:
                    on_line("GRAFANA_ADMIN_PASSWORD=abc123\n")
                return 0

            self.controller.run_script.side_effect = _run_script_side_effect

            self._init_scripts_button = _Widget()
            self.bg_canvas = _Canvas()
            self._init_scripts_status_label = 0

            self._script_tab_index = {}
            self.script_consoles = {
                "init_common_services.sh": _Console(),
                "common_services_log": _Console(),
                "init_certmgr.sh": _Console(),
                "init_cloudnative_pg.sh": _Console(),
                "init_cnpg_backup.sh": _Console(),
                "init_kong.sh": _Console(),
                "init_monitoring.sh": _Console(),
                "init_nginx_ingress.sh": _Console(),
                "init_port_forwards.sh": _Console(),
                self.KUBECTL_STATUS_TAB: _Console(),
            }

            self.db_username = _Var("prole")
            self.db_password = _Var("pw")
            self.db_namespace = _Var("default")

            self.kerberos_enabled = _Var(False)
            self.kerberos_realm = _Var("")
            self.kerberos_kdc = _Var("")
            self.kerberos_user = _Var("")
            self.kerberos_password = _Var("")
            self.cluster_env = _Var("prole-service-cluster")

            self.prole_cfg_data = {
                "Initialization Scripts": {"STATUS": ""},
                "Monitoring": {"EXISTING_MONITORING_KEY": "keep-me"},
            }
            self._scripts_success = False

        def safe_after(self, fn, delay: int = 0):
            return fn()

        def _deployment_mode(self):
            return "k3s"

        def _resolve_prole_logs_dir(self):
            return tmp_path

        def _get_service_namespace(self):
            return "default"

        def _sync_port_forward_mappings(self):
            return None

        def _update_legacy_port_mapping(self):
            return None

        def _record_install_log(self, _path):
            return None

        def _process_script_output_line(self, _line: str):
            return None

        def _run_cmd_capture(self, _args):
            return 0, ""

        def _save_prole_cfg(self):
            return None

        def update_footer(self):
            return None

        def check_services_status_async(self):
            return None

    screen = _DummyScreen()
    screen.run_init_scripts()

    mon = screen.prole_cfg_data["Monitoring"]
    assert mon["EXISTING_MONITORING_KEY"] == "keep-me"
    assert mon["GRAFANA_ADMIN_PASSWORD"] == "abc123"