mirror of
https://github.com/dredx/prole.git
synced 2026-09-24 19:04:31 +00:00
133b719479
Prefer infra-managed manifests for bootstrap; reconcile CNPG instances based on Ready+schedulable labeled db nodes; update manifests to use node-role affinity + anti-affinity; add policy/tests and config touch-ups (incl. prole.cfg).
275 lines
7.8 KiB
Python
275 lines
7.8 KiB
Python
from __future__ import annotations
|
|
|
|
|
|
from unittest.mock import MagicMock
|
|
|
|
|
|
def test_tk_init_scripts_runner_invokes_init_certmgr(tmp_path, monkeypatch):
|
|
import installer.ui.screens.services as services
|
|
from installer.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)
|
|
|
|
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
|
|
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 installer.ui.screens.services as services
|
|
from installer.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)
|
|
|
|
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"
|