"""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/ 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