mirror of
https://github.com/dredx/prole.git
synced 2026-09-23 11:03:59 +00:00
d397b3000e
- Add explicit Apply button for kubectl context switching to avoid half-applied changes - Allow kubeconfig/context-based auth without requiring K3S_TOKEN when kubeconfig is valid - Prompt cleanup/reset of previous Common Core services namespace to prevent resource collisions - Remove hardcoded 'common-services' registry namespace; default registry deploy/check to SERVICE_NAMESPACE/REGISTRY_NAMESPACE - Update mocks and add regression tests for namespace resolution and installer behavior
433 lines
15 KiB
Python
433 lines
15 KiB
Python
"""Tests for installer/core/actions.py – ProleInstallerBase helper methods."""
|
||
|
||
from __future__ import annotations
|
||
import os
|
||
import sys
|
||
from pathlib import Path
|
||
from unittest import mock
|
||
from unittest.mock import MagicMock, patch
|
||
|
||
import pytest
|
||
|
||
from installer.core.actions import (
|
||
ProleInstallerBase,
|
||
ProleSilentInstaller,
|
||
_configure_unbuffered_io,
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Concrete subclass for testing the abstract base
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class _TestableInstaller(ProleInstallerBase):
|
||
"""Minimal concrete subclass for testing base-class helpers."""
|
||
|
||
def __init__(self, inputs=None, project_root=None):
|
||
self._inputs = inputs or {}
|
||
self.project_root = project_root or Path("/tmp/fake-prole")
|
||
self.controller = MagicMock()
|
||
self._init_shared_state()
|
||
|
||
def _get_input(self, key: str, default: str | None = None) -> str:
|
||
return self._inputs.get(key, default or "")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _configure_unbuffered_io
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestConfigureUnbufferedIO:
|
||
def test_sets_env_var(self):
|
||
with mock.patch.dict(os.environ, {}, clear=False):
|
||
os.environ.pop("PYTHONUNBUFFERED", None)
|
||
_configure_unbuffered_io()
|
||
assert os.environ.get("PYTHONUNBUFFERED") == "1"
|
||
|
||
def test_does_not_overwrite_existing(self):
|
||
with mock.patch.dict(os.environ, {"PYTHONUNBUFFERED": "already"}, clear=False):
|
||
_configure_unbuffered_io()
|
||
assert os.environ["PYTHONUNBUFFERED"] == "already"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _init_shared_state
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestInitSharedState:
|
||
def test_initial_sections(self):
|
||
inst = _TestableInstaller()
|
||
expected_sections = [
|
||
"Global",
|
||
"Welcome",
|
||
"Dependencies",
|
||
"Network",
|
||
"Port Forwards",
|
||
"System Environment",
|
||
"Kerberos Authentication",
|
||
"Ollama",
|
||
"Optional Features",
|
||
"Database Creation",
|
||
"Docker Build",
|
||
"Initialize Cluster",
|
||
"Initialization Scripts",
|
||
"Deployment",
|
||
"Dev Cluster (k3d)",
|
||
"Service Cluster (k3s)",
|
||
"Prod Cluster (k8s)",
|
||
"Install",
|
||
]
|
||
for section in expected_sections:
|
||
assert section in inst.prole_cfg_data
|
||
|
||
def test_initial_flags(self):
|
||
inst = _TestableInstaller()
|
||
assert inst._cfg_secret_cache == {}
|
||
assert inst._secrets_finalized is False
|
||
assert inst._managed_kubeconfig is None
|
||
assert inst._repair_ran is False
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _get_input_bool
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestGetInputBool:
|
||
def test_true_values(self):
|
||
for val in ("true", "1", "yes", "on", "True", "YES"):
|
||
inst = _TestableInstaller(inputs={"key": val})
|
||
assert inst._get_input_bool("key") is True
|
||
|
||
def test_false_values(self):
|
||
for val in ("false", "0", "no", "off", "False", "NO"):
|
||
inst = _TestableInstaller(inputs={"key": val})
|
||
assert inst._get_input_bool("key") is False
|
||
|
||
def test_default(self):
|
||
inst = _TestableInstaller()
|
||
assert inst._get_input_bool("missing_key", default=True) is True
|
||
assert inst._get_input_bool("missing_key", default=False) is False
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Logging
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestLogging:
|
||
def test_log_prints(self, capsys):
|
||
inst = _TestableInstaller()
|
||
inst.log("hello world")
|
||
captured = capsys.readouterr()
|
||
assert "hello world" in captured.out
|
||
|
||
def test_err_prints_to_stderr(self, capsys):
|
||
inst = _TestableInstaller()
|
||
inst.err("error msg")
|
||
captured = capsys.readouterr()
|
||
assert "error msg" in captured.err
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _deployment_mode
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestDeploymentMode:
|
||
def test_dev_mode(self):
|
||
inst = _TestableInstaller(inputs={"init_cluster.cluster_env": "dev"})
|
||
assert inst._deployment_mode() == "k3d"
|
||
|
||
def test_service_mode(self):
|
||
inst = _TestableInstaller(inputs={"init_cluster.cluster_env": "service"})
|
||
assert inst._deployment_mode() == "k3s"
|
||
|
||
def test_prod_mode(self):
|
||
inst = _TestableInstaller(inputs={"init_cluster.cluster_env": "prod"})
|
||
assert inst._deployment_mode() == "k8s"
|
||
|
||
def test_empty_mode(self):
|
||
inst = _TestableInstaller(inputs={"init_cluster.cluster_env": ""})
|
||
result = inst._deployment_mode()
|
||
assert result == "" or result is not None
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _secret_namespace
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestSecretNamespace:
|
||
def test_from_db_namespace(self):
|
||
inst = _TestableInstaller(inputs={"init_password.db_namespace": "my-ns"})
|
||
assert inst._secret_namespace() == "my-ns"
|
||
|
||
def test_from_env_namespace(self):
|
||
inst = _TestableInstaller(inputs={"env_setup.NAMESPACE": "env-ns"})
|
||
assert inst._secret_namespace() == "env-ns"
|
||
|
||
def test_from_cfg_data(self):
|
||
inst = _TestableInstaller()
|
||
inst.prole_cfg_data["Global"]["NAMESPACE"] = "cfg-ns"
|
||
assert inst._secret_namespace() == "cfg-ns"
|
||
|
||
def test_default(self):
|
||
inst = _TestableInstaller()
|
||
assert inst._secret_namespace() == "default"
|
||
|
||
def test_priority_db_over_env(self):
|
||
inst = _TestableInstaller(
|
||
inputs={
|
||
"init_password.db_namespace": "db-ns",
|
||
"env_setup.NAMESPACE": "env-ns",
|
||
}
|
||
)
|
||
assert inst._secret_namespace() == "db-ns"
|
||
|
||
def test_strips_whitespace(self):
|
||
inst = _TestableInstaller(inputs={"init_password.db_namespace": " my-ns "})
|
||
assert inst._secret_namespace() == "my-ns"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _service_namespace
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestServiceNamespace:
|
||
def test_from_cfg_data(self):
|
||
inst = _TestableInstaller()
|
||
inst.prole_cfg_data["Global"]["SERVICE_NAMESPACE"] = "svc-ns"
|
||
assert inst._service_namespace() == "svc-ns"
|
||
|
||
def test_from_env_var(self):
|
||
inst = _TestableInstaller()
|
||
with mock.patch.dict(os.environ, {"SERVICE_NAMESPACE": "env-svc-ns"}):
|
||
assert inst._service_namespace() == "env-svc-ns"
|
||
|
||
def test_default(self):
|
||
inst = _TestableInstaller()
|
||
with mock.patch.dict(os.environ, {}, clear=False):
|
||
os.environ.pop("SERVICE_NAMESPACE", None)
|
||
assert inst._service_namespace() == "default"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _collect_dependent_images
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestCollectDependentImages:
|
||
def test_filters_supabase_when_disabled(self, tmp_path):
|
||
base = tmp_path / "k8s" / "prole"
|
||
base.mkdir(parents=True)
|
||
f = base / "test.yaml"
|
||
f.write_text(
|
||
"""
|
||
apiVersion: v1
|
||
kind: Pod
|
||
spec:
|
||
containers:
|
||
- name: a
|
||
image: nginx:1.2
|
||
- name: b
|
||
image: supabase/postgres:15
|
||
""".lstrip()
|
||
)
|
||
|
||
inst = _TestableInstaller(project_root=tmp_path)
|
||
|
||
images = inst._collect_dependent_images(
|
||
include_supabase=False,
|
||
include_kerberos_proxy=False,
|
||
)
|
||
assert "nginx:1.2" in images
|
||
assert all("supabase" not in img for img in images)
|
||
|
||
def test_includes_supabase_and_kerberos_when_enabled(self, tmp_path):
|
||
base = tmp_path / "k8s" / "prole"
|
||
base.mkdir(parents=True)
|
||
f = base / "test.yaml"
|
||
f.write_text(
|
||
"""
|
||
apiVersion: v1
|
||
kind: Pod
|
||
spec:
|
||
containers:
|
||
- name: a
|
||
image: nginx:1.2
|
||
- name: b
|
||
image: supabase/postgres:15
|
||
""".lstrip()
|
||
)
|
||
|
||
inst = _TestableInstaller(project_root=tmp_path)
|
||
|
||
images = inst._collect_dependent_images(
|
||
include_supabase=True,
|
||
include_kerberos_proxy=True,
|
||
)
|
||
assert "nginx:1.2" in images
|
||
assert "supabase/postgres:15" in images
|
||
assert "ghcr.io/bsharp-tech/prole-kerberos-proxy:latest" in images
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _argocd_namespace / _registry_namespace
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class _TestableSilentInstaller(ProleSilentInstaller):
|
||
"""Testable subclass of ProleSilentInstaller."""
|
||
|
||
def __init__(self, inputs=None, project_root=None):
|
||
self._inputs_dict = inputs or {}
|
||
self.project_root = project_root or Path("/tmp/fake-prole")
|
||
self.controller = MagicMock()
|
||
self.cfg_path = None
|
||
self.silent = True
|
||
self._init_shared_state()
|
||
|
||
def _get_input(self, key: str, default: str | None = None) -> str:
|
||
return self._inputs_dict.get(key, default or "")
|
||
|
||
|
||
class TestNamespaceHelpers:
|
||
def test_argocd_from_cfg(self):
|
||
inst = _TestableSilentInstaller()
|
||
inst.prole_cfg_data["Global"]["ARGOCD_NAMESPACE"] = "argo-ns"
|
||
assert inst._argocd_namespace() == "argo-ns"
|
||
|
||
def test_argocd_from_env(self):
|
||
inst = _TestableSilentInstaller()
|
||
with mock.patch.dict(os.environ, {"ARGOCD_NAMESPACE": "env-argo"}):
|
||
assert inst._argocd_namespace() == "env-argo"
|
||
|
||
def test_argocd_default(self):
|
||
inst = _TestableSilentInstaller()
|
||
with mock.patch.dict(os.environ, {}, clear=False):
|
||
os.environ.pop("ARGOCD_NAMESPACE", None)
|
||
assert inst._argocd_namespace() == "argocd"
|
||
|
||
def test_registry_from_cfg(self):
|
||
inst = _TestableSilentInstaller()
|
||
inst.prole_cfg_data["Global"]["REGISTRY_NAMESPACE"] = "reg-ns"
|
||
assert inst._registry_namespace() == "reg-ns"
|
||
|
||
def test_registry_from_env(self):
|
||
inst = _TestableSilentInstaller()
|
||
with mock.patch.dict(os.environ, {"REGISTRY_NAMESPACE": "env-reg"}):
|
||
assert inst._registry_namespace() == "env-reg"
|
||
|
||
def test_registry_default(self):
|
||
# Registry is a common-core service; default should follow SERVICE_NAMESPACE.
|
||
# For k3s/service clusters, SERVICE_NAMESPACE defaults to knoe-system.
|
||
inst = _TestableSilentInstaller(inputs={"init_cluster.cluster_env": "service"})
|
||
with mock.patch.dict(os.environ, {}, clear=False):
|
||
os.environ.pop("REGISTRY_NAMESPACE", None)
|
||
assert inst._registry_namespace() == "knoe-system"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _read_k3s_cfg_values
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestReadK3sCfgValues:
|
||
def test_no_cfg_path(self):
|
||
inst = _TestableInstaller()
|
||
env, url, token = inst._read_k3s_cfg_values(None)
|
||
# With no cfg, should return empty/defaults
|
||
assert isinstance(env, str)
|
||
assert isinstance(url, str)
|
||
assert isinstance(token, str)
|
||
|
||
def test_with_cfg_path(self, tmp_path):
|
||
import configparser
|
||
|
||
cfg = configparser.ConfigParser(interpolation=None)
|
||
cfg.optionxform = str
|
||
cfg.add_section("Global")
|
||
cfg.set("Global", "CLUSTER_ENV", "service")
|
||
cfg.add_section("Initialize Cluster")
|
||
cfg.set("Initialize Cluster", "K3S_SERVER_URL", "https://10.0.0.1:6443")
|
||
cfg.set("Initialize Cluster", "K3S_TOKEN", "mytoken")
|
||
cfg_path = tmp_path / "prole.cfg"
|
||
with open(cfg_path, "w") as f:
|
||
cfg.write(f)
|
||
inst = _TestableInstaller()
|
||
env, url, token = inst._read_k3s_cfg_values(cfg_path)
|
||
assert env == "service"
|
||
assert url == "https://10.0.0.1:6443"
|
||
assert token == "mytoken"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _run_script
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestRunScript:
|
||
def test_run_script_delegates(self):
|
||
inst = _TestableSilentInstaller(inputs={"init_cluster.cluster_env": "dev"})
|
||
inst.controller.run_script.return_value = 0
|
||
rc = inst._run_script("test_script.sh", args=["arg1"])
|
||
assert rc == 0
|
||
inst.controller.run_script.assert_called_once()
|
||
|
||
def test_run_script_adds_mode(self):
|
||
inst = _TestableSilentInstaller(inputs={"init_cluster.cluster_env": "dev"})
|
||
inst.controller.run_script.return_value = 0
|
||
inst._run_script("init_common_services.sh", args=["update"])
|
||
call_kwargs = inst.controller.run_script.call_args
|
||
args_passed = call_kwargs[1].get(
|
||
"args", call_kwargs[0][1] if len(call_kwargs[0]) > 1 else []
|
||
)
|
||
assert "--mode" in args_passed
|
||
assert "k3d" in args_passed
|
||
|
||
def test_run_script_no_mode_for_non_init(self):
|
||
inst = _TestableSilentInstaller(inputs={"init_cluster.cluster_env": "dev"})
|
||
inst.controller.run_script.return_value = 0
|
||
inst._run_script("build-a-bao.sh", args=[])
|
||
call_kwargs = inst.controller.run_script.call_args
|
||
args_passed = call_kwargs[1].get(
|
||
"args", call_kwargs[0][1] if len(call_kwargs[0]) > 1 else []
|
||
)
|
||
assert "--mode" not in args_passed
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _step_init_scripts
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestStepInitScripts:
|
||
def test_includes_init_certmgr(self, monkeypatch):
|
||
inst = _TestableSilentInstaller(
|
||
inputs={
|
||
"init_cluster.cluster_env": "service",
|
||
"init_password.db_namespace": "default",
|
||
"init_password.db_password": "pw",
|
||
"kerberos_config.enabled": "false",
|
||
"init_scripts.run_scripts": "true",
|
||
}
|
||
)
|
||
|
||
inst.controller.run_script.return_value = 0
|
||
monkeypatch.setattr(inst, "_deployment_mode", lambda: "k3s")
|
||
monkeypatch.setattr(inst, "_run_cmd", lambda *a, **k: 0, raising=False)
|
||
|
||
inst._step_init_scripts()
|
||
|
||
scripts = [c.args[0] for c in inst.controller.run_script.call_args_list]
|
||
assert "init_common_services.sh" in scripts
|
||
assert "init_certmgr.sh" in scripts
|
||
assert scripts.index("init_certmgr.sh") == scripts.index("init_common_services.sh") + 1
|
||
assert inst.prole_cfg_data["Initialization Scripts"]["STATUS"] == "Completed"
|
||
assert inst._scripts_success is True
|