"""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): inst = _TestableSilentInstaller() with mock.patch.dict(os.environ, {}, clear=False): os.environ.pop("REGISTRY_NAMESPACE", None) assert inst._registry_namespace() == "default" # --------------------------------------------------------------------------- # _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