mirror of
https://github.com/dredx/prole.git
synced 2026-09-23 12:03:59 +00:00
472 lines
18 KiB
Python
472 lines
18 KiB
Python
"""Tests for knoe/deployment.py – KnoeDeployment class."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import configparser
|
||
import os
|
||
import subprocess
|
||
from pathlib import Path
|
||
from unittest import mock
|
||
from unittest.mock import MagicMock, patch, call
|
||
|
||
from knoe.deployment import (
|
||
KnoeDeployment,
|
||
_CFG_KEY_MAP,
|
||
KUBECONFIG_FILENAME,
|
||
FETCH_PLAYBOOK,
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Helpers
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _make_controller(tmp_path):
|
||
ctrl = MagicMock()
|
||
ctrl.run_script = MagicMock(return_value=0)
|
||
return ctrl
|
||
|
||
|
||
def _make_cfg_file(tmp_path, sections=None):
|
||
"""Write a minimal knoe.cfg and return its path."""
|
||
cfg = configparser.ConfigParser(interpolation=None)
|
||
cfg.optionxform = str
|
||
defaults = {
|
||
"Global": {"NAMESPACE": "test-ns"},
|
||
"Initialize Cluster": {"ENVIRONMENT": "service"},
|
||
"Database Creation": {"DB_PASSWORD": "s3cret"},
|
||
"System Environment": {"KNOE_CONF": str(tmp_path / "conf")},
|
||
"Optional Features": {"SUPABASE_ENABLED": "false"},
|
||
"Kerberos Authentication": {"ENABLED": "false"},
|
||
}
|
||
if sections:
|
||
defaults.update(sections)
|
||
for sec, kvs in defaults.items():
|
||
cfg.add_section(sec)
|
||
for k, v in kvs.items():
|
||
cfg.set(sec, k, v)
|
||
conf_dir = tmp_path / "conf"
|
||
conf_dir.mkdir(exist_ok=True)
|
||
cfg_path = conf_dir / "knoe.cfg"
|
||
with open(cfg_path, "w") as f:
|
||
cfg.write(f)
|
||
return cfg_path
|
||
|
||
|
||
def _make_deployment(tmp_path, sections=None):
|
||
cfg_path = _make_cfg_file(tmp_path, sections)
|
||
ctrl = _make_controller(tmp_path)
|
||
with mock.patch.dict(os.environ, {"KNOE_CONF": str(tmp_path / "conf")}):
|
||
dep = KnoeDeployment(ctrl, tmp_path)
|
||
return dep, ctrl
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tests: __init__ and _load_knoe_cfg
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestKnoeDeploymentInit:
|
||
def test_init_loads_cfg(self, tmp_path):
|
||
dep, _ = _make_deployment(tmp_path)
|
||
assert dep.knoe_cfg_data["Global"]["NAMESPACE"] == "test-ns"
|
||
assert dep.knoe_cfg_data["Initialize Cluster"]["ENVIRONMENT"] == "service"
|
||
|
||
def test_init_default_namespace(self, tmp_path):
|
||
"""When knoe.cfg has no namespace, default is set."""
|
||
cfg = configparser.ConfigParser(interpolation=None)
|
||
cfg.optionxform = str
|
||
cfg.add_section("Global")
|
||
conf_dir = tmp_path / "conf"
|
||
conf_dir.mkdir(exist_ok=True)
|
||
cfg_path = conf_dir / "knoe.cfg"
|
||
with open(cfg_path, "w") as f:
|
||
cfg.write(f)
|
||
ctrl = _make_controller(tmp_path)
|
||
with mock.patch.dict(os.environ, {"KNOE_CONF": str(conf_dir)}):
|
||
dep = KnoeDeployment(ctrl, tmp_path)
|
||
assert dep.knoe_cfg_data["Global"]["NAMESPACE"] == "knoe-db"
|
||
|
||
def test_init_no_cfg_file(self, tmp_path):
|
||
"""When no knoe.cfg exists, it still initialises."""
|
||
ctrl = _make_controller(tmp_path)
|
||
with mock.patch.dict(os.environ, {"KNOE_CONF": ""}, clear=False):
|
||
dep = KnoeDeployment(ctrl, tmp_path)
|
||
assert dep.knoe_cfg_data["Global"]["NAMESPACE"] == "knoe-db"
|
||
|
||
def test_load_cfg_from_directory(self, tmp_path):
|
||
"""KNOE_CONF pointing to a directory finds knoe.cfg inside."""
|
||
_make_cfg_file(tmp_path)
|
||
ctrl = _make_controller(tmp_path)
|
||
with mock.patch.dict(os.environ, {"KNOE_CONF": str(tmp_path / "conf")}):
|
||
dep = KnoeDeployment(ctrl, tmp_path)
|
||
assert dep.cfg_path == tmp_path / "conf" / "knoe.cfg"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tests: _get_input
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestGetInput:
|
||
def test_mapped_key(self, tmp_path):
|
||
dep, _ = _make_deployment(tmp_path)
|
||
assert dep._get_input("init_password.db_namespace") == "test-ns"
|
||
|
||
def test_mapped_key_db_password(self, tmp_path):
|
||
dep, _ = _make_deployment(tmp_path)
|
||
assert dep._get_input("init_password.db_password") == "s3cret"
|
||
|
||
def test_unmapped_key_returns_default(self, tmp_path):
|
||
dep, _ = _make_deployment(tmp_path)
|
||
assert dep._get_input("nonexistent.key", "fallback") == "fallback"
|
||
|
||
def test_unmapped_key_returns_empty(self, tmp_path):
|
||
dep, _ = _make_deployment(tmp_path)
|
||
assert dep._get_input("nonexistent.key") == ""
|
||
|
||
def test_all_cfg_key_map_entries(self, tmp_path):
|
||
"""Ensure every key in _CFG_KEY_MAP resolves a section."""
|
||
dep, _ = _make_deployment(tmp_path)
|
||
for key, (section, cfg_key) in _CFG_KEY_MAP.items():
|
||
# Should not raise
|
||
dep._get_input(key, "default")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tests: _build_deploy_env
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestBuildDeployEnv:
|
||
def test_basic_env(self, tmp_path):
|
||
dep, _ = _make_deployment(tmp_path)
|
||
dep.knoe_cfg_data["Deployment"] = {"KUBECONFIG_PATH": ""}
|
||
env = dep._build_deploy_env()
|
||
assert env["KNOE_HOME"] == str(tmp_path)
|
||
assert env["KNOE_SERVICE"] == str(tmp_path)
|
||
assert env["NAMESPACE"] == "test-ns"
|
||
assert env["KNOE_MODE"] == "k3s"
|
||
|
||
def test_kubeconfig_set_when_exists(self, tmp_path):
|
||
kc_path = tmp_path / "kube.cfg"
|
||
kc_path.write_text("dummy")
|
||
dep, _ = _make_deployment(tmp_path)
|
||
dep.knoe_cfg_data["Deployment"] = {"KUBECONFIG_PATH": str(kc_path)}
|
||
env = dep._build_deploy_env()
|
||
assert env["KUBECONFIG"] == str(kc_path)
|
||
|
||
def test_kubeconfig_not_set_when_missing(self, tmp_path):
|
||
dep, _ = _make_deployment(tmp_path)
|
||
dep.knoe_cfg_data["Deployment"] = {"KUBECONFIG_PATH": "/nonexistent/path"}
|
||
env = dep._build_deploy_env()
|
||
# Should not have set KUBECONFIG to a non-existent file
|
||
assert env.get("KUBECONFIG") != "/nonexistent/path"
|
||
|
||
def test_db_password_in_env(self, tmp_path):
|
||
dep, _ = _make_deployment(tmp_path)
|
||
dep.knoe_cfg_data["Deployment"] = {"KUBECONFIG_PATH": ""}
|
||
env = dep._build_deploy_env()
|
||
assert env.get("DB_PASSWORD") == "s3cret"
|
||
assert env.get("OPENTOFU_ADMIN_PASSWORD") == "s3cret"
|
||
|
||
def test_knoe_conf_in_env(self, tmp_path):
|
||
dep, _ = _make_deployment(tmp_path)
|
||
dep.knoe_cfg_data["Deployment"] = {"KUBECONFIG_PATH": ""}
|
||
env = dep._build_deploy_env()
|
||
assert env.get("KNOE_CONF") == str(tmp_path / "conf")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tests: _fetch_kubeconfig
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestFetchKubeconfig:
|
||
def test_playbook_not_found(self, tmp_path):
|
||
dep, _ = _make_deployment(tmp_path)
|
||
result = dep._fetch_kubeconfig()
|
||
assert result is None
|
||
|
||
@patch("subprocess.run")
|
||
def test_ansible_playbook_success(self, mock_run, tmp_path):
|
||
dep, _ = _make_deployment(tmp_path)
|
||
playbook = tmp_path / FETCH_PLAYBOOK
|
||
playbook.parent.mkdir(parents=True, exist_ok=True)
|
||
playbook.write_text("---")
|
||
kc = tmp_path / KUBECONFIG_FILENAME
|
||
kc.write_text("apiVersion: v1")
|
||
mock_run.return_value = MagicMock(returncode=0)
|
||
result = dep._fetch_kubeconfig()
|
||
assert result == kc
|
||
|
||
@patch(
|
||
"subprocess.run",
|
||
side_effect=subprocess.CalledProcessError(1, "ansible-playbook"),
|
||
)
|
||
def test_ansible_playbook_failure(self, mock_run, tmp_path):
|
||
dep, _ = _make_deployment(tmp_path)
|
||
playbook = tmp_path / FETCH_PLAYBOOK
|
||
playbook.parent.mkdir(parents=True, exist_ok=True)
|
||
playbook.write_text("---")
|
||
result = dep._fetch_kubeconfig()
|
||
assert result is None
|
||
|
||
@patch("subprocess.run", side_effect=FileNotFoundError())
|
||
def test_ansible_not_installed(self, mock_run, tmp_path):
|
||
dep, _ = _make_deployment(tmp_path)
|
||
playbook = tmp_path / FETCH_PLAYBOOK
|
||
playbook.parent.mkdir(parents=True, exist_ok=True)
|
||
playbook.write_text("---")
|
||
result = dep._fetch_kubeconfig()
|
||
assert result is None
|
||
|
||
@patch("subprocess.run")
|
||
def test_kubeconfig_not_written(self, mock_run, tmp_path):
|
||
dep, _ = _make_deployment(tmp_path)
|
||
playbook = tmp_path / FETCH_PLAYBOOK
|
||
playbook.parent.mkdir(parents=True, exist_ok=True)
|
||
playbook.write_text("---")
|
||
mock_run.return_value = MagicMock(returncode=0)
|
||
# Don't create the kubeconfig file
|
||
result = dep._fetch_kubeconfig()
|
||
assert result is None
|
||
|
||
@patch("subprocess.run")
|
||
def test_vault_pass_used(self, mock_run, tmp_path):
|
||
dep, _ = _make_deployment(tmp_path)
|
||
playbook = tmp_path / FETCH_PLAYBOOK
|
||
playbook.parent.mkdir(parents=True, exist_ok=True)
|
||
playbook.write_text("---")
|
||
vault_pass = tmp_path / ".vault_pass"
|
||
vault_pass.write_text("secret")
|
||
kc = tmp_path / KUBECONFIG_FILENAME
|
||
kc.write_text("apiVersion: v1")
|
||
mock_run.return_value = MagicMock(returncode=0)
|
||
result = dep._fetch_kubeconfig()
|
||
assert result is not None
|
||
cmd_args = mock_run.call_args[0][0]
|
||
assert "--vault-password-file" in cmd_args
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tests: apply
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestApply:
|
||
def test_no_pipeline_dir(self, tmp_path):
|
||
dep, _ = _make_deployment(tmp_path)
|
||
dep.knoe_cfg_data["Deployment"] = {}
|
||
assert dep.apply() is False
|
||
|
||
def test_pipeline_dir_nonexistent(self, tmp_path):
|
||
dep, _ = _make_deployment(tmp_path)
|
||
dep.knoe_cfg_data["Deployment"] = {"OPENTOFU_PIPELINE_DIR": "/nonexistent"}
|
||
assert dep.apply() is False
|
||
|
||
@patch("subprocess.run")
|
||
def test_tofu_init_fails(self, mock_run, tmp_path):
|
||
pipeline = tmp_path / "pipeline"
|
||
pipeline.mkdir()
|
||
dep, _ = _make_deployment(tmp_path)
|
||
dep.knoe_cfg_data["Deployment"] = {"OPENTOFU_PIPELINE_DIR": str(pipeline)}
|
||
mock_run.side_effect = subprocess.CalledProcessError(1, "tofu")
|
||
assert dep.apply() is False
|
||
|
||
@patch("subprocess.run", side_effect=FileNotFoundError())
|
||
def test_tofu_not_found(self, mock_run, tmp_path):
|
||
pipeline = tmp_path / "pipeline"
|
||
pipeline.mkdir()
|
||
dep, _ = _make_deployment(tmp_path)
|
||
dep.knoe_cfg_data["Deployment"] = {"OPENTOFU_PIPELINE_DIR": str(pipeline)}
|
||
assert dep.apply() is False
|
||
|
||
@patch.object(KnoeDeployment, "_run_post_apply_scripts", return_value=True)
|
||
@patch("subprocess.run")
|
||
def test_apply_success(self, mock_run, mock_post, tmp_path):
|
||
pipeline = tmp_path / "pipeline"
|
||
pipeline.mkdir()
|
||
dep, ctrl = _make_deployment(tmp_path)
|
||
dep.knoe_cfg_data["Deployment"] = {
|
||
"OPENTOFU_PIPELINE_DIR": str(pipeline),
|
||
"KUBECONFIG_PATH": "",
|
||
}
|
||
mock_run.return_value = MagicMock(returncode=0)
|
||
assert dep.apply() is True
|
||
# Verify tofu init and apply were called
|
||
assert mock_run.call_count == 2
|
||
mock_post.assert_called_once()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tests: _run_post_apply_scripts
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestRunPostApplyScripts:
|
||
def test_all_scripts_succeed(self, tmp_path):
|
||
dep, ctrl = _make_deployment(tmp_path)
|
||
dep.knoe_cfg_data["Deployment"] = {"KUBECONFIG_PATH": ""}
|
||
dep._run_script = MagicMock(return_value=0)
|
||
dep._run_supabase_deploy = MagicMock(return_value=True)
|
||
with patch("knoe.deployment.cnpg_initialize", return_value=None):
|
||
assert dep._run_post_apply_scripts() is True
|
||
assert dep._run_script.call_count == 4
|
||
|
||
def test_script_failure(self, tmp_path):
|
||
dep, ctrl = _make_deployment(tmp_path)
|
||
dep.knoe_cfg_data["Deployment"] = {"KUBECONFIG_PATH": ""}
|
||
dep._run_script = MagicMock(return_value=1)
|
||
dep._run_supabase_deploy = MagicMock(return_value=True)
|
||
with patch("knoe.deployment.cnpg_initialize", return_value=None):
|
||
assert dep._run_post_apply_scripts() is False
|
||
|
||
def test_kerberos_flag(self, tmp_path):
|
||
dep, ctrl = _make_deployment(
|
||
tmp_path,
|
||
{
|
||
"Kerberos Authentication": {"ENABLED": "true"},
|
||
},
|
||
)
|
||
dep.knoe_cfg_data["Deployment"] = {"KUBECONFIG_PATH": ""}
|
||
dep._run_script = MagicMock(return_value=0)
|
||
dep._run_supabase_deploy = MagicMock(return_value=True)
|
||
with patch("knoe.deployment.cnpg_initialize", return_value=None):
|
||
dep._run_post_apply_scripts()
|
||
first_call = dep._run_script.call_args_list[0]
|
||
assert "init_common_services" in str(first_call)
|
||
assert "-k" in str(first_call)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tests: _run_supabase_deploy
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestRunSupabaseDeploy:
|
||
def test_supabase_disabled(self, tmp_path):
|
||
dep, ctrl = _make_deployment(tmp_path)
|
||
dep._run_cmd = MagicMock(return_value=0)
|
||
env = os.environ.copy()
|
||
assert dep._run_supabase_deploy(env) is True
|
||
dep._run_cmd.assert_not_called()
|
||
|
||
def test_supabase_enabled_script_missing(self, tmp_path):
|
||
dep, ctrl = _make_deployment(
|
||
tmp_path,
|
||
{
|
||
"Optional Features": {"SUPABASE_ENABLED": "true"},
|
||
},
|
||
)
|
||
env = os.environ.copy()
|
||
assert dep._run_supabase_deploy(env) is False
|
||
|
||
def test_supabase_enabled_success(self, tmp_path):
|
||
dep, ctrl = _make_deployment(
|
||
tmp_path,
|
||
{
|
||
"Optional Features": {"SUPABASE_ENABLED": "true"},
|
||
},
|
||
)
|
||
script = tmp_path / "supabase" / "deploy.sh"
|
||
script.parent.mkdir(parents=True, exist_ok=True)
|
||
script.write_text("#!/bin/bash\nexit 0")
|
||
dep._run_cmd = MagicMock(return_value=0)
|
||
env = os.environ.copy()
|
||
assert dep._run_supabase_deploy(env) is True
|
||
|
||
def test_supabase_enabled_failure(self, tmp_path):
|
||
dep, ctrl = _make_deployment(
|
||
tmp_path,
|
||
{
|
||
"Optional Features": {"SUPABASE_ENABLED": "true"},
|
||
},
|
||
)
|
||
script = tmp_path / "supabase" / "deploy.sh"
|
||
script.parent.mkdir(parents=True, exist_ok=True)
|
||
script.write_text("#!/bin/bash\nexit 1")
|
||
dep._run_cmd = MagicMock(return_value=1)
|
||
env = os.environ.copy()
|
||
assert dep._run_supabase_deploy(env) is False
|
||
|
||
def test_supabase_with_cfg_path(self, tmp_path):
|
||
dep, ctrl = _make_deployment(
|
||
tmp_path,
|
||
{
|
||
"Optional Features": {"SUPABASE_ENABLED": "true"},
|
||
},
|
||
)
|
||
script = tmp_path / "supabase" / "deploy.sh"
|
||
script.parent.mkdir(parents=True, exist_ok=True)
|
||
script.write_text("#!/bin/bash\nexit 0")
|
||
dep._run_cmd = MagicMock(return_value=0)
|
||
env = os.environ.copy()
|
||
dep._run_supabase_deploy(env)
|
||
cmd_args = dep._run_cmd.call_args[0][0]
|
||
assert "--mode" in cmd_args
|
||
assert "k8s" in cmd_args
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tests: duplicate_k3d_to_k3s
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestDuplicateK3dToK3s:
|
||
@patch.object(KnoeDeployment, "_fetch_kubeconfig", return_value=None)
|
||
@patch(
|
||
"knoe.deployment._detect_ansible_topology",
|
||
return_value={"k3s_server_url": "https://example.com:6443"},
|
||
)
|
||
def test_no_kubeconfig(self, mock_topology, mock_fetch, tmp_path):
|
||
dep, _ = _make_deployment(tmp_path)
|
||
assert dep.duplicate_k3d_to_k3s() is False
|
||
|
||
@patch("knoe.deployment._detect_ansible_topology", return_value={})
|
||
def test_no_server_url(self, mock_topology, tmp_path):
|
||
dep, _ = _make_deployment(tmp_path)
|
||
assert dep.duplicate_k3d_to_k3s() is False
|
||
|
||
@patch("knoe.deployment._sync_opentofu_pipeline", side_effect=Exception("boom"))
|
||
@patch.object(KnoeDeployment, "_fetch_kubeconfig")
|
||
@patch(
|
||
"knoe.deployment._detect_ansible_topology",
|
||
return_value={
|
||
"k3s_server_url": "https://example.com:6443",
|
||
"k3s_token": "tok123",
|
||
},
|
||
)
|
||
def test_sync_failure(self, mock_topology, mock_fetch, mock_sync, tmp_path):
|
||
kc = tmp_path / KUBECONFIG_FILENAME
|
||
kc.write_text("dummy")
|
||
mock_fetch.return_value = kc
|
||
dep, _ = _make_deployment(tmp_path)
|
||
assert dep.duplicate_k3d_to_k3s() is False
|
||
|
||
@patch("knoe.deployment._sync_opentofu_pipeline")
|
||
@patch.object(KnoeDeployment, "_fetch_kubeconfig")
|
||
@patch(
|
||
"knoe.deployment._detect_ansible_topology",
|
||
return_value={
|
||
"k3s_server_url": "https://example.com:6443",
|
||
"k3s_token": "tok123",
|
||
},
|
||
)
|
||
def test_success(self, mock_topology, mock_fetch, mock_sync, tmp_path):
|
||
kc = tmp_path / KUBECONFIG_FILENAME
|
||
kc.write_text("dummy")
|
||
mock_fetch.return_value = kc
|
||
pipeline_dir = tmp_path / "deploy" / "opentofu" / "k3s"
|
||
pipeline_dir.mkdir(parents=True)
|
||
tfvars = pipeline_dir / "opentofu.auto.tfvars"
|
||
tfvars.write_text('namespace = "test"\n')
|
||
mock_sync.return_value = pipeline_dir
|
||
dep, _ = _make_deployment(tmp_path)
|
||
assert dep.duplicate_k3d_to_k3s() is True
|
||
assert dep.knoe_cfg_data["Deployment"]["OPENTOFU_PIPELINE_DIR"] == str(
|
||
pipeline_dir
|
||
)
|
||
assert dep.knoe_cfg_data["Deployment"]["KUBECONFIG_PATH"] == str(kc)
|