mirror of
https://github.com/dredx/prole.git
synced 2026-09-24 18:24:32 +00:00
1198 lines
45 KiB
Python
1198 lines
45 KiB
Python
"""Tests for installer/core/actions.py – KnoeInstallerBase 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-knoe")
|
||
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", <path>, "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.knoe_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.knoe_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.knoe_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" / "knoe"
|
||
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" / "knoe"
|
||
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-knoe")
|
||
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.knoe_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.knoe_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.knoe_cfg_data.setdefault("Docker Build", {})
|
||
inst.knoe_cfg_data["Docker Build"]["LOCAL_REGISTRY"] = "localhost:5000"
|
||
inst.knoe_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.knoe_cfg_data["Docker Build"]["LOCAL_REGISTRY"] == "k3s.example.test:5000"
|
||
assert (
|
||
inst.knoe_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.knoe_cfg_data.setdefault("Docker Build", {})
|
||
inst.knoe_cfg_data["Docker Build"].pop("LOCAL_REGISTRY", None)
|
||
inst.knoe_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.knoe_cfg_data["Docker Build"]["STATUS"] == "Failed"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 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 / ".knoe").mkdir(parents=True)
|
||
(fake_home / ".knoe" / "env.sh").write_text(
|
||
'\n'.join(
|
||
[
|
||
'export KNOE_HOME="/Users/testuser/dev/knoe"',
|
||
'export KNOE_CONF="/Users/testuser/dev/knoe/conf"',
|
||
'export PROLE_DATA="/Users/testuser/.knoe/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("KNOE_HOME", None)
|
||
os.environ.pop("KNOE_CONF", None)
|
||
os.environ.pop("PROLE_DATA", None)
|
||
env = inst._env_defaults(namespace="knoe-db")
|
||
|
||
assert env["KNOE_HOME"] == str(fake_home / "dev" / "knoe")
|
||
assert env["KNOE_CONF"] == str(fake_home / "dev" / "knoe" / "conf")
|
||
assert env["PROLE_DATA"] == str(fake_home / ".knoe" / "data")
|
||
|
||
def test_save_env_to_file_uses_home_var_not_absolute_home(self, tmp_path, monkeypatch):
|
||
fake_home = tmp_path / "testuser"
|
||
knoe_home = fake_home / "dev" / "knoe"
|
||
(knoe_home / "conf").mkdir(parents=True)
|
||
(knoe_home / "etc").mkdir(parents=True)
|
||
|
||
inst = _TestableInstaller(project_root=tmp_path)
|
||
monkeypatch.setattr("knoe.core.actions.Path.home", lambda: fake_home)
|
||
|
||
values = {
|
||
"KNOE_HOME": str(knoe_home),
|
||
"KNOE_CONF": str(knoe_home / "conf"),
|
||
"PROLE_DATA": str(fake_home / ".knoe" / "data"),
|
||
"PROLE_LOGS": "/opt/knoe/logs/chrisfu",
|
||
"KNOE_SERVICE": str(knoe_home / "etc"),
|
||
}
|
||
|
||
inst._save_env_to_file(values)
|
||
|
||
env_sh = (knoe_home / "env.sh").read_text()
|
||
assert 'export KNOE_HOME="$HOME/dev/knoe"' in env_sh
|
||
assert str(fake_home) not in env_sh
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# knoe.cfg path variable handling
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestCfgPathVariables:
|
||
def test_apply_inputs_keeps_cfg_path_variables_literal(self):
|
||
inst = _TestableSilentInstaller()
|
||
loaded = {
|
||
"env_setup.KNOE_HOME": "$HOME/dev/knoe",
|
||
"env_setup.KNOE_CONF": "$HOME/dev/knoe/conf",
|
||
"env_setup.PROLE_DATA": "$HOME/.knoe/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.KNOE_HOME"] == "$HOME/dev/knoe"
|
||
assert inst.inputs["env_setup.KNOE_CONF"] == "$HOME/dev/knoe/conf"
|
||
assert inst.inputs["env_setup.PROLE_DATA"] == "$HOME/.knoe/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 / "knoe.cfg"
|
||
cfg = configparser.ConfigParser(interpolation=None)
|
||
cfg.optionxform = str
|
||
cfg.add_section("System Environment")
|
||
cfg.set("System Environment", "KNOE_HOME", "$HOME/dev/knoe")
|
||
cfg.set("System Environment", "KNOE_CONF", "$HOME/dev/knoe/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.KNOE_HOME"] == "$HOME/dev/knoe"
|
||
assert loaded["env_setup.KNOE_CONF"] == "$HOME/dev/knoe/conf"
|
||
|
||
def test_load_inputs_from_cfg_backfills_auth_oidc_refs_from_global_when_inputs_missing(
|
||
self, tmp_path
|
||
):
|
||
import configparser
|
||
|
||
cfg_path = tmp_path / "gke-test.cfg"
|
||
cfg = configparser.ConfigParser(interpolation=None)
|
||
cfg.optionxform = str
|
||
cfg.add_section("Inputs")
|
||
cfg.set("Inputs", "init_cluster.cluster_env", "prod")
|
||
cfg.add_section("Global")
|
||
cfg.set("Global", "OIDC_CLIENT_ID_REF", "secretref://google-oidc-client-id")
|
||
cfg.set("Global", "OIDC_CLIENT_SECRET_REF", "secretref://google-oidc-client-secret")
|
||
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["auth.clientId"] == "secretref://google-oidc-client-id"
|
||
assert loaded["auth.clientSecret"] == "secretref://google-oidc-client-secret"
|
||
|
||
def test_load_inputs_from_cfg_resolves_auth_oidc_secretrefs_from_env(self, tmp_path):
|
||
import configparser
|
||
|
||
cfg_path = tmp_path / "gke-test.cfg"
|
||
cfg = configparser.ConfigParser(interpolation=None)
|
||
cfg.optionxform = str
|
||
cfg.add_section("Inputs")
|
||
cfg.set("Inputs", "auth.clientId", "secretref://google-oidc-client-id")
|
||
cfg.set("Inputs", "auth.clientSecret", "secretref://google-oidc-client-secret")
|
||
with open(cfg_path, "w") as f:
|
||
cfg.write(f)
|
||
|
||
inst = _TestableSilentInstaller(project_root=tmp_path)
|
||
inst.cfg_path = cfg_path
|
||
|
||
with mock.patch.dict(
|
||
os.environ,
|
||
{
|
||
"GITLAB_OIDC_CLIENT_ID": "resolved-client-id",
|
||
"GITLAB_OIDC_CLIENT_SECRET": "resolved-client-secret",
|
||
},
|
||
clear=False,
|
||
):
|
||
loaded = inst._load_inputs_from_cfg()
|
||
|
||
assert loaded["auth.clientId"] == "resolved-client-id"
|
||
assert loaded["auth.clientSecret"] == "resolved-client-secret"
|
||
|
||
def test_load_inputs_from_cfg_resolves_legacy_auth_oidc_secretrefs_from_env(
|
||
self, tmp_path
|
||
):
|
||
import configparser
|
||
|
||
cfg_path = tmp_path / "gke-test.cfg"
|
||
cfg = configparser.ConfigParser(interpolation=None)
|
||
cfg.optionxform = str
|
||
cfg.add_section("Auth")
|
||
cfg.set("Auth", "clientIdRef", "secretref://google-oidc-client-id")
|
||
cfg.set("Auth", "clientSecretRef", "secretref://google-oidc-client-secret")
|
||
with open(cfg_path, "w") as f:
|
||
cfg.write(f)
|
||
|
||
inst = _TestableSilentInstaller(project_root=tmp_path)
|
||
inst.cfg_path = cfg_path
|
||
|
||
with mock.patch.dict(
|
||
os.environ,
|
||
{
|
||
"GITLAB_OIDC_CLIENT_ID": "legacy-client-id",
|
||
"GITLAB_OIDC_CLIENT_SECRET": "legacy-client-secret",
|
||
},
|
||
clear=False,
|
||
):
|
||
loaded = inst._load_inputs_from_cfg()
|
||
|
||
assert loaded["auth.clientId"] == "legacy-client-id"
|
||
assert loaded["auth.clientSecret"] == "legacy-client-secret"
|
||
|
||
def test_default_inputs_include_auth_oidc_secret_refs(self):
|
||
inst = _TestableSilentInstaller()
|
||
inst.dependencies = []
|
||
defaults = inst._default_inputs()
|
||
|
||
assert defaults["auth.clientId"] == "secretref://google-oidc-client-id"
|
||
assert defaults["auth.clientSecret"] == "secretref://google-oidc-client-secret"
|
||
|
||
def test_write_cfg_normalizes_global_knoe_home_to_home_var(self, tmp_path, monkeypatch):
|
||
fake_home = tmp_path / "testuser"
|
||
project_root = fake_home / "dev" / "knoe"
|
||
project_root.mkdir(parents=True, exist_ok=True)
|
||
cfg_path = tmp_path / "knoe.cfg"
|
||
|
||
inst = _TestableSilentInstaller(
|
||
inputs={
|
||
"env_setup.KNOE_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 "KNOE_HOME = $HOME/dev/knoe" 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 / "knoe.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.KNOE_CONF": str(tmp_path / "conf"),
|
||
"init_password.db_namespace": "knoe-system",
|
||
},
|
||
project_root=tmp_path,
|
||
)
|
||
inst.knoe_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": "knoe-prod-cluster"})
|
||
inst.knoe_cfg_data.setdefault("Global", {})["KUBECONTEXT"] = (
|
||
"gke_plenary-truck-485623-p7_us-west3_knoe-dev-0"
|
||
)
|
||
inst.knoe_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_knoe_ed25519").write_text("PRIVATE")
|
||
(ssh_dir / "id_knoe_ed25519.pub").write_text("PUBLIC")
|
||
|
||
monkeypatch.setattr(Path, "home", classmethod(lambda cls: fake_home))
|
||
|
||
service_dir = tmp_path / "etc"
|
||
inst = _TestableInstaller(inputs={"env_setup.KNOE_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=knoe" 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_knoe_ed25519").write_text("PRIVATE")
|
||
(ssh_dir / "id_knoe_ed25519.pub").write_text("PUBLIC")
|
||
monkeypatch.setattr(Path, "home", classmethod(lambda cls: fake_home))
|
||
|
||
service_dir = tmp_path / "etc"
|
||
inst = _TestableInstaller(inputs={"env_setup.KNOE_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.knoe_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.knoe_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.knoe_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.knoe_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.knoe_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.knoe.org:6443",
|
||
"init_cluster.k3s_token": "dummy",
|
||
},
|
||
project_root=tmp_path,
|
||
)
|
||
inst.knoe_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.knoe_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.knoe_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
|