mirror of
https://github.com/dredx/prole.git
synced 2026-09-24 19:34:31 +00:00
6741a684dd
- Added `gke-gcloud-auth-plugin` as a required dependency for `linux/apt` environments. - Enhanced GKE context acquisition with writable kubeconfig target resolution. - Improved failure reporting with detailed messages for GKE dependency and context issues. - Expanded test coverage for scenarios involving missing auth plugins, kubeconfig errors, and GKE configurations.
1004 lines
42 KiB
Python
1004 lines
42 KiB
Python
"""
|
|
Extended unit tests for installer/core/milestones.py
|
|
|
|
Covers the previously untested milestone classes:
|
|
_stream_line, DatabaseCreationMilestone, DockerBuildMilestone,
|
|
GitOpsMilestone, KerberosMilestone, SupabaseMilestone
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch, call
|
|
import pytest
|
|
|
|
from knoe.core.milestones import (
|
|
_stream_line,
|
|
DatabaseCreationMilestone,
|
|
DockerBuildMilestone,
|
|
GitOpsMilestone,
|
|
KerberosMilestone,
|
|
SupabaseImagePreloadMilestone,
|
|
SupabaseMilestone,
|
|
DependenciesMilestone,
|
|
InitializationScriptsMilestone,
|
|
)
|
|
from knoe.state import InstallerState
|
|
from knoe.core.controller import KnoeController
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _make_state(**inputs) -> InstallerState:
|
|
state = InstallerState()
|
|
state.inputs.update(inputs)
|
|
state.controller = MagicMock(spec=KnoeController)
|
|
state.controller.project_root = Path("/tmp/fake-prole")
|
|
state.controller.get_knoe_db_version.return_value = "18-001"
|
|
state.controller.run_script.return_value = 0
|
|
return state
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _stream_line
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_stream_line_writes_to_stdout(capsys):
|
|
_stream_line("hello world\n")
|
|
out = capsys.readouterr().out
|
|
assert "hello world" in out
|
|
|
|
|
|
def test_stream_line_empty_string(capsys):
|
|
_stream_line("")
|
|
# Should not raise
|
|
|
|
|
|
def test_stream_line_tolerates_broken_stdout():
|
|
with patch("sys.stdout") as mock_stdout:
|
|
mock_stdout.write.side_effect = OSError("broken pipe")
|
|
# Should not raise
|
|
_stream_line("test")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# DatabaseCreationMilestone
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_database_creation_defaults():
|
|
state = _make_state()
|
|
milestone = DatabaseCreationMilestone()
|
|
with patch("knoe.core.milestones.inst_config._encrypt_cfg_secret", return_value="enc:xxx"), \
|
|
patch("knoe.core.milestones.inst_config._is_prole_secret", return_value=False):
|
|
milestone.execute(state)
|
|
assert (
|
|
state.config_data.get("Database Creation", {}).get("DATABASE_NAMESPACE")
|
|
== "default"
|
|
)
|
|
assert state.config_data.get("Database Creation", {}).get("CLUSTER_NAME") == "knoe-db"
|
|
|
|
|
|
def test_database_creation_with_provided_values():
|
|
state = _make_state(**{
|
|
"init_password.db_namespace": "mydb",
|
|
"init_password.db_username": "alice",
|
|
"init_password.db_password": "secret123",
|
|
"init_password.generate_ssh_key": "False",
|
|
})
|
|
milestone = DatabaseCreationMilestone()
|
|
with patch("knoe.core.milestones.inst_config._encrypt_cfg_secret", return_value="enc:xxx"), \
|
|
patch("knoe.core.milestones.inst_config._is_prole_secret", return_value=False):
|
|
milestone.execute(state)
|
|
db = state.config_data.get("Database Creation", {})
|
|
assert db["DB_USER"] == "alice"
|
|
assert db["DATABASE_NAMESPACE"] == "mydb"
|
|
|
|
|
|
def test_database_creation_generates_password_when_missing():
|
|
state = _make_state(**{
|
|
"init_password.db_username": "bob",
|
|
"init_password.generate_ssh_key": "False",
|
|
})
|
|
milestone = DatabaseCreationMilestone()
|
|
with patch("knoe.core.milestones.inst_config._encrypt_cfg_secret", return_value="enc:xxx"), \
|
|
patch("knoe.core.milestones.inst_config._is_prole_secret", return_value=False):
|
|
milestone.execute(state)
|
|
# Password should have been generated and stored
|
|
assert state.inputs.get("init_password.db_password")
|
|
assert len(state.inputs["init_password.db_password"]) >= 24
|
|
|
|
|
|
def test_database_creation_resolves_secret_ref():
|
|
state = _make_state(**{
|
|
"init_password.db_username": "bob",
|
|
"init_password.db_password": "${openbao:kv/prole/db}",
|
|
"init_password.generate_ssh_key": "False",
|
|
})
|
|
milestone = DatabaseCreationMilestone()
|
|
with patch("knoe.core.milestones.inst_config._resolve_secret_value", return_value="${openbao:kv/prole/db}"), \
|
|
patch("knoe.core.milestones.inst_config._encrypt_cfg_secret", return_value="enc:xxx"), \
|
|
patch("knoe.core.milestones.inst_config._is_prole_secret", return_value=False):
|
|
milestone.execute(state)
|
|
# Unresolvable ref → new password generated
|
|
pw = state.inputs.get("init_password.db_password", "")
|
|
assert not pw.startswith("${")
|
|
|
|
|
|
def test_database_creation_resolves_secret_ref_success():
|
|
state = _make_state(**{
|
|
"init_password.db_username": "bob",
|
|
"init_password.db_password": "${openbao:kv/prole/db}",
|
|
"init_password.generate_ssh_key": "False",
|
|
})
|
|
milestone = DatabaseCreationMilestone()
|
|
with patch("knoe.core.milestones.inst_config._resolve_secret_value", return_value="realpassword"), \
|
|
patch("knoe.core.milestones.inst_config._encrypt_cfg_secret", return_value="enc:xxx"), \
|
|
patch("knoe.core.milestones.inst_config._is_prole_secret", return_value=False):
|
|
milestone.execute(state)
|
|
assert state.inputs["init_password.db_password"] == "realpassword"
|
|
|
|
|
|
def test_database_creation_with_progress():
|
|
state = _make_state(**{"init_password.generate_ssh_key": "False"})
|
|
milestone = DatabaseCreationMilestone()
|
|
progress = MagicMock()
|
|
with patch("knoe.core.milestones.inst_config._encrypt_cfg_secret", return_value="enc:xxx"), \
|
|
patch("knoe.core.milestones.inst_config._is_prole_secret", return_value=False):
|
|
milestone.execute(state, progress=progress)
|
|
progress.assert_called()
|
|
|
|
|
|
def test_database_creation_uses_env_namespace_fallback():
|
|
state = _make_state(**{
|
|
"env_setup.NAMESPACE": "staging",
|
|
"init_password.generate_ssh_key": "False",
|
|
})
|
|
milestone = DatabaseCreationMilestone()
|
|
with patch("knoe.core.milestones.inst_config._encrypt_cfg_secret", return_value="enc:xxx"), \
|
|
patch("knoe.core.milestones.inst_config._is_prole_secret", return_value=False):
|
|
milestone.execute(state)
|
|
assert state.config_data["Database Creation"]["DATABASE_NAMESPACE"] == "staging"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# DockerBuildMilestone
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_docker_build_disabled():
|
|
state = _make_state(**{"init_db_build.run_build": "False"})
|
|
milestone = DockerBuildMilestone()
|
|
milestone.execute(state)
|
|
state.controller.run_script.assert_not_called()
|
|
|
|
|
|
def test_docker_build_runs_script():
|
|
state = _make_state(**{
|
|
"init_db_build.run_build": "True",
|
|
"init_cluster.cluster_env": "dev",
|
|
})
|
|
milestone = DockerBuildMilestone()
|
|
with patch("knoe.core.env._normalize_cluster_env", return_value="dev"):
|
|
milestone.execute(state)
|
|
state.controller.run_script.assert_called_once()
|
|
args, kwargs = state.controller.run_script.call_args
|
|
assert args[0] == "build_db.sh"
|
|
|
|
|
|
def test_docker_build_non_dev_env_adds_registry():
|
|
state = _make_state(**{
|
|
"init_db_build.run_build": "True",
|
|
"init_cluster.cluster_env": "prod",
|
|
"init_cluster.k3s_server_url": "https://myrddin.prole.org:6443",
|
|
# When init scripts are disabled, the DB build milestone is responsible
|
|
# for pushing the image to the registry in non-dev envs.
|
|
"init_scripts.run_scripts": "False",
|
|
})
|
|
milestone = DockerBuildMilestone()
|
|
with patch("knoe.core.env._normalize_cluster_env", return_value="prod"):
|
|
milestone.execute(state)
|
|
state.controller.run_script.assert_called_once()
|
|
_, kwargs = state.controller.run_script.call_args
|
|
build_args = kwargs.get("args", state.controller.run_script.call_args[1].get("args", []))
|
|
# Should have --push in args
|
|
assert "--push" in build_args
|
|
|
|
|
|
def test_docker_build_non_dev_env_skips_when_init_scripts_enabled():
|
|
state = _make_state(**{
|
|
"init_db_build.run_build": "True",
|
|
"init_cluster.cluster_env": "prod",
|
|
"init_cluster.k3s_server_url": "https://myrddin.prole.org:6443",
|
|
"init_scripts.run_scripts": "True",
|
|
})
|
|
milestone = DockerBuildMilestone()
|
|
with patch("knoe.core.env._normalize_cluster_env", return_value="prod"):
|
|
milestone.execute(state)
|
|
state.controller.run_script.assert_not_called()
|
|
|
|
|
|
def test_docker_build_failed_rc():
|
|
state = _make_state(**{"init_db_build.run_build": "True"})
|
|
state.controller.run_script.return_value = 1
|
|
milestone = DockerBuildMilestone()
|
|
with patch("knoe.core.env._normalize_cluster_env", return_value="dev"):
|
|
# Should not raise — just logs error
|
|
milestone.execute(state)
|
|
|
|
|
|
def test_docker_build_with_progress():
|
|
state = _make_state(**{"init_db_build.run_build": "True"})
|
|
milestone = DockerBuildMilestone()
|
|
progress = MagicMock()
|
|
with patch("knoe.core.env._normalize_cluster_env", return_value="dev"):
|
|
milestone.execute(state, progress=progress)
|
|
progress.assert_called()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GitOpsMilestone
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_gitops_disabled():
|
|
state = _make_state(**{"init_cluster.gitops_enabled": "False"})
|
|
milestone = GitOpsMilestone()
|
|
milestone.execute(state)
|
|
assert state.config_data["GitOps"]["STATUS"] == "Skipped"
|
|
state.controller.run_script.assert_not_called()
|
|
|
|
|
|
def test_gitops_disabled_default():
|
|
state = _make_state()
|
|
milestone = GitOpsMilestone()
|
|
milestone.execute(state)
|
|
assert state.config_data["GitOps"]["STATUS"] == "Skipped"
|
|
|
|
|
|
def test_gitops_enabled_success():
|
|
state = _make_state(**{"init_cluster.gitops_enabled": "True"})
|
|
milestone = GitOpsMilestone()
|
|
with patch.object(milestone, "_get_script_env", return_value={"PROLE_MODE": "k3d"}):
|
|
milestone.execute(state)
|
|
assert state.config_data["GitOps"]["STATUS"] == "Deployed"
|
|
|
|
|
|
def test_gitops_enabled_with_progress():
|
|
state = _make_state(**{"init_cluster.gitops_enabled": "True"})
|
|
milestone = GitOpsMilestone()
|
|
progress = MagicMock()
|
|
with patch.object(milestone, "_get_script_env", return_value={"PROLE_MODE": "k3d"}):
|
|
milestone.execute(state, progress=progress)
|
|
progress.assert_called()
|
|
|
|
|
|
def test_gitops_enabled_failure_raises():
|
|
state = _make_state(**{"init_cluster.gitops_enabled": "True"})
|
|
state.controller.run_script.return_value = 1
|
|
milestone = GitOpsMilestone()
|
|
with patch.object(milestone, "_get_script_env", return_value={"PROLE_MODE": "k3d"}):
|
|
with pytest.raises(Exception, match="[Gg]itea"):
|
|
milestone.execute(state)
|
|
assert state.config_data["GitOps"]["STATUS"] == "Attempted"
|
|
|
|
|
|
def test_gitops_uses_cfg_path_when_exists(tmp_path):
|
|
state = _make_state(**{"init_cluster.gitops_enabled": "True"})
|
|
state.controller.project_root = tmp_path
|
|
(tmp_path / "conf").mkdir()
|
|
(tmp_path / "conf" / "prole.cfg").write_text("[General]\n")
|
|
milestone = GitOpsMilestone()
|
|
with patch.object(milestone, "_get_script_env", return_value={"PROLE_MODE": ""}):
|
|
milestone.execute(state)
|
|
_, kwargs = state.controller.run_script.call_args
|
|
assert "-c" in kwargs.get("args", [])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# KerberosMilestone
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_kerberos_disabled():
|
|
state = _make_state(**{"kerberos_config.enabled": "False"})
|
|
milestone = KerberosMilestone()
|
|
milestone.execute(state)
|
|
assert state.config_data["Kerberos Authentication"]["STATUS"] == "Skipped"
|
|
state.controller.run_script.assert_not_called()
|
|
|
|
|
|
def test_kerberos_disabled_default():
|
|
state = _make_state()
|
|
milestone = KerberosMilestone()
|
|
milestone.execute(state)
|
|
assert state.config_data["Kerberos Authentication"]["STATUS"] == "Skipped"
|
|
|
|
|
|
def test_kerberos_disabled_with_progress():
|
|
state = _make_state(**{"kerberos_config.enabled": "False"})
|
|
milestone = KerberosMilestone()
|
|
progress = MagicMock()
|
|
milestone.execute(state, progress=progress)
|
|
progress.assert_called_with("Kerberos disabled; skipping", 1.0)
|
|
|
|
|
|
def test_kerberos_enabled_success():
|
|
state = _make_state(**{"kerberos_config.enabled": "True"})
|
|
milestone = KerberosMilestone()
|
|
with patch.object(milestone, "_get_script_env", return_value={"PROLE_MODE": "k3d"}):
|
|
milestone.execute(state)
|
|
assert state.config_data["Kerberos Authentication"]["STATUS"] == "Initialized"
|
|
state.controller.run_script.assert_called_once()
|
|
_, kwargs = state.controller.run_script.call_args
|
|
assert "initialize" in kwargs.get("args", [])
|
|
|
|
|
|
def test_kerberos_enabled_with_progress():
|
|
state = _make_state(**{"kerberos_config.enabled": "True"})
|
|
milestone = KerberosMilestone()
|
|
progress = MagicMock()
|
|
with patch.object(milestone, "_get_script_env", return_value={"PROLE_MODE": ""}):
|
|
milestone.execute(state, progress=progress)
|
|
progress.assert_called()
|
|
|
|
|
|
def test_kerberos_enabled_failure_raises():
|
|
state = _make_state(**{"kerberos_config.enabled": "True"})
|
|
state.controller.run_script.return_value = 1
|
|
milestone = KerberosMilestone()
|
|
with patch.object(milestone, "_get_script_env", return_value={"PROLE_MODE": ""}):
|
|
with pytest.raises(Exception, match="[Kk]erberos"):
|
|
milestone.execute(state)
|
|
assert state.config_data["Kerberos Authentication"]["STATUS"] == "Attempted"
|
|
|
|
|
|
def test_kerberos_enabled_mode_in_args():
|
|
state = _make_state(**{"kerberos_config.enabled": "True"})
|
|
milestone = KerberosMilestone()
|
|
with patch.object(milestone, "_get_script_env", return_value={"PROLE_MODE": "k3d"}):
|
|
milestone.execute(state)
|
|
_, kwargs = state.controller.run_script.call_args
|
|
assert "--mode" in kwargs.get("args", [])
|
|
assert "k3d" in kwargs.get("args", [])
|
|
|
|
|
|
def test_kerberos_enabled_no_mode():
|
|
state = _make_state(**{"kerberos_config.enabled": "True"})
|
|
milestone = KerberosMilestone()
|
|
with patch.object(milestone, "_get_script_env", return_value={"PROLE_MODE": ""}):
|
|
milestone.execute(state)
|
|
_, kwargs = state.controller.run_script.call_args
|
|
assert "--mode" not in kwargs.get("args", [])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Supabase milestones
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_supabase_image_preload_disabled():
|
|
state = _make_state(**{"init_cluster.supabase_enabled": "False"})
|
|
milestone = SupabaseImagePreloadMilestone()
|
|
milestone.execute(state)
|
|
assert state.config_data["Supabase"]["IMAGES_STATUS"] == "Skipped"
|
|
|
|
|
|
def test_supabase_image_preload_success(tmp_path):
|
|
state = _make_state(**{
|
|
"init_cluster.supabase_enabled": "True",
|
|
"init_cluster.cluster_env": "dev",
|
|
})
|
|
state.controller.project_root = tmp_path
|
|
supa_dir = tmp_path / "supabase"
|
|
supa_dir.mkdir()
|
|
(supa_dir / "deploy.sh").write_text("#!/bin/bash\nexit 0\n")
|
|
mock_result = MagicMock()
|
|
mock_result.returncode = 0
|
|
captured_cmd = []
|
|
|
|
def capture_run(cmd, **kwargs):
|
|
captured_cmd.extend(cmd)
|
|
return mock_result
|
|
|
|
milestone = SupabaseImagePreloadMilestone()
|
|
with patch.object(milestone, "_get_script_env", return_value={}), \
|
|
patch("knoe.core.milestones.subprocess.run", side_effect=capture_run), \
|
|
patch("knoe.core.env._normalize_cluster_env", return_value="dev"), \
|
|
patch("knoe.core.env._parse_bool", return_value=False):
|
|
milestone.execute(state)
|
|
|
|
assert state.config_data["Supabase"]["IMAGES_STATUS"] == "Prepared"
|
|
assert state.inputs["supabase.images_preloaded"] == "True"
|
|
assert "--prefetch-images-only" in captured_cmd
|
|
assert "k3d" in captured_cmd
|
|
|
|
|
|
def test_supabase_image_preload_failure_raises(tmp_path):
|
|
state = _make_state(**{
|
|
"init_cluster.supabase_enabled": "True",
|
|
"init_cluster.cluster_env": "dev",
|
|
})
|
|
state.controller.project_root = tmp_path
|
|
supa_dir = tmp_path / "supabase"
|
|
supa_dir.mkdir()
|
|
(supa_dir / "deploy.sh").write_text("#!/bin/bash\nexit 1\n")
|
|
mock_result = MagicMock()
|
|
mock_result.returncode = 1
|
|
milestone = SupabaseImagePreloadMilestone()
|
|
|
|
with patch.object(milestone, "_get_script_env", return_value={}), \
|
|
patch("knoe.core.milestones.subprocess.run", return_value=mock_result), \
|
|
patch("knoe.core.env._normalize_cluster_env", return_value="dev"), \
|
|
patch("knoe.core.env._parse_bool", return_value=False):
|
|
with pytest.raises(Exception, match="[Pp]reload"):
|
|
milestone.execute(state)
|
|
|
|
assert state.config_data["Supabase"]["IMAGES_STATUS"] == "Attempted"
|
|
|
|
|
|
def test_supabase_image_preload_skips_for_k8s_mode(tmp_path):
|
|
state = _make_state(**{
|
|
"init_cluster.supabase_enabled": "True",
|
|
"init_cluster.cluster_env": "prod",
|
|
})
|
|
state.controller.project_root = tmp_path
|
|
supa_dir = tmp_path / "supabase"
|
|
supa_dir.mkdir()
|
|
(supa_dir / "deploy.sh").write_text("#!/bin/bash\nexit 0\n")
|
|
|
|
milestone = SupabaseImagePreloadMilestone()
|
|
with patch.object(milestone, "_get_script_env", return_value={}), \
|
|
patch("knoe.core.milestones.subprocess.run") as mock_run, \
|
|
patch("knoe.core.env._normalize_cluster_env", return_value="prod"):
|
|
milestone.execute(state)
|
|
|
|
assert state.config_data["Supabase"]["IMAGES_STATUS"] == "Skipped"
|
|
mock_run.assert_not_called()
|
|
|
|
def test_supabase_disabled():
|
|
state = _make_state(**{"init_cluster.supabase_enabled": "False"})
|
|
milestone = SupabaseMilestone()
|
|
milestone.execute(state)
|
|
assert state.config_data["Supabase"]["STATUS"] == "Skipped"
|
|
|
|
|
|
def test_supabase_disabled_default():
|
|
state = _make_state()
|
|
milestone = SupabaseMilestone()
|
|
milestone.execute(state)
|
|
assert state.config_data["Supabase"]["STATUS"] == "Skipped"
|
|
|
|
|
|
def test_supabase_disabled_with_progress():
|
|
state = _make_state(**{"init_cluster.supabase_enabled": "False"})
|
|
milestone = SupabaseMilestone()
|
|
progress = MagicMock()
|
|
milestone.execute(state, progress=progress)
|
|
progress.assert_called_with("Supabase disabled; skipping", 1.0)
|
|
|
|
|
|
def test_supabase_enabled_script_missing(tmp_path):
|
|
state = _make_state(**{"init_cluster.supabase_enabled": "True"})
|
|
state.controller.project_root = tmp_path
|
|
milestone = SupabaseMilestone()
|
|
with patch.object(milestone, "_get_script_env", return_value={}):
|
|
milestone.execute(state)
|
|
# deploy.sh doesn't exist → status Attempted, no exception
|
|
assert state.config_data["Supabase"]["STATUS"] == "Attempted"
|
|
|
|
|
|
def test_supabase_enabled_success(tmp_path):
|
|
state = _make_state(**{
|
|
"init_cluster.supabase_enabled": "True",
|
|
"init_cluster.cluster_env": "dev",
|
|
})
|
|
state.controller.project_root = tmp_path
|
|
supa_dir = tmp_path / "supabase"
|
|
supa_dir.mkdir()
|
|
(supa_dir / "deploy.sh").write_text("#!/bin/bash\nexit 0\n")
|
|
mock_result = MagicMock()
|
|
mock_result.returncode = 0
|
|
milestone = SupabaseMilestone()
|
|
with patch.object(milestone, "_get_script_env", return_value={}), \
|
|
patch("knoe.core.milestones.subprocess.run", return_value=mock_result), \
|
|
patch("knoe.core.env._normalize_cluster_env", return_value="dev"), \
|
|
patch("knoe.core.env._parse_bool", return_value=False):
|
|
milestone.execute(state)
|
|
assert state.config_data["Supabase"]["STATUS"] == "Deployed"
|
|
|
|
|
|
def test_supabase_enabled_failure(tmp_path):
|
|
state = _make_state(**{
|
|
"init_cluster.supabase_enabled": "True",
|
|
"init_cluster.cluster_env": "dev",
|
|
})
|
|
state.controller.project_root = tmp_path
|
|
supa_dir = tmp_path / "supabase"
|
|
supa_dir.mkdir()
|
|
(supa_dir / "deploy.sh").write_text("#!/bin/bash\nexit 1\n")
|
|
mock_result = MagicMock()
|
|
mock_result.returncode = 1
|
|
milestone = SupabaseMilestone()
|
|
with patch.object(milestone, "_get_script_env", return_value={}), \
|
|
patch("knoe.core.milestones.subprocess.run", return_value=mock_result), \
|
|
patch("knoe.core.env._normalize_cluster_env", return_value="dev"), \
|
|
patch("knoe.core.env._parse_bool", return_value=False):
|
|
with pytest.raises(Exception, match="[Ss]upabase"):
|
|
milestone.execute(state)
|
|
assert state.config_data["Supabase"]["STATUS"] == "Attempted"
|
|
|
|
|
|
def test_supabase_enabled_with_progress(tmp_path):
|
|
state = _make_state(**{
|
|
"init_cluster.supabase_enabled": "True",
|
|
"init_cluster.cluster_env": "dev",
|
|
})
|
|
state.controller.project_root = tmp_path
|
|
supa_dir = tmp_path / "supabase"
|
|
supa_dir.mkdir()
|
|
(supa_dir / "deploy.sh").write_text("#!/bin/bash\nexit 0\n")
|
|
mock_result = MagicMock()
|
|
mock_result.returncode = 0
|
|
milestone = SupabaseMilestone()
|
|
progress = MagicMock()
|
|
with patch.object(milestone, "_get_script_env", return_value={}), \
|
|
patch("knoe.core.milestones.subprocess.run", return_value=mock_result), \
|
|
patch("knoe.core.env._normalize_cluster_env", return_value="dev"), \
|
|
patch("knoe.core.env._parse_bool", return_value=False):
|
|
milestone.execute(state, progress=progress)
|
|
progress.assert_called()
|
|
|
|
|
|
def test_supabase_deploy_skips_prefetch_when_preloaded(tmp_path):
|
|
state = _make_state(**{
|
|
"init_cluster.supabase_enabled": "True",
|
|
"init_cluster.cluster_env": "dev",
|
|
"supabase.images_preloaded": "True",
|
|
})
|
|
state.controller.project_root = tmp_path
|
|
supa_dir = tmp_path / "supabase"
|
|
supa_dir.mkdir()
|
|
(supa_dir / "deploy.sh").write_text("#!/bin/bash\nexit 0\n")
|
|
mock_result = MagicMock()
|
|
mock_result.returncode = 0
|
|
captured_cmd = []
|
|
|
|
def capture_run(cmd, **kwargs):
|
|
captured_cmd.extend(cmd)
|
|
return mock_result
|
|
|
|
milestone = SupabaseMilestone()
|
|
with patch.object(milestone, "_get_script_env", return_value={}), \
|
|
patch("knoe.core.milestones.subprocess.run", side_effect=capture_run), \
|
|
patch("knoe.core.env._normalize_cluster_env", return_value="dev"):
|
|
milestone.execute(state)
|
|
|
|
assert "--skip-prefetch" in captured_cmd
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# DependenciesMilestone — error paths not covered by existing tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_dependencies_auto_install_disabled():
|
|
"""When auto_install=False, set status Missing and return without installing."""
|
|
state = _make_state(**{"dependencies.auto_install_missing": "False"})
|
|
milestone = DependenciesMilestone()
|
|
dep = {"id": "python", "name": "python", "install_cmd": "apt install python3"}
|
|
with patch("knoe.config.get_dep_info", return_value=(False, None, None)), \
|
|
patch("knoe.config.get_required_dependencies", return_value=[dep]), \
|
|
patch("knoe.config.get_required_dependency_ids", return_value={"python"}), \
|
|
pytest.raises(RuntimeError, match="auto-install disabled"):
|
|
milestone.execute(state)
|
|
assert state.config_data.get("Dependencies", {}).get("STATUS") == "Missing"
|
|
|
|
|
|
def test_dependencies_no_install_cmd():
|
|
"""A missing dep with no install_cmd should log error and be skipped."""
|
|
state = _make_state(**{"dependencies.auto_install_missing": "True"})
|
|
milestone = DependenciesMilestone()
|
|
# First call: dep missing; second call (re-verify): still missing
|
|
with patch("knoe.config.get_dep_info", return_value=(False, None, None)), \
|
|
patch("knoe.config.get_required_dependencies", return_value=[{"id": "mytool", "name": "MyTool"}]), \
|
|
patch("knoe.config.get_required_dependency_ids", return_value={"mytool"}), \
|
|
pytest.raises(RuntimeError, match="Required dependencies still missing"):
|
|
milestone.execute(state)
|
|
assert state.config_data.get("Dependencies", {}).get("STATUS") == "Missing"
|
|
|
|
|
|
def test_dependencies_install_skipped_by_config():
|
|
"""Dep with install disabled in state.inputs should be skipped."""
|
|
state = _make_state(**{
|
|
"dependencies.auto_install_missing": "True",
|
|
"dependencies.python.install": "False",
|
|
})
|
|
milestone = DependenciesMilestone()
|
|
dep = {"id": "python", "name": "python", "install_cmd": "apt install python3"}
|
|
with patch("knoe.config.get_dep_info", return_value=(False, None, None)), \
|
|
patch("knoe.config.get_required_dependencies", return_value=[dep]), \
|
|
patch("knoe.config.get_required_dependency_ids", return_value={"python"}), \
|
|
pytest.raises(RuntimeError, match="Required dependencies still missing"):
|
|
milestone.execute(state)
|
|
assert state.config_data.get("Dependencies", {}).get("STATUS") == "Missing"
|
|
|
|
|
|
def test_dependencies_install_fails():
|
|
"""When the install command returns non-zero, still mark Missing."""
|
|
state = _make_state(**{"dependencies.auto_install_missing": "True"})
|
|
milestone = DependenciesMilestone()
|
|
dep = {"id": "python", "name": "python", "install_cmd": "apt install python3"}
|
|
with patch("knoe.config.get_dep_info", return_value=(False, None, None)), \
|
|
patch("knoe.config.get_required_dependencies", return_value=[dep]), \
|
|
patch("knoe.config.get_required_dependency_ids", return_value={"python"}), \
|
|
patch.object(milestone, "_run_cmd", return_value=1), \
|
|
pytest.raises(RuntimeError, match="Required dependencies still missing"):
|
|
milestone.execute(state)
|
|
assert state.config_data.get("Dependencies", {}).get("STATUS") == "Missing"
|
|
|
|
|
|
def test_dependencies_install_succeeds_all():
|
|
"""After install, re-verify passes → status 'All installed'."""
|
|
state = _make_state(**{"dependencies.auto_install_missing": "True"})
|
|
milestone = DependenciesMilestone()
|
|
dep = {"id": "python", "name": "python", "install_cmd": "apt install python3"}
|
|
call_count = {"n": 0}
|
|
|
|
def _dep_info(d):
|
|
call_count["n"] += 1
|
|
# First call (initial check): missing. Re-verify calls: installed.
|
|
if call_count["n"] <= len([dep]):
|
|
return (False, None, None)
|
|
return (True, "/usr/bin/python3", "3.10")
|
|
|
|
with patch("knoe.config.get_dep_info", side_effect=_dep_info), \
|
|
patch("knoe.config.get_required_dependencies", return_value=[dep]), \
|
|
patch("knoe.config.get_required_dependency_ids", return_value={"python"}), \
|
|
patch.object(milestone, "_run_cmd", return_value=0):
|
|
milestone.execute(state)
|
|
assert state.config_data.get("Dependencies", {}).get("STATUS") == "All installed"
|
|
|
|
|
|
def test_dependencies_k8s_fails_without_gcloud_auth_session():
|
|
state = _make_state(
|
|
**{
|
|
"dependencies.auto_install_missing": "True",
|
|
"init_cluster.cluster_env": "prod",
|
|
}
|
|
)
|
|
milestone = DependenciesMilestone()
|
|
dep = {"id": "gcloud", "name": "gcloud", "install_cmd": "apt install google-cloud-cli"}
|
|
|
|
no_token = MagicMock(returncode=1, stdout="", stderr="not logged in")
|
|
|
|
def _unexpected_gcloud_plain(*_a, **_k):
|
|
raise AssertionError("_gcloud_plain should not be called for auth preflight")
|
|
|
|
with patch("knoe.config.get_dep_info", return_value=(True, "/usr/bin/gcloud", "1.0")), \
|
|
patch("knoe.config.get_required_dependencies", return_value=[dep]), \
|
|
patch("knoe.config.get_required_dependency_ids", return_value={"gcloud"}), \
|
|
patch("knoe.config._augment_env_for_dependency_backend", return_value={"PATH": "x"}), \
|
|
patch("knoe.config._gcloud_plain", side_effect=_unexpected_gcloud_plain), \
|
|
patch("knoe.core.milestones.subprocess.run", return_value=no_token), \
|
|
patch("knoe.core.milestones.sys.stdin.isatty", return_value=False), \
|
|
pytest.raises(RuntimeError, match="gcloud authentication is required"):
|
|
milestone.execute(state)
|
|
|
|
assert state.config_data.get("Dependencies", {}).get("STATUS") == "Missing"
|
|
|
|
|
|
def test_dependencies_k8s_interactive_login_recovers_auth_session():
|
|
state = _make_state(
|
|
**{
|
|
"dependencies.auto_install_missing": "True",
|
|
"init_cluster.cluster_env": "prod",
|
|
}
|
|
)
|
|
milestone = DependenciesMilestone()
|
|
dep = {"id": "gcloud", "name": "gcloud", "install_cmd": "apt install google-cloud-cli"}
|
|
|
|
no_token = MagicMock(returncode=1, stdout="", stderr="not logged in")
|
|
with_token = MagicMock(returncode=0, stdout="tok123\n", stderr="")
|
|
|
|
def _unexpected_gcloud_plain(*_a, **_k):
|
|
raise AssertionError("_gcloud_plain should not be called for auth preflight")
|
|
|
|
with patch("knoe.config.get_dep_info", return_value=(True, "/usr/bin/gcloud", "1.0")), \
|
|
patch("knoe.config.get_required_dependencies", return_value=[dep]), \
|
|
patch("knoe.config.get_required_dependency_ids", return_value={"gcloud"}), \
|
|
patch("knoe.config._augment_env_for_dependency_backend", return_value={"PATH": "x"}), \
|
|
patch("knoe.config._gcloud_plain", side_effect=_unexpected_gcloud_plain), \
|
|
patch("knoe.core.milestones.subprocess.run", side_effect=[no_token, with_token]), \
|
|
patch("knoe.core.milestones.sys.stdin.isatty", return_value=True), \
|
|
patch.object(milestone, "_ensure_gke_auth_plugin_for_k8s", return_value=True), \
|
|
patch.object(milestone, "_ensure_gke_contexts_for_k8s", return_value=True), \
|
|
patch.object(milestone, "_run_cmd", return_value=0) as run_cmd_mock:
|
|
milestone.execute(state)
|
|
|
|
assert state.config_data.get("Dependencies", {}).get("STATUS") == "All installed"
|
|
run_cmd_mock.assert_called_once()
|
|
|
|
|
|
def test_dependencies_k8s_acquires_missing_contexts_via_gcloud():
|
|
state = _make_state(
|
|
**{
|
|
"dependencies.auto_install_missing": "True",
|
|
"init_cluster.cluster_env": "prod",
|
|
"init_cluster.project_id": "proj-1",
|
|
"init_cluster.app_cluster_name": "app-cluster",
|
|
"init_cluster.app_cluster_region": "us-west3",
|
|
"init_cluster.app_cluster_kubecontext": "ctx-app",
|
|
"init_cluster.db_cluster_name": "db-cluster",
|
|
"init_cluster.db_cluster_region": "us-west3",
|
|
"init_cluster.db_cluster_kubecontext": "ctx-db",
|
|
}
|
|
)
|
|
milestone = DependenciesMilestone()
|
|
dep = {"id": "gcloud", "name": "gcloud", "install_cmd": "apt install google-cloud-cli"}
|
|
|
|
token_ok = MagicMock(returncode=0, stdout="tok123\n", stderr="")
|
|
contexts_before = MagicMock(returncode=0, stdout="", stderr="")
|
|
contexts_after = MagicMock(returncode=0, stdout="ctx-app\nctx-db\n", stderr="")
|
|
|
|
with patch("knoe.config.get_dep_info", return_value=(True, "/usr/bin/gcloud", "1.0")), \
|
|
patch("knoe.config.get_required_dependencies", return_value=[dep]), \
|
|
patch("knoe.config.get_required_dependency_ids", return_value={"gcloud"}), \
|
|
patch("knoe.config._augment_env_for_dependency_backend", return_value={"PATH": "x"}), \
|
|
patch("knoe.core.milestones.shutil.which", return_value="/usr/bin/gcloud"), \
|
|
patch(
|
|
"knoe.core.milestones.subprocess.run",
|
|
side_effect=[token_ok, contexts_before, contexts_after],
|
|
), \
|
|
patch.object(milestone, "_run_cmd", return_value=0) as run_cmd_mock:
|
|
milestone.execute(state)
|
|
|
|
assert state.config_data.get("Dependencies", {}).get("STATUS") == "All installed"
|
|
assert run_cmd_mock.call_count == 2
|
|
|
|
|
|
def test_dependencies_k8s_fails_when_auth_plugin_missing():
|
|
state = _make_state(
|
|
**{
|
|
"dependencies.auto_install_missing": "True",
|
|
"init_cluster.cluster_env": "prod",
|
|
}
|
|
)
|
|
milestone = DependenciesMilestone()
|
|
dep = {"id": "gcloud", "name": "gcloud", "install_cmd": "apt install google-cloud-cli"}
|
|
|
|
token_ok = MagicMock(returncode=0, stdout="tok123\n", stderr="")
|
|
|
|
with patch("knoe.config.get_dep_info", return_value=(True, "/usr/bin/gcloud", "1.0")), \
|
|
patch("knoe.config.get_required_dependencies", return_value=[dep]), \
|
|
patch("knoe.config.get_required_dependency_ids", return_value={"gcloud"}), \
|
|
patch("knoe.config._augment_env_for_dependency_backend", return_value={"PATH": "x"}), \
|
|
patch("knoe.core.milestones.subprocess.run", return_value=token_ok), \
|
|
patch("knoe.core.milestones.shutil.which", return_value=None), \
|
|
pytest.raises(RuntimeError, match="gke-gcloud-auth-plugin is required"):
|
|
milestone.execute(state)
|
|
|
|
assert state.config_data.get("Dependencies", {}).get("STATUS") == "Missing"
|
|
|
|
|
|
def test_dependencies_k8s_fails_when_context_missing_and_gcloud_unavailable():
|
|
state = _make_state(
|
|
**{
|
|
"dependencies.auto_install_missing": "True",
|
|
"init_cluster.cluster_env": "prod",
|
|
"init_cluster.app_cluster_name": "app-cluster",
|
|
"init_cluster.app_cluster_region": "us-west3",
|
|
"init_cluster.app_cluster_kubecontext": "ctx-app",
|
|
}
|
|
)
|
|
milestone = DependenciesMilestone()
|
|
dep = {"id": "gcloud", "name": "gcloud", "install_cmd": "apt install google-cloud-cli"}
|
|
|
|
token_ok = MagicMock(returncode=0, stdout="tok123\n", stderr="")
|
|
contexts_before = MagicMock(returncode=0, stdout="", stderr="")
|
|
|
|
with patch("knoe.config.get_dep_info", return_value=(True, "/usr/bin/gcloud", "1.0")), \
|
|
patch("knoe.config.get_required_dependencies", return_value=[dep]), \
|
|
patch("knoe.config.get_required_dependency_ids", return_value={"gcloud"}), \
|
|
patch("knoe.config._augment_env_for_dependency_backend", return_value={"PATH": "x"}), \
|
|
patch(
|
|
"knoe.core.milestones.shutil.which",
|
|
side_effect=lambda binary: "/usr/bin/gke-gcloud-auth-plugin"
|
|
if binary == "gke-gcloud-auth-plugin"
|
|
else None,
|
|
), \
|
|
patch(
|
|
"knoe.core.milestones.subprocess.run",
|
|
side_effect=[token_ok, contexts_before],
|
|
), \
|
|
pytest.raises(RuntimeError, match="gcloud is missing while required GKE contexts are absent"):
|
|
milestone.execute(state)
|
|
|
|
assert state.config_data.get("Dependencies", {}).get("STATUS") == "Missing"
|
|
|
|
|
|
def test_dependencies_k8s_uses_user_kubeconfig_when_k3s_kubeconfig_is_set(tmp_path):
|
|
state = _make_state(
|
|
**{
|
|
"init_cluster.cluster_env": "prod",
|
|
"init_cluster.app_cluster_name": "app-cluster",
|
|
"init_cluster.app_cluster_region": "us-west3",
|
|
"init_cluster.app_cluster_kubecontext": "ctx-app",
|
|
}
|
|
)
|
|
milestone = DependenciesMilestone()
|
|
|
|
contexts_before = MagicMock(returncode=0, stdout="", stderr="")
|
|
contexts_after = MagicMock(returncode=0, stdout="ctx-app\n", stderr="")
|
|
|
|
with patch("knoe.config.get_required_dependency_ids", return_value={"gcloud"}), \
|
|
patch("knoe.config._augment_env_for_dependency_backend", return_value={"PATH": "x"}), \
|
|
patch("knoe.core.milestones.shutil.which", return_value="/usr/bin/gcloud"), \
|
|
patch("knoe.core.milestones.Path.home", return_value=tmp_path), \
|
|
patch.dict("knoe.core.milestones.os.environ", {"KUBECONFIG": "/etc/rancher/k3s/k3s.yaml"}, clear=False), \
|
|
patch(
|
|
"knoe.core.milestones.subprocess.run",
|
|
side_effect=[contexts_before, contexts_after],
|
|
), \
|
|
patch.object(milestone, "_run_cmd", return_value=0) as run_cmd_mock:
|
|
assert milestone._ensure_gke_contexts_for_k8s(state) is True
|
|
|
|
expected_target = str(tmp_path / ".kube" / "config")
|
|
assert run_cmd_mock.call_count == 1
|
|
assert run_cmd_mock.call_args.kwargs["env"]["KUBECONFIG"] == expected_target
|
|
|
|
|
|
def test_dependencies_k8s_fails_with_clear_message_when_kubeconfig_target_not_writable(tmp_path):
|
|
state = _make_state(
|
|
**{
|
|
"dependencies.auto_install_missing": "True",
|
|
"init_cluster.cluster_env": "prod",
|
|
"init_cluster.app_cluster_name": "app-cluster",
|
|
"init_cluster.app_cluster_region": "us-west3",
|
|
"init_cluster.app_cluster_kubecontext": "ctx-app",
|
|
}
|
|
)
|
|
milestone = DependenciesMilestone()
|
|
dep = {"id": "gcloud", "name": "gcloud", "install_cmd": "apt install google-cloud-cli"}
|
|
|
|
token_ok = MagicMock(returncode=0, stdout="tok123\n", stderr="")
|
|
|
|
with patch("knoe.config.get_dep_info", return_value=(True, "/usr/bin/gcloud", "1.0")), \
|
|
patch("knoe.config.get_required_dependencies", return_value=[dep]), \
|
|
patch("knoe.config.get_required_dependency_ids", return_value={"gcloud"}), \
|
|
patch("knoe.config._augment_env_for_dependency_backend", return_value={"PATH": "x"}), \
|
|
patch("knoe.core.milestones.subprocess.run", return_value=token_ok), \
|
|
patch("knoe.core.milestones.shutil.which", return_value="/usr/bin/gke-gcloud-auth-plugin"), \
|
|
patch("knoe.core.milestones.Path.home", return_value=tmp_path), \
|
|
patch.object(milestone, "_is_kubeconfig_target_writable", return_value=False), \
|
|
pytest.raises(RuntimeError, match="kubeconfig target '.+/.kube/config' is not writable"):
|
|
milestone.execute(state)
|
|
|
|
assert state.config_data.get("Dependencies", {}).get("STATUS") == "Missing"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# InitializationScriptsMilestone._regenerate_port_mapping_cfg
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_regenerate_port_mapping_cfg_k3d_mode(tmp_path):
|
|
state = _make_state()
|
|
state.config_data["Port Forwards"] = {
|
|
"PORT_FORWARD_K3D_MAPPING_argocd": (
|
|
"id=argocd;namespace=argocd;target=svc/argocd-server;"
|
|
"hostPort=8080;servicePort=80;address=127.0.0.1"
|
|
),
|
|
}
|
|
milestone = InitializationScriptsMilestone()
|
|
with patch("knoe.core.milestones.inst_config.PROJECT_ROOT", tmp_path):
|
|
(tmp_path / "conf").mkdir(exist_ok=True)
|
|
milestone._regenerate_port_mapping_cfg(state, "k3d")
|
|
mapping = (tmp_path / "conf" / "port-mapping.cfg").read_text()
|
|
assert "argocd: local=8080" in mapping
|
|
|
|
|
|
def test_regenerate_port_mapping_cfg_k3s_mode(tmp_path):
|
|
state = _make_state()
|
|
state.config_data["Port Forwards"] = {
|
|
"PORT_FORWARD_K3S_MAPPING_openbao": (
|
|
"id=openbao;namespace=default;target=svc/openbao;"
|
|
"hostPort=8200;servicePort=8200;address=127.0.0.1"
|
|
),
|
|
}
|
|
milestone = InitializationScriptsMilestone()
|
|
with patch("knoe.core.milestones.inst_config.PROJECT_ROOT", tmp_path):
|
|
(tmp_path / "conf").mkdir(exist_ok=True)
|
|
milestone._regenerate_port_mapping_cfg(state, "k3s")
|
|
mapping = (tmp_path / "conf" / "port-mapping.cfg").read_text()
|
|
assert "openbao: local=8200" in mapping
|
|
|
|
|
|
def test_regenerate_port_mapping_cfg_skips_wrong_prefix(tmp_path):
|
|
state = _make_state()
|
|
state.config_data["Port Forwards"] = {
|
|
"PORT_FORWARD_K3S_MAPPING_x": (
|
|
"id=x;namespace=default;target=svc/x;hostPort=9000;servicePort=9000;address=127.0.0.1"
|
|
),
|
|
}
|
|
milestone = InitializationScriptsMilestone()
|
|
with patch("knoe.core.milestones.inst_config.PROJECT_ROOT", tmp_path):
|
|
(tmp_path / "conf").mkdir(exist_ok=True)
|
|
# k3d mode should skip k3s-prefixed entries
|
|
milestone._regenerate_port_mapping_cfg(state, "k3d")
|
|
mapping = (tmp_path / "conf" / "port-mapping.cfg").read_text()
|
|
assert "x: local=" not in mapping
|
|
|
|
|
|
def test_regenerate_port_mapping_cfg_resolves_namespace(tmp_path):
|
|
state = _make_state(**{"init_password.db_namespace": "myns"})
|
|
state.config_data["Port Forwards"] = {
|
|
"PORT_FORWARD_K3D_MAPPING_svc": (
|
|
"id=svc;namespace=${NAMESPACE};target=svc/mysvc;"
|
|
"hostPort=5000;servicePort=5000;address=127.0.0.1"
|
|
),
|
|
}
|
|
milestone = InitializationScriptsMilestone()
|
|
with patch("knoe.core.milestones.inst_config.PROJECT_ROOT", tmp_path):
|
|
(tmp_path / "conf").mkdir(exist_ok=True)
|
|
milestone._regenerate_port_mapping_cfg(state, "k3d")
|
|
mapping = (tmp_path / "conf" / "port-mapping.cfg").read_text()
|
|
assert "ns=myns" in mapping
|
|
|
|
|
|
def test_regenerate_port_mapping_cfg_skips_unresolvable_port(tmp_path):
|
|
state = _make_state()
|
|
state.config_data["Port Forwards"] = {
|
|
"PORT_FORWARD_K3D_MAPPING_bad": (
|
|
"id=bad;namespace=default;target=svc/bad;"
|
|
"hostPort=${UNRESOLVED_PORT};servicePort=9999;address=127.0.0.1"
|
|
),
|
|
}
|
|
milestone = InitializationScriptsMilestone()
|
|
with patch("knoe.core.milestones.inst_config.PROJECT_ROOT", tmp_path):
|
|
(tmp_path / "conf").mkdir(exist_ok=True)
|
|
milestone._regenerate_port_mapping_cfg(state, "k3d")
|
|
mapping = (tmp_path / "conf" / "port-mapping.cfg").read_text()
|
|
assert "bad: local=" not in mapping
|
|
|
|
|
|
def test_regenerate_port_mapping_cfg_empty_section(tmp_path):
|
|
state = _make_state()
|
|
milestone = InitializationScriptsMilestone()
|
|
with patch("knoe.core.milestones.inst_config.PROJECT_ROOT", tmp_path):
|
|
(tmp_path / "conf").mkdir(exist_ok=True)
|
|
milestone._regenerate_port_mapping_cfg(state, "k3d")
|
|
mapping = (tmp_path / "conf" / "port-mapping.cfg").read_text()
|
|
assert "# Port mappings" in mapping
|
|
|
|
|
|
def test_regenerate_port_mapping_cfg_write_error(tmp_path):
|
|
"""Write failure should be caught and logged, not raised."""
|
|
state = _make_state()
|
|
milestone = InitializationScriptsMilestone()
|
|
with patch("knoe.core.milestones.inst_config.PROJECT_ROOT", tmp_path), \
|
|
patch("pathlib.Path.write_text", side_effect=OSError("disk full")):
|
|
(tmp_path / "conf").mkdir(exist_ok=True)
|
|
# Should not raise
|
|
milestone._regenerate_port_mapping_cfg(state, "k3d")
|
|
|
|
|
|
def test_supabase_k8s_mode_for_prod(tmp_path):
|
|
state = _make_state(**{
|
|
"init_cluster.supabase_enabled": "True",
|
|
"init_cluster.cluster_env": "prod",
|
|
})
|
|
state.controller.project_root = tmp_path
|
|
supa_dir = tmp_path / "supabase"
|
|
supa_dir.mkdir()
|
|
(supa_dir / "deploy.sh").write_text("#!/bin/bash\nexit 0\n")
|
|
mock_result = MagicMock()
|
|
mock_result.returncode = 0
|
|
captured_cmd = []
|
|
def capture_run(cmd, **kwargs):
|
|
captured_cmd.extend(cmd)
|
|
return mock_result
|
|
milestone = SupabaseMilestone()
|
|
with patch.object(milestone, "_get_script_env", return_value={}), \
|
|
patch("knoe.core.milestones.subprocess.run", side_effect=capture_run), \
|
|
patch("knoe.core.env._normalize_cluster_env", return_value="prod"), \
|
|
patch("knoe.core.env._parse_bool", return_value=False):
|
|
milestone.execute(state)
|
|
assert "k8s" in captured_cmd
|