"""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 from knoe.core.actions import ( KnoeInstaller, KnoeConsoleInstaller, _configure_unbuffered_io, _select_existing_kubeconfig, ) # --------------------------------------------------------------------------- # Concrete subclass for testing the abstract base # --------------------------------------------------------------------------- class _TestableInstaller(KnoeInstaller): """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" # --------------------------------------------------------------------------- # _select_existing_kubeconfig # --------------------------------------------------------------------------- class TestSelectExistingKubeconfig: def test_prefers_first_working_candidate(self, tmp_path, monkeypatch): import knoe.core.actions as actions_mod bad = tmp_path / "bad.kubeconfig" good = tmp_path / "good.kubeconfig" bad.write_text("bad") good.write_text("good") def _fake_run(cmd, **_kwargs): # cmd = ["kubectl", "--kubeconfig", , "cluster-info"] kc_path = cmd[2] if kc_path == str(bad): return subprocess.CompletedProcess(cmd, 1, stdout="", stderr="Unauthorized") if kc_path == str(good): return subprocess.CompletedProcess(cmd, 0, stdout="ok", stderr="") return subprocess.CompletedProcess(cmd, 1, stdout="", stderr="") monkeypatch.setattr(actions_mod.subprocess, "run", _fake_run) assert _select_existing_kubeconfig([bad, good]) == good def test_falls_back_to_first_existing_when_none_work(self, tmp_path, monkeypatch): import knoe.core.actions as actions_mod first = tmp_path / "first.kubeconfig" second = tmp_path / "second.kubeconfig" first.write_text("first") second.write_text("second") def _fake_run(cmd, **_kwargs): return subprocess.CompletedProcess(cmd, 1, stdout="", stderr="") monkeypatch.setattr(actions_mod.subprocess, "run", _fake_run) assert _select_existing_kubeconfig([first, second]) == first # --------------------------------------------------------------------------- # _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() == "knoe-system" # --------------------------------------------------------------------------- # _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) with mock.patch.dict( os.environ, {"KRB5_AD_PROXY_IMAGE": "krb-proxy:test"}, clear=False ): 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 "krb-proxy:test" in images # --------------------------------------------------------------------------- # _argocd_namespace / _registry_namespace # --------------------------------------------------------------------------- class _TestableSilentInstaller(KnoeConsoleInstaller): """Testable subclass of KnoeConsoleInstaller.""" 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_knoe_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("knoe.core.actions.subprocess.run", side_effect=fake_run), mock.patch("knoe.core.actions._registry_image_ref_exists", return_value=False), mock.patch("knoe.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("knoe.core.actions.platform.system", lambda: "Linux") monkeypatch.setattr("knoe.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="knoe-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("knoe.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 # --------------------------------------------------------------------------- # prole.cfg path variable handling # --------------------------------------------------------------------------- class TestCfgPathVariables: def test_apply_inputs_keeps_cfg_path_variables_literal(self): inst = _TestableSilentInstaller() loaded = { "env_setup.PROLE_HOME": "$HOME/dev/prole", "env_setup.PROLE_CONF": "$HOME/dev/prole/conf", "env_setup.PROLE_DATA": "$HOME/.prole/data", } with ( mock.patch.object(inst, "_default_inputs", return_value={}), mock.patch.object(inst, "_load_inputs_from_cfg", return_value=loaded), ): inst._apply_inputs() assert inst.inputs["env_setup.PROLE_HOME"] == "$HOME/dev/prole" assert inst.inputs["env_setup.PROLE_CONF"] == "$HOME/dev/prole/conf" assert inst.inputs["env_setup.PROLE_DATA"] == "$HOME/.prole/data" def test_load_inputs_from_cfg_keeps_legacy_system_environment_shell_vars(self, tmp_path): import configparser cfg_dir = tmp_path / "dev" cfg_dir.mkdir(parents=True, exist_ok=True) cfg_path = cfg_dir / "prole.cfg" cfg = configparser.ConfigParser(interpolation=None) cfg.optionxform = str cfg.add_section("System Environment") cfg.set("System Environment", "PROLE_HOME", "$HOME/dev/prole") cfg.set("System Environment", "PROLE_CONF", "$HOME/dev/prole/conf") with open(cfg_path, "w") as f: cfg.write(f) inst = _TestableSilentInstaller(project_root=tmp_path) inst.cfg_path = cfg_path loaded = inst._load_inputs_from_cfg() assert loaded["env_setup.PROLE_HOME"] == "$HOME/dev/prole" assert loaded["env_setup.PROLE_CONF"] == "$HOME/dev/prole/conf" def test_write_cfg_normalizes_global_prole_home_to_home_var(self, tmp_path, monkeypatch): fake_home = tmp_path / "testuser" project_root = fake_home / "dev" / "prole" project_root.mkdir(parents=True, exist_ok=True) cfg_path = tmp_path / "prole.cfg" inst = _TestableSilentInstaller( inputs={ "env_setup.PROLE_HOME": str(project_root), "init_cluster.cluster_env": "prod", }, project_root=project_root, ) inst.inputs = {} inst.docker_import_dir = None inst.cfg_path = cfg_path monkeypatch.setattr("knoe.core.actions.Path.home", lambda: fake_home) inst._write_cfg() cfg_text = cfg_path.read_text() assert "PROLE_HOME = $HOME/dev/prole" in cfg_text assert str(project_root) not in cfg_text # --------------------------------------------------------------------------- # _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" def test_exports_cnpg_planner_rendered_env_for_shell_consumption(self, tmp_path): inst = _TestableInstaller( inputs={ "env_setup.PROLE_CONF": str(tmp_path / "conf"), "init_password.db_namespace": "knoe-system", }, project_root=tmp_path, ) inst.prole_cfg_data.setdefault("Global", {})["CNPG_ELIGIBLE_NODES"] = "db-b,db-a" env = inst._script_env_for_namespace("knoe-system") plan_file = Path(env.get("CNPG_PLACEMENT_PLAN_FILE", "")) assert plan_file.exists() assert env.get("CNPG_PLACEMENT_ELIGIBLE_NODES") == "db-a,db-b" assert env.get("CNPG_PLACEMENT_PLAN_ID", "").startswith("cnpg-placement-") # Existing shell flow can consume compatibility selector/instance inputs. assert env.get("CNPG_CLUSTER_NAME") == "knoe-db" assert env.get("CNPG_INSTANCES") == "3" def test_resolves_db_kubecontext_from_selected_gke_context(self): inst = _TestableInstaller(inputs={"init_cluster.cluster_env": "prole-prod-cluster"}) inst.prole_cfg_data.setdefault("Global", {})["KUBECONTEXT"] = ( "gke_plenary-truck-485623-p7_us-west3_knoe-dev-0" ) inst.prole_cfg_data.setdefault("Global", {})["DB_CLUSTER_NAME"] = "knoe-cnpg-0" env = inst._script_env_for_namespace("knoe-db-0", cluster_role="db") assert ( env.get("KUBECONTEXT") == "gke_plenary-truck-485623-p7_us-west3_knoe-cnpg-0" ) # --------------------------------------------------------------------------- # 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["knoe.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] == "knoe-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] == "knoe-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["knoe.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): import knoe.core.actions as actions_mod 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) monkeypatch.setattr(inst, "_ensure_cnpg_storage_provisioned", lambda *a, **k: None) # CNPG is now Python-owned; patch to avoid real kubectl calls in unit tests monkeypatch.setattr(actions_mod, "cnpg_initialize", lambda **_kw: None) monkeypatch.setattr(actions_mod.registry_ops, "update", lambda **_kw: None) monkeypatch.setattr(actions_mod.openbao_ops, "update", lambda **_kw: None) monkeypatch.setattr(actions_mod.garage_store_ops, "update", lambda **_kw: None) monkeypatch.setattr(actions_mod.opentofu_ops, "update", lambda **_kw: None) monkeypatch.setattr(actions_mod.monitoring_ops, "initialize", lambda **_kw: None) inst._step_init_scripts() scripts = [c.args[0] for c in inst.controller.run_script.call_args_list] assert "init_common_services.sh" not in scripts assert "init_certmgr.sh" in scripts # CNPG shell script must NOT be dispatched assert "init_cloudnative_pg.sh" not in scripts assert inst.prole_cfg_data["Initialization Scripts"]["STATUS"] == "Completed" assert inst._scripts_success is True def test_verifies_critical_secrets_with_db_env(self, monkeypatch): import knoe.core.actions as actions_mod 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") db_env = { "NAMESPACE": "default", "DATABASE_NAMESPACE": "default", "CLUSTER_NAME": "knoe-db", } app_env = {"NAMESPACE": "knoe-system"} def _fake_script_env(_ns, cluster_role="db"): return db_env if cluster_role == "db" else app_env monkeypatch.setattr(inst, "_script_env_for_namespace", _fake_script_env) monkeypatch.setattr(inst, "ensure_db_k8s_secrets", lambda *a, **k: None) monkeypatch.setattr(inst, "_ensure_cnpg_storage_provisioned", lambda *a, **k: None) monkeypatch.setattr(inst, "_optional_workloads_policy", lambda _env: (False, 0, "skipped")) monkeypatch.setattr(actions_mod, "cnpg_initialize", lambda **_kw: None) monkeypatch.setattr(actions_mod.registry_ops, "update", lambda **_kw: None) monkeypatch.setattr(actions_mod.openbao_ops, "update", lambda **_kw: None) monkeypatch.setattr(actions_mod.garage_store_ops, "update", lambda **_kw: None) monkeypatch.setattr(actions_mod.opentofu_ops, "update", lambda **_kw: None) secret_calls = [] def _fake_run_cmd(cmd, cwd=None, env=None, **_kwargs): if isinstance(cmd, list) and cmd[:3] == ["kubectl", "get", "secret"]: secret_calls.append((cmd, env)) return 0 monkeypatch.setattr(inst, "_run_cmd", _fake_run_cmd, raising=False) inst._step_init_scripts() assert len(secret_calls) == 3 assert all(call_env is db_env for _, call_env in secret_calls) # --------------------------------------------------------------------------- # _step_cnpg_deploy # --------------------------------------------------------------------------- class TestStepCnpgDeploy: def test_deploy_invokes_cnpg_with_db_env(self, monkeypatch): import knoe.core.actions as actions_mod inst = _TestableSilentInstaller( inputs={ "init_cnpg_deploy.run_deploy": "true", "init_cnpg_deploy.force_rollout": "false", "init_password.db_namespace": "knoe-db", } ) inst.prole_cfg_data.setdefault("Deployment", {}) env = {"CNPG_CLUSTER_NAME": "cluster-a"} monkeypatch.setattr( inst, "_script_env_for_namespace", lambda _ns, cluster_role="db": env, ) calls = {"deploy": 0} def _fake_cnpg_deploy(namespace, cluster_name, env=None, **_kwargs): calls["deploy"] += 1 assert namespace == "knoe-db" assert cluster_name == "cluster-a" assert env is env monkeypatch.setattr(actions_mod, "cnpg_deploy", _fake_cnpg_deploy) inst._step_cnpg_deploy() assert calls["deploy"] == 1 assert inst._cnpg_success is True assert inst.prole_cfg_data["Deployment"]["STATUS"] == "Deployed" def test_stops_deploy_on_storage_provision_failure(self, monkeypatch): inst = _TestableSilentInstaller( inputs={ "init_cnpg_deploy.run_deploy": "true", "init_cnpg_deploy.force_rollout": "false", "init_password.db_namespace": "knoe-db", } ) inst.prole_cfg_data.setdefault("Deployment", {}) monkeypatch.setattr( inst, "_script_env_for_namespace", lambda _ns, cluster_role="db": {"CNPG_CLUSTER_NAME": "cluster-a"}, ) monkeypatch.setattr( inst, "_ensure_cnpg_storage_provisioned", lambda *_args, **_kwargs: (_ for _ in ()).throw(RuntimeError("boom")), ) run_mock = MagicMock(return_value=0) monkeypatch.setattr(inst, "_run_cmd", run_mock) inst._step_cnpg_deploy() run_mock.assert_not_called() assert inst._cnpg_success is False assert inst.prole_cfg_data["Deployment"]["STATUS"] == "Attempted" # --------------------------------------------------------------------------- # _prepare_opentofu_pipeline # --------------------------------------------------------------------------- class TestPrepareOpenTofuPipeline: def test_runs_for_k3d_mode_when_k3s_settings_present(self, tmp_path, monkeypatch): import knoe.core.actions as actions_mod inst = _TestableSilentInstaller( inputs={ "init_cluster.cluster_env": "dev", "init_cluster.mode": "k3d", "init_password.db_namespace": "test-ns", "init_cluster.k3s_server_url": "myrddin.prole.org:6443", "init_cluster.k3s_token": "dummy", }, project_root=tmp_path, ) inst.prole_cfg_data.setdefault("Deployment", {}) expected_dir = tmp_path / "deploy" / "opentofu" / "k3s" called = {"count": 0} def _fake_sync(project_root, namespace, k3s_server_url, k3s_token, log_fn=None): called["count"] += 1 assert project_root == tmp_path assert namespace == "test-ns" assert k3s_server_url.startswith("https://") assert k3s_token == "dummy" return expected_dir monkeypatch.setattr(actions_mod, "_sync_opentofu_pipeline", _fake_sync) inst._prepare_opentofu_pipeline() assert called["count"] == 1 assert inst.prole_cfg_data["Deployment"]["OPENTOFU_PIPELINE_DIR"] == str( expected_dir ) # --------------------------------------------------------------------------- # _step_dependencies / gcloud auth preflight # --------------------------------------------------------------------------- class TestStepDependencies: def test_step_dependencies_marks_missing_when_k8s_auth_missing(self, monkeypatch): inst = _TestableSilentInstaller(inputs={"init_cluster.cluster_env": "prod"}) inst.dependencies = [{"id": "gcloud", "name": "gcloud"}] monkeypatch.setattr( "knoe.core.actions.inst_config.get_dep_info", lambda _dep: (True, "/usr/bin/gcloud", "1.0"), ) monkeypatch.setattr(inst, "_ensure_gcloud_auth_for_k8s", lambda: False) ok = inst._step_dependencies() assert ok is False assert inst.prole_cfg_data["Dependencies"]["STATUS"] == "Missing" def test_ensure_gcloud_auth_for_k8s_non_interactive_failure(self, monkeypatch): import knoe.core.actions as actions_mod inst = _TestableSilentInstaller(inputs={"init_cluster.cluster_env": "prod"}) def _unexpected_gcloud_plain(*_a, **_k): raise AssertionError("_gcloud_plain should not be called for auth preflight") monkeypatch.setattr( actions_mod.inst_config, "_gcloud_plain", _unexpected_gcloud_plain, ) monkeypatch.setattr( actions_mod.inst_config, "_augment_env_for_brew", lambda _env: {"PATH": "x"}, ) monkeypatch.setattr( actions_mod.subprocess, "run", lambda *_a, **_k: subprocess.CompletedProcess([], 1, stdout="", stderr="no auth"), ) monkeypatch.setattr(actions_mod.sys.stdin, "isatty", lambda: False, raising=False) errors = [] monkeypatch.setattr(inst, "err", lambda msg: errors.append(msg)) ok = inst._ensure_gcloud_auth_for_k8s() assert ok is False assert any("gcloud auth login --no-launch-browser" in msg for msg in errors) def test_ensure_gcloud_auth_for_k8s_interactive_login_success(self, monkeypatch): import knoe.core.actions as actions_mod inst = _TestableSilentInstaller(inputs={"init_cluster.cluster_env": "prod"}) def _unexpected_gcloud_plain(*_a, **_k): raise AssertionError("_gcloud_plain should not be called for auth preflight") monkeypatch.setattr( actions_mod.inst_config, "_gcloud_plain", _unexpected_gcloud_plain, ) monkeypatch.setattr( actions_mod.inst_config, "_augment_env_for_brew", lambda _env: {"PATH": "x"}, ) responses = [ subprocess.CompletedProcess([], 1, stdout="", stderr="no auth"), subprocess.CompletedProcess([], 0, stdout="tok123\n", stderr=""), ] def _fake_run(*_a, **_k): return responses.pop(0) monkeypatch.setattr(actions_mod.subprocess, "run", _fake_run) monkeypatch.setattr(actions_mod.sys.stdin, "isatty", lambda: True, raising=False) login_calls = [] monkeypatch.setattr(inst, "_run_cmd", lambda cmd, **_kw: login_calls.append(cmd) or 0) ok = inst._ensure_gcloud_auth_for_k8s() assert ok is True assert login_calls