mirror of
https://github.com/dredx/prole.git
synced 2026-09-24 18:44:33 +00:00
caa2685fc2
- Rewrite stale foreign home prefixes when loading env defaults (macOS <-> Linux)
- Write /Users/chrisfu-relative values to env.sh; sanitize prole.cfg paths to use ${HOME}
- Remove remaining /Users/chrisfu defaults from helper scripts and OpenTofu kubeconfig path
- Make installer tests more reliable by forcing repo root to front of sys.path
Co-authored-by: Junie <junie@jetbrains.com>
495 lines
17 KiB
Python
495 lines
17 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"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# env.sh portability (avoid hardcoded /Users/<user> paths)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestEnvPortability:
|
||
def test_env_defaults_rewrites_foreign_macos_home_paths_on_linux(
|
||
self, tmp_path, monkeypatch
|
||
):
|
||
# Simulate a shared/synced HOME that contains an env.sh written on macOS.
|
||
fake_home = tmp_path / "testuser"
|
||
(fake_home / ".prole").mkdir(parents=True)
|
||
(fake_home / ".prole" / "env.sh").write_text(
|
||
'\n'.join(
|
||
[
|
||
'export PROLE_HOME="/Users/testuser/dev/prole"',
|
||
'export PROLE_CONF="/Users/testuser/dev/prole/conf"',
|
||
'export PROLE_DATA="/Users/testuser/.prole/data"',
|
||
"",
|
||
]
|
||
)
|
||
)
|
||
|
||
inst = _TestableInstaller(project_root=tmp_path)
|
||
|
||
monkeypatch.setattr("installer.core.actions.platform.system", lambda: "Linux")
|
||
monkeypatch.setattr("installer.core.actions.Path.home", lambda: fake_home)
|
||
|
||
with mock.patch.dict(os.environ, {}, clear=False):
|
||
os.environ.pop("PROLE_HOME", None)
|
||
os.environ.pop("PROLE_CONF", None)
|
||
os.environ.pop("PROLE_DATA", None)
|
||
env = inst._env_defaults(namespace="prole-db")
|
||
|
||
assert env["PROLE_HOME"] == str(fake_home / "dev" / "prole")
|
||
assert env["PROLE_CONF"] == str(fake_home / "dev" / "prole" / "conf")
|
||
assert env["PROLE_DATA"] == str(fake_home / ".prole" / "data")
|
||
|
||
def test_save_env_to_file_uses_home_var_not_absolute_home(self, tmp_path, monkeypatch):
|
||
fake_home = tmp_path / "testuser"
|
||
prole_home = fake_home / "dev" / "prole"
|
||
(prole_home / "conf").mkdir(parents=True)
|
||
(prole_home / "etc").mkdir(parents=True)
|
||
|
||
inst = _TestableInstaller(project_root=tmp_path)
|
||
monkeypatch.setattr("installer.core.actions.Path.home", lambda: fake_home)
|
||
|
||
values = {
|
||
"PROLE_HOME": str(prole_home),
|
||
"PROLE_CONF": str(prole_home / "conf"),
|
||
"PROLE_DATA": str(fake_home / ".prole" / "data"),
|
||
"PROLE_LOGS": "/opt/prole/logs/chrisfu",
|
||
"PROLE_SERVICE": str(prole_home / "etc"),
|
||
}
|
||
|
||
inst._save_env_to_file(values)
|
||
|
||
env_sh = (prole_home / "env.sh").read_text()
|
||
assert 'export PROLE_HOME="$HOME/dev/prole"' in env_sh
|
||
assert str(fake_home) not in env_sh
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _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
|