chrisfu/prole
1
0
Fork 0
mirror of https://github.com/dredx/prole.git synced 2026-09-23 10:13:58 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity
5618b662dd
prole/tests/installer/test_milestones_extended.py
chrisfu 5618b662dd Remove prole-db-manager; simplify deployment via prole-authority; fix pg18 downgrade & cluster name 
Summary:
Removed the prole-db-manager microservice and simplified deployment to use
prole-authority as the internal management and authorization point. Fixed two
blocking bugs that prevented silent install from completing on knoe-dev-cluster.

Removed: prole-db-manager
- Deleted db-manager-deployment.yaml and db-manager-service.yaml from opentofu manifests
- Deleted src/db-manager/ (Dockerfile, server.js, package.json, tests)
- Removed prole-db-manager port-forward mapping from installer/core/env.py
- Removed init_db_manager.sh from Initialization Scripts (milestones.py, actions.py)
- Removed init_certmgr.sh and init_db_manager.sh tabs from services screen (services.py)
- Removed live k8s Deployment/Service from knoe-dev-cluster

Fixed: PostgreSQL version downgrade error (pg17 -> pg18)
- Created conf/postgresql/.version with value 18
- Updated k8s/prole/prole-db.yaml and prole-db-recovery.yaml.tpl imageName to prole-db:18-089
- Fixed _init_database_options_state() to restore saved version_type from prole.cfg
  so db_version_type defaults to v18 (pg18) instead of silently reverting to pg17
- Added database_options.* keys to _collect_input_snapshot() in cfg.py so
  distribution, version_type, and all extension toggles persist to prole.cfg

Fixed: Cluster name inconsistency
- Removed stale prole-dev-cluster references; all scripts now use knoe-dev-cluster
- Added knoe-dev-cluster to mode-detection case in etc/prole_cfg.sh

Config: conf/prole.cfg
- Set kerberos_config.enabled = False, KERBEROS_AUTO_ENABLED = False
- Added database_options.distribution = percona, version_type = v18
- Added all 13 extension flags set to True (postgis, pgvector, pgcrypto, pgaudit,
  pg_repack, pg_stat_statements, pg_buffercache, pg_freespacemap, pgrowlocks,
  postgres_fdw, dblink, pg_stat_monitor, pgbadger)

Verification:
./install.py -s -l -v -c conf/prole.cfg completed successfully.
CNPG deployed prole-db:18-089 to knoe-dev-cluster; all milestones passed.

Co-authored-by: Junie <junie@jetbrains.com>
2026-03-01 20:40:44 -08:00

650 lines
26 KiB
Python
Raw Blame History

"""
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 installer.core.milestones import (
_stream_line,
DatabaseCreationMilestone,
DockerBuildMilestone,
GitOpsMilestone,
KerberosMilestone,
SupabaseMilestone,
DependenciesMilestone,
InitializationScriptsMilestone,
)
from installer.state import InstallerState
from installer.core.controller import ProleController
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _make_state(**inputs) -> InstallerState:
state = InstallerState()
state.inputs.update(inputs)
state.controller = MagicMock(spec=ProleController)
state.controller.project_root = Path("/tmp/fake-prole")
state.controller.get_prole_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("installer.core.milestones.inst_config._encrypt_cfg_secret", return_value="enc:xxx"), \
patch("installer.core.milestones.inst_config._is_prole_secret", return_value=False):
milestone.execute(state)
assert state.config_data.get("Database Creation", {}).get("DB_NAME") == "default"
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("installer.core.milestones.inst_config._encrypt_cfg_secret", return_value="enc:xxx"), \
patch("installer.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["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("installer.core.milestones.inst_config._encrypt_cfg_secret", return_value="enc:xxx"), \
patch("installer.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("installer.core.milestones.inst_config._resolve_secret_value", return_value="${openbao:kv/prole/db}"), \
patch("installer.core.milestones.inst_config._encrypt_cfg_secret", return_value="enc:xxx"), \
patch("installer.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("installer.core.milestones.inst_config._resolve_secret_value", return_value="realpassword"), \
patch("installer.core.milestones.inst_config._encrypt_cfg_secret", return_value="enc:xxx"), \
patch("installer.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("installer.core.milestones.inst_config._encrypt_cfg_secret", return_value="enc:xxx"), \
patch("installer.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("installer.core.milestones.inst_config._encrypt_cfg_secret", return_value="enc:xxx"), \
patch("installer.core.milestones.inst_config._is_prole_secret", return_value=False):
milestone.execute(state)
assert state.config_data["Database Creation"]["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("installer.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",
})
milestone = DockerBuildMilestone()
with patch("installer.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_failed_rc():
state = _make_state(**{"init_db_build.run_build": "True"})
state.controller.run_script.return_value = 1
milestone = DockerBuildMilestone()
with patch("installer.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("installer.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", [])
# ---------------------------------------------------------------------------
# SupabaseMilestone
# ---------------------------------------------------------------------------
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("installer.core.milestones.subprocess.run", return_value=mock_result), \
patch("installer.core.env._normalize_cluster_env", return_value="dev"), \
patch("installer.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("installer.core.milestones.subprocess.run", return_value=mock_result), \
patch("installer.core.env._normalize_cluster_env", return_value="dev"), \
patch("installer.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("installer.core.milestones.subprocess.run", return_value=mock_result), \
patch("installer.core.env._normalize_cluster_env", return_value="dev"), \
patch("installer.core.env._parse_bool", return_value=False):
milestone.execute(state, progress=progress)
progress.assert_called()
# ---------------------------------------------------------------------------
# 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()
missing_dep = {"id": "brew", "name": "Homebrew", "install_cmd": "brew-install"}
with patch("installer.config.get_dep_info", return_value=(False, None, None)):
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("installer.config.get_dep_info", return_value=(False, None, None)), \
patch("installer.config.DEPENDENCIES",
[{"id": "mytool", "name": "MyTool"}]):
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.brew.install": "False",
})
milestone = DependenciesMilestone()
dep = {"id": "brew", "name": "Homebrew", "install_cmd": "brew-install"}
with patch("installer.config.get_dep_info", return_value=(False, None, None)), \
patch("installer.config.DEPENDENCIES", [dep]):
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": "brew", "name": "Homebrew", "install_cmd": "brew-install"}
with patch("installer.config.get_dep_info", return_value=(False, None, None)), \
patch("installer.config.DEPENDENCIES", [dep]), \
patch.object(milestone, "_run_cmd", return_value=1):
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": "brew", "name": "Homebrew", "install_cmd": "brew-install"}
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/brew", "4.0")
with patch("installer.config.get_dep_info", side_effect=_dep_info), \
patch("installer.config.DEPENDENCIES", [dep]), \
patch.object(milestone, "_run_cmd", return_value=0):
milestone.execute(state)
assert state.config_data.get("Dependencies", {}).get("STATUS") == "All installed"
# ---------------------------------------------------------------------------
# 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("installer.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("installer.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("installer.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("installer.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("installer.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("installer.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("installer.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("installer.core.milestones.subprocess.run", side_effect=capture_run), \
patch("installer.core.env._normalize_cluster_env", return_value="prod"), \
patch("installer.core.env._parse_bool", return_value=False):
milestone.execute(state)
assert "k8s" in captured_cmd
Reference in New Issue View Git Blame Copy Permalink