"""Tests for installer/core/actions.py – ProleInstallerBase helper methods.""" from __future__ import annotations import os import subprocess 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" # --------------------------------------------------------------------------- # k3s registry defaults (regression) # --------------------------------------------------------------------------- class TestK3sRegistryDefaults: def test_ensure_registry_defaults_k3s_overrides_localhost(self): inst = _TestableSilentInstaller(inputs={"init_cluster.cluster_env": "service"}) inst.prole_cfg_data.setdefault("Docker Build", {}) inst.prole_cfg_data["Docker Build"]["LOCAL_REGISTRY"] = "localhost:5000" inst.prole_cfg_data["Docker Build"].pop("LOCAL_REGISTRY_INTERNAL", None) with mock.patch.dict( os.environ, {"PROLE_K3S_SERVER": "https://k3s.example.test:6443"}, clear=False, ): inst._ensure_registry_defaults("service") assert inst.prole_cfg_data["Docker Build"]["LOCAL_REGISTRY"] == "k3s.example.test:5000" assert ( inst.prole_cfg_data["Docker Build"]["LOCAL_REGISTRY_INTERNAL"] == "registry.knoe-system.svc.cluster.local:5000" ) def test_db_build_k3s_push_failure_marks_failed(self): inst = _TestableSilentInstaller(inputs={"init_cluster.cluster_env": "service"}) inst.prole_cfg_data.setdefault("Docker Build", {}) inst.prole_cfg_data["Docker Build"].pop("LOCAL_REGISTRY", None) inst.prole_cfg_data["Docker Build"].pop("LOCAL_REGISTRY_INTERNAL", None) inst.controller.get_prole_db_version.return_value = "latest" def fake_run(cmd, *args, **kwargs): # Pretend the image already exists so we exercise the push path. if cmd[:2] == ["docker", "inspect"]: return subprocess.CompletedProcess(cmd, 0, stdout="", stderr="") return subprocess.CompletedProcess(cmd, 0, stdout="", stderr="") with ( mock.patch.dict( os.environ, {"PROLE_K3S_SERVER": "https://k3s.example.test:6443"}, clear=False, ), mock.patch.object(inst, "_get_input_bool", return_value=True), mock.patch("installer.core.actions.subprocess.run", side_effect=fake_run), mock.patch("installer.core.actions._registry_image_ref_exists", return_value=False), mock.patch("installer.core.actions._push_docker_image", return_value=False), ): inst._step_db_build() assert inst._db_built_success is False assert inst.prole_cfg_data["Docker Build"]["STATUS"] == "Failed" # --------------------------------------------------------------------------- # env.sh portability (avoid hardcoded /Users/ 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" # --------------------------------------------------------------------------- # _script_env_for_namespace # --------------------------------------------------------------------------- class TestScriptEnvForNamespace: def test_exports_db_password_for_shell_scripts(self): inst = _TestableInstaller( inputs={ "init_password.db_username": "postgres", "init_password.db_password": "root-master-password", } ) env = inst._script_env_for_namespace("knoe-system") assert env.get("DB_PASSWORD") == "root-master-password" # OpenTofu bootstraps admin password from DB_PASSWORD when unset. assert env.get("OPENTOFU_ADMIN_PASSWORD") == "root-master-password" # --------------------------------------------------------------------------- # ensure_db_k8s_secrets # --------------------------------------------------------------------------- class TestEnsureDbK8sSecrets: def test_applies_db_user_and_superuser_and_creates_admin_key_when_missing( self, tmp_path, monkeypatch ): # Fake HOME with an ssh keypair already present fake_home = tmp_path / "home" ssh_dir = fake_home / ".ssh" ssh_dir.mkdir(parents=True) (ssh_dir / "id_prole_ed25519").write_text("PRIVATE") (ssh_dir / "id_prole_ed25519.pub").write_text("PUBLIC") monkeypatch.setattr(Path, "home", classmethod(lambda cls: fake_home)) service_dir = tmp_path / "etc" inst = _TestableInstaller(inputs={"env_setup.PROLE_SERVICE": str(service_dir)}) calls: list[list[str]] = [] logs: list[str] = [] ns = "knoe-db" pw = "user-entered-password" def fake_run(cmd, **kwargs): calls.append(list(cmd)) # Namespace exists if cmd[:3] == ["kubectl", "get", "namespace"]: return subprocess.CompletedProcess(cmd, 0, "", "") # cnpg-admin-key missing if cmd[:5] == ["kubectl", "-n", ns, "get", "secret"] and cmd[5] == "cnpg-admin-key": return subprocess.CompletedProcess(cmd, 1, "", "") # kubectl create secret ... -o yaml should return YAML (do not log) if cmd[:4] == ["kubectl", "create", "secret", "generic"] and "-o" in cmd and "yaml" in cmd: return subprocess.CompletedProcess(cmd, 0, "apiVersion: v1\nkind: Secret\n", "") # kubectl apply reads stdin if cmd[:2] == ["kubectl", "apply"]: return subprocess.CompletedProcess(cmd, 0, "applied\n", "") # kubectl create namespace fallback if cmd[:3] == ["kubectl", "create", "namespace"]: return subprocess.CompletedProcess(cmd, 0, "created\n", "") return subprocess.CompletedProcess(cmd, 0, "", "") monkeypatch.setattr( sys.modules["installer.core.actions"].subprocess, "run", fake_run ) inst.ensure_db_k8s_secrets(ns, pw, log_fn=logs.append) # Ensure secrets dir was created and key files copied secrets_dir = service_dir / "secrets" assert (secrets_dir / "admin.key").exists() assert (secrets_dir / "admin.pub").exists() assert (secrets_dir / "admin_ed25519").exists() assert (secrets_dir / "admin_ed25519.pub").exists() # Ensure DB user/superuser secrets were rendered with expected usernames rendered_cmds = [c for c in calls if c[:4] == ["kubectl", "create", "secret", "generic"]] assert any( c[4] == "prole-db-user" and any(a == "--from-literal=username=prole" for a in c) and any( a == f"--from-literal=password={pw}" for a in c ) for c in rendered_cmds ) assert any( c[4] == "prole-db-superuser" and any(a == "--from-literal=username=postgres" for a in c) for c in rendered_cmds ) assert any(c[4] == "cnpg-admin-key" for c in rendered_cmds) # Ensure we did not leak secret YAML into logs assert not any("apiVersion:" in m for m in logs) def test_does_not_rotate_admin_key_secret_when_present(self, tmp_path, monkeypatch): fake_home = tmp_path / "home" ssh_dir = fake_home / ".ssh" ssh_dir.mkdir(parents=True) (ssh_dir / "id_prole_ed25519").write_text("PRIVATE") (ssh_dir / "id_prole_ed25519.pub").write_text("PUBLIC") monkeypatch.setattr(Path, "home", classmethod(lambda cls: fake_home)) service_dir = tmp_path / "etc" inst = _TestableInstaller(inputs={"env_setup.PROLE_SERVICE": str(service_dir)}) calls: list[list[str]] = [] ns = "knoe-db" def fake_run(cmd, **kwargs): calls.append(list(cmd)) if cmd[:3] == ["kubectl", "get", "namespace"]: return subprocess.CompletedProcess(cmd, 0, "", "") if cmd[:5] == ["kubectl", "-n", ns, "get", "secret"] and cmd[5] == "cnpg-admin-key": return subprocess.CompletedProcess(cmd, 0, "", "") if cmd[:4] == ["kubectl", "create", "secret", "generic"] and "-o" in cmd and "yaml" in cmd: return subprocess.CompletedProcess(cmd, 0, "apiVersion: v1\nkind: Secret\n", "") if cmd[:2] == ["kubectl", "apply"]: return subprocess.CompletedProcess(cmd, 0, "applied\n", "") return subprocess.CompletedProcess(cmd, 0, "", "") monkeypatch.setattr( sys.modules["installer.core.actions"].subprocess, "run", fake_run ) inst.ensure_db_k8s_secrets(ns, "pw") # Ensure we did not attempt to create cnpg-admin-key when it already exists assert not any( c[:5] == ["kubectl", "create", "secret", "generic", "cnpg-admin-key"] for c in calls ) # --------------------------------------------------------------------------- # _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) monkeypatch.setattr(inst, "ensure_db_k8s_secrets", lambda *a, **k: None) 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