mirror of
https://github.com/dredx/prole.git
synced 2026-09-23 11:03:59 +00:00
1062 lines
32 KiB
Python
1062 lines
32 KiB
Python
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)
|
|
monkeypatch.setattr(services.registry_ops, "update", lambda **_kw: None)
|
|
monkeypatch.setattr(services.openbao_ops, "update", lambda **_kw: None)
|
|
monkeypatch.setattr(services.garage_store_ops, "update", lambda **_kw: None)
|
|
monkeypatch.setattr(services.opentofu_ops, "update", lambda **_kw: None)
|
|
monkeypatch.setattr(services.monitoring_ops, "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("knoe")
|
|
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("knoe-service-cluster")
|
|
|
|
self.knoe_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 _cluster_kubecontext(self, cluster_role: str = "app") -> str:
|
|
if cluster_role == "db":
|
|
return "gke_db_ctx"
|
|
return "gke_app_ctx"
|
|
|
|
def _resolve_knoe_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_knoe_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_cnpg_backup.sh" in called_scripts
|
|
assert "init_common_services.sh" not in called_scripts
|
|
# CNPG is now Python-owned; shell script must never be dispatched
|
|
assert "init_cloudnative_pg.sh" not in called_scripts
|
|
assert "init_db_manager.sh" not in called_scripts
|
|
assert screen.knoe_cfg_data["Initialization Scripts"]["STATUS"] == "Completed"
|
|
|
|
|
|
def test_tk_init_scripts_common_services_prefers_service_namespace(
|
|
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)
|
|
monkeypatch.setattr(services, "cnpg_initialize", lambda **_kw: None)
|
|
monkeypatch.setattr(services.registry_ops, "update", lambda **_kw: None)
|
|
monkeypatch.setattr(services.garage_store_ops, "update", lambda **_kw: None)
|
|
monkeypatch.setattr(services.opentofu_ops, "update", lambda **_kw: None)
|
|
monkeypatch.setattr(services.monitoring_ops, "initialize", lambda **_kw: None)
|
|
|
|
openbao_calls: list[dict] = []
|
|
|
|
def _openbao_update(**kwargs):
|
|
openbao_calls.append(kwargs)
|
|
|
|
monkeypatch.setattr(services.openbao_ops, "update", _openbao_update)
|
|
|
|
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("knoe")
|
|
self.db_password = _Var("pw")
|
|
self.db_namespace = _Var("knoe-db")
|
|
|
|
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("knoe-service-cluster")
|
|
|
|
self.knoe_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 _cluster_kubecontext(self, cluster_role: str = "app") -> str:
|
|
if cluster_role == "db":
|
|
return "gke_db_ctx"
|
|
return "gke_app_ctx"
|
|
|
|
def _resolve_knoe_logs_dir(self):
|
|
return tmp_path
|
|
|
|
def _get_service_namespace(self):
|
|
return "knoe-system"
|
|
|
|
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_knoe_cfg(self):
|
|
return None
|
|
|
|
def update_footer(self):
|
|
return None
|
|
|
|
def check_services_status_async(self):
|
|
return None
|
|
|
|
screen = _DummyScreen()
|
|
screen.run_init_scripts()
|
|
|
|
assert openbao_calls
|
|
assert openbao_calls[0]["namespace"] == "knoe-system"
|
|
|
|
|
|
def test_tk_init_scripts_uses_fallback_database_namespace_for_secret_checks(
|
|
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)
|
|
monkeypatch.setattr(services, "cnpg_initialize", lambda **_kw: None)
|
|
monkeypatch.setattr(services.registry_ops, "update", lambda **_kw: None)
|
|
monkeypatch.setattr(services.openbao_ops, "update", lambda **_kw: None)
|
|
monkeypatch.setattr(services.garage_store_ops, "update", lambda **_kw: None)
|
|
monkeypatch.setattr(services.opentofu_ops, "update", lambda **_kw: None)
|
|
monkeypatch.setattr(services.monitoring_ops, "initialize", lambda **_kw: None)
|
|
|
|
secret_check_calls: list[list[str]] = []
|
|
|
|
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("knoe")
|
|
self.db_password = _Var("pw")
|
|
# Reproducer: UI field is blank
|
|
self.db_namespace = _Var("")
|
|
|
|
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("knoe-service-cluster")
|
|
|
|
self.knoe_cfg_data = {
|
|
"Initialization Scripts": {"STATUS": ""},
|
|
"Global": {"DATABASE_NAMESPACE": "knoe-db"},
|
|
}
|
|
self._scripts_success = False
|
|
|
|
def safe_after(self, fn, delay: int = 0):
|
|
return fn()
|
|
|
|
def _deployment_mode(self):
|
|
return "k3s"
|
|
|
|
def _resolve_knoe_logs_dir(self):
|
|
return tmp_path
|
|
|
|
def _get_service_namespace(self):
|
|
return "knoe-system"
|
|
|
|
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):
|
|
secret_check_calls.append(args)
|
|
try:
|
|
get_idx = args.index("get")
|
|
except ValueError:
|
|
return 0, ""
|
|
|
|
if get_idx + 2 >= len(args) or args[get_idx + 1] != "secret":
|
|
return 0, ""
|
|
|
|
if args[get_idx + 2] in {
|
|
"knoe-db-user",
|
|
"knoe-db-superuser",
|
|
"cnpg-admin-key",
|
|
}:
|
|
return (0, "") if args[-1] == "knoe-db" else (1, "")
|
|
return 0, ""
|
|
|
|
def _save_knoe_cfg(self):
|
|
return None
|
|
|
|
def update_footer(self):
|
|
return None
|
|
|
|
def check_services_status_async(self):
|
|
return None
|
|
|
|
screen = _DummyScreen()
|
|
screen.run_init_scripts()
|
|
|
|
assert screen.knoe_cfg_data["Initialization Scripts"]["STATUS"] == "Completed"
|
|
checked_ns = {
|
|
args[-1]
|
|
for args in secret_check_calls
|
|
if "-n" in args and "get" in args and "secret" in args
|
|
}
|
|
assert "knoe-db" in checked_ns
|
|
|
|
|
|
def test_tk_init_scripts_secret_checks_use_kubectl_base_cmd(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)
|
|
monkeypatch.setattr(services, "cnpg_initialize", lambda **_kw: None)
|
|
monkeypatch.setattr(services.registry_ops, "update", lambda **_kw: None)
|
|
monkeypatch.setattr(services.openbao_ops, "update", lambda **_kw: None)
|
|
monkeypatch.setattr(services.garage_store_ops, "update", lambda **_kw: None)
|
|
monkeypatch.setattr(services.opentofu_ops, "update", lambda **_kw: None)
|
|
monkeypatch.setattr(services.monitoring_ops, "initialize", lambda **_kw: None)
|
|
|
|
expected_prefix = ["kubectl", "--kubeconfig", "/tmp/knoe-k3s.kubeconfig"]
|
|
secret_check_calls: list[list[str]] = []
|
|
|
|
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("knoe")
|
|
self.db_password = _Var("pw")
|
|
self.db_namespace = _Var("knoe-db")
|
|
|
|
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("knoe-service-cluster")
|
|
|
|
self.knoe_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_knoe_logs_dir(self):
|
|
return tmp_path
|
|
|
|
def _get_service_namespace(self):
|
|
return "knoe-system"
|
|
|
|
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 _kubectl_base_cmd(self, _mode=None):
|
|
return list(expected_prefix)
|
|
|
|
def _run_cmd_capture(self, args):
|
|
secret_check_calls.append(args)
|
|
try:
|
|
get_idx = args.index("get")
|
|
except ValueError:
|
|
return 0, ""
|
|
|
|
if get_idx + 1 >= len(args) or args[get_idx + 1] != "secret":
|
|
return 0, ""
|
|
|
|
if args[: len(expected_prefix)] != expected_prefix:
|
|
return 1, ""
|
|
return 0, ""
|
|
|
|
def _save_knoe_cfg(self):
|
|
return None
|
|
|
|
def update_footer(self):
|
|
return None
|
|
|
|
def check_services_status_async(self):
|
|
return None
|
|
|
|
screen = _DummyScreen()
|
|
screen.run_init_scripts()
|
|
|
|
assert screen.knoe_cfg_data["Initialization Scripts"]["STATUS"] == "Completed"
|
|
secret_calls = [
|
|
args
|
|
for args in secret_check_calls
|
|
if "get" in args and "secret" in args and "-n" in args
|
|
]
|
|
assert secret_calls
|
|
for args in secret_calls:
|
|
assert args[: len(expected_prefix)] == expected_prefix
|
|
|
|
|
|
def test_tk_init_scripts_missing_critical_secret_check_does_not_block_next(
|
|
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)
|
|
monkeypatch.setattr(services, "cnpg_initialize", lambda **_kw: None)
|
|
monkeypatch.setattr(services.registry_ops, "update", lambda **_kw: None)
|
|
monkeypatch.setattr(services.openbao_ops, "update", lambda **_kw: None)
|
|
monkeypatch.setattr(services.garage_store_ops, "update", lambda **_kw: None)
|
|
monkeypatch.setattr(services.opentofu_ops, "update", lambda **_kw: None)
|
|
monkeypatch.setattr(services.monitoring_ops, "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("knoe")
|
|
self.db_password = _Var("pw")
|
|
self.db_namespace = _Var("knoe-db")
|
|
|
|
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("knoe-service-cluster")
|
|
|
|
self.knoe_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_knoe_logs_dir(self):
|
|
return tmp_path
|
|
|
|
def _get_service_namespace(self):
|
|
return "knoe-system"
|
|
|
|
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):
|
|
if "get" not in args or "secret" not in args:
|
|
return 0, ""
|
|
if "knoe-db-superuser" in args:
|
|
return 1, "Error from server (NotFound): secrets \"knoe-db-superuser\" not found"
|
|
return 0, ""
|
|
|
|
def _save_knoe_cfg(self):
|
|
return None
|
|
|
|
def update_footer(self):
|
|
return None
|
|
|
|
def check_services_status_async(self):
|
|
return None
|
|
|
|
screen = _DummyScreen()
|
|
screen.run_init_scripts()
|
|
|
|
assert screen._scripts_success is True
|
|
assert screen.knoe_cfg_data["Initialization Scripts"]["STATUS"] == "Completed"
|
|
assert any(
|
|
"Missing secrets" in entry for entry in screen.script_consoles["init_cloudnative_pg.sh"].buf
|
|
)
|
|
|
|
|
|
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)
|
|
monkeypatch.setattr(services.registry_ops, "update", lambda **_kw: None)
|
|
monkeypatch.setattr(services.openbao_ops, "update", lambda **_kw: None)
|
|
monkeypatch.setattr(services.garage_store_ops, "update", lambda **_kw: None)
|
|
monkeypatch.setattr(services.opentofu_ops, "update", lambda **_kw: None)
|
|
|
|
def _monitoring_initialize(**kwargs):
|
|
log = kwargs.get("log")
|
|
if callable(log):
|
|
log("GRAFANA_ADMIN_PASSWORD=abc123")
|
|
|
|
monkeypatch.setattr(services.monitoring_ops, "initialize", _monitoring_initialize)
|
|
|
|
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("knoe")
|
|
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("knoe-service-cluster")
|
|
|
|
self.knoe_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_knoe_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_knoe_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.knoe_cfg_data["Monitoring"]
|
|
assert mon["EXISTING_MONITORING_KEY"] == "keep-me"
|
|
assert mon["GRAFANA_ADMIN_PASSWORD"] == "abc123"
|
|
|
|
|
|
def test_tk_init_scripts_k8s_sets_artifact_registry_env_for_cnpg(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)
|
|
monkeypatch.setattr(services.registry_ops, "update", lambda **_kw: None)
|
|
monkeypatch.setattr(services.openbao_ops, "update", lambda **_kw: None)
|
|
monkeypatch.setattr(services.garage_store_ops, "update", lambda **_kw: None)
|
|
monkeypatch.setattr(services.opentofu_ops, "update", lambda **_kw: None)
|
|
monkeypatch.setattr(services.monitoring_ops, "initialize", lambda **_kw: None)
|
|
|
|
captured_cnpg_env: dict[str, str] = {}
|
|
|
|
def _fake_cnpg_initialize(**kwargs):
|
|
captured_cnpg_env.update(kwargs.get("env") or {})
|
|
|
|
monkeypatch.setattr(services, "cnpg_initialize", _fake_cnpg_initialize)
|
|
|
|
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("knoe")
|
|
self.db_password = _Var("pw")
|
|
self.db_namespace = _Var("knoe-db-0")
|
|
|
|
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("knoe-prod-cluster")
|
|
|
|
self.knoe_cfg_data = {
|
|
"Initialization Scripts": {"STATUS": ""},
|
|
"GCP": {"project_id": "plenary-truck-485623-p7"},
|
|
"Global": {
|
|
"KUBECONTEXT": "gke_plenary-truck-485623-p7_us-west3_knoe-dev-0",
|
|
"SERVICE_NAMESPACE": "knoe",
|
|
},
|
|
}
|
|
self._scripts_success = False
|
|
|
|
def safe_after(self, fn, delay: int = 0):
|
|
return fn()
|
|
|
|
def _deployment_mode(self):
|
|
return "k8s"
|
|
|
|
def _cluster_kubecontext(self, cluster_role: str = "app") -> str:
|
|
if cluster_role == "db":
|
|
return "gke_plenary-truck-485623-p7_us-west3_knoe-cnpg-0"
|
|
return "gke_plenary-truck-485623-p7_us-west3_knoe-dev-0"
|
|
|
|
def _resolve_knoe_logs_dir(self):
|
|
return tmp_path
|
|
|
|
def _get_service_namespace(self):
|
|
return "knoe-system"
|
|
|
|
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_knoe_cfg(self):
|
|
return None
|
|
|
|
def update_footer(self):
|
|
return None
|
|
|
|
def check_services_status_async(self):
|
|
return None
|
|
|
|
def _load_gcp_cfg_into_knoe_data(self):
|
|
return None
|
|
|
|
screen = _DummyScreen()
|
|
screen.run_init_scripts()
|
|
|
|
assert captured_cnpg_env["GCP_PROJECT_ID"] == "plenary-truck-485623-p7"
|
|
assert captured_cnpg_env["GCP_REGION"] == "us-west3"
|
|
assert (
|
|
captured_cnpg_env["ARTIFACT_REGISTRY"]
|
|
== "us-west3-docker.pkg.dev/plenary-truck-485623-p7/knoe"
|
|
)
|