chrisfu/prole
1
0
Fork 0
mirror of https://github.com/dredx/prole.git synced 2026-09-23 11:03:59 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

chore: refactor dependency resolution and auto-install logic

Browse Source 
- Replaced static dependency configuration with dynamic resolution using `get_required_dependencies` and `get_required_dependency_ids`.
- Streamlined runtime checks, fallback behaviors, and handling of missing/optional dependencies.
- Hardened GKE context acquisition and validation for Kubernetes clusters.
- Enhanced test coverage for dependency resolution under various deployment modes (`dev`, `prod`, `gke`).
This commit is contained in:
chrisfu 2026-04-12 18:17:35 -07:00
parent cedcaae2c5
commit 10bfc42bad
5 changed files with 426 additions and 58 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

43
knoe/config.py
View File

@ -1301,6 +1301,49 @@ def get_platform_dependencies() -> list[dict]:
return []
def _deployment_mode_hint(inputs: dict | None = None) -> str:
from knoe.core.env import _normalize_cluster_env
data = inputs or {}
cluster_env_raw = str(data.get("init_cluster.cluster_env", "")).strip()
normalized_env = (_normalize_cluster_env(cluster_env_raw) or "").strip().lower()
if normalized_env in {"prod", "production", "k8s"}:
return "gke"
if normalized_env in {"dev", "k3d", "k3s"}:
return normalized_env
deployment_mode = str(
data.get("build.deploy_env")
or data.get("deployment_mode")
or data.get("DEPLOYMENT_MODE")
or ""
).strip().lower()
if deployment_mode in {"prod", "production", "k8s", "gke"}:
return "gke"
if deployment_mode in {"dev", "k3d", "k3s"}:
return deployment_mode
return "generic"
def get_required_dependency_ids(inputs: dict | None = None) -> set[str]:
mode = _deployment_mode_hint(inputs)
base_required = {"python", "ansible", "kubectl", "kubectx", "docker"}
if mode == "gke":
return base_required | {"gcloud"}
if mode in {"dev", "k3d", "k3s"}:
return base_required | {"k3d"}
return base_required
def get_required_dependencies(inputs: dict | None = None) -> list[dict]:
dependencies = get_platform_dependencies()
required_ids = get_required_dependency_ids(inputs)
return [dep for dep in dependencies if str(dep.get("id")) in required_ids]
def _validate_dependency_backend(dependencies: list[dict] | None = None) -> tuple[bool, str]:
info = detect_dependency_platform()
if not info.get("supported"):

277
knoe/core/milestones.py
View File

@ -2,6 +2,7 @@ from __future__ import annotations
import logging
import os
import shlex
import shutil
import subprocess
import sys
import time
@ -50,6 +51,7 @@ class DependenciesMilestone(Milestone):
self, state: InstallerState, progress: ProgressCallback | None = None
) -> None:
platform_info = inst_config.detect_dependency_platform()
verify_all = self._parse_bool(state.inputs.get("dependencies.verify_all", "False"))
self.logger.info(
"Dependency platform detected: %s",
inst_config.get_dependency_platform_summary(),
@ -58,26 +60,28 @@ class DependenciesMilestone(Milestone):
if progress:
progress("Checking dependencies...", 0.1)
dependencies = inst_config.DEPENDENCIES
if not dependencies:
if verify_all:
dependencies = inst_config.get_platform_dependencies()
else:
dependencies = inst_config.get_required_dependencies(state.inputs)
backend_ok, backend_reason = inst_config._validate_dependency_backend()
required_ids = inst_config.get_required_dependency_ids(state.inputs)
backend_ok, backend_reason = inst_config._validate_dependency_backend(dependencies)
if not backend_ok:
self.logger.error("Dependency backend unsupported: %s", backend_reason)
self._set_status(state, "Missing")
return
self._fatal_dependency_failure(
state,
f"Dependency backend unsupported: {backend_reason}",
)
if not dependencies:
reason = platform_info.get("reason") or "No dependencies configured for detected platform"
self.logger.error("Dependency resolution failed: %s", reason)
self._set_status(state, "Missing")
return
self._fatal_dependency_failure(state, f"Dependency resolution failed: {reason}")
missing = []
missing: list[dict] = []
for i, dep in enumerate(dependencies):
ok, location, version = inst_config.get_dep_info(dep)
ok, _, version = inst_config.get_dep_info(dep)
if ok:
self.logger.info(f"[OK] {dep['name']} {version or ''}".strip())
else:
@ -90,12 +94,7 @@ class DependenciesMilestone(Milestone):
)
if not missing:
if not self._ensure_gcloud_auth_for_k8s(state):
self._set_status(state, "Missing")
return
self._set_status(state, "All installed")
if progress:
progress("All dependencies installed", 1.0)
self._finalize_dependency_prerequisites(state, progress)
return
# Check if auto-install is enabled
@ -103,9 +102,12 @@ class DependenciesMilestone(Milestone):
state.inputs.get("dependencies.auto_install_missing", "True")
)
if not auto_install:
self.logger.error("Dependencies missing and auto-install disabled.")
self._set_status(state, "Missing")
return
self._fatal_missing_dependencies(
state,
missing,
required_ids,
"Dependencies missing and auto-install disabled",
)
for i, dep in enumerate(missing):
install_cmd = dep.get("install_cmd")
@ -126,29 +128,238 @@ class DependenciesMilestone(Milestone):
rc = self._run_cmd(install_cmd)
if rc != 0:
self.logger.error(f"Install failed for {dep['name']} (code {rc})")
self.logger.error(
"Install failed for %s (code %s) [backend=%s, cmd=%s]",
dep["name"],
rc,
platform_info.get("package_manager") or "unknown",
install_cmd,
)
# Re-verify
still_missing = []
still_missing: list[dict] = []
for dep in missing:
ok, _, _ = inst_config.get_dep_info(dep)
if not ok:
still_missing.append(dep["name"])
still_missing.append(dep)
if still_missing:
self.logger.error(f"Still missing: {', '.join(still_missing)}")
self._set_status(state, "Missing")
self._fatal_missing_dependencies(
state,
still_missing,
required_ids,
"Dependencies unresolved after install attempts",
)
else:
if not self._ensure_gcloud_auth_for_k8s(state):
self._set_status(state, "Missing")
return
self._set_status(state, "All installed")
if progress:
progress("All dependencies installed", 1.0)
self._finalize_dependency_prerequisites(state, progress)
def _finalize_dependency_prerequisites(
self, state: InstallerState, progress: ProgressCallback | None = None
) -> None:
if not self._ensure_gcloud_auth_for_k8s(state):
self._fatal_dependency_failure(
state,
"GKE bootstrap failed: gcloud authentication is required before cluster operations.",
)
if not self._ensure_gke_contexts_for_k8s(state):
self._fatal_dependency_failure(
state,
"GKE bootstrap failed: required kube contexts are missing or could not be acquired.",
)
self._set_status(state, "All installed")
if progress:
progress("All dependencies installed", 1.0)
def _fatal_dependency_failure(self, state: InstallerState, message: str) -> None:
self.logger.error(message)
self._set_status(state, "Missing")
raise RuntimeError(message)
def _fatal_missing_dependencies(
self,
state: InstallerState,
missing: list[dict],
required_ids: set[str],
reason: str,
) -> None:
required_missing = [
str(dep.get("name") or dep.get("id"))
for dep in missing
if str(dep.get("id")) in required_ids
]
optional_missing = [
str(dep.get("name") or dep.get("id"))
for dep in missing
if str(dep.get("id")) not in required_ids
]
if optional_missing:
self.logger.warning("Optional dependencies still missing: %s", ", ".join(optional_missing))
if required_missing:
self._fatal_dependency_failure(
state,
f"{reason}. Required dependencies still missing: {', '.join(required_missing)}",
)
def _is_gke_mode(self, state: InstallerState) -> bool:
return "gcloud" in inst_config.get_required_dependency_ids(state.inputs)
def _gke_context_targets(self, state: InstallerState) -> list[dict]:
def _first_non_empty(*keys: str) -> str:
for key in keys:
value = str(state.inputs.get(key, "")).strip()
if value:
return value
return ""
project = _first_non_empty("init_cluster.project_id")
targets: list[dict] = []
for kind in ("app", "db"):
context = _first_non_empty(
f"init_cluster.{kind}_cluster_kubecontext",
f"env_setup.{kind.upper()}_CLUSTER_KUBECONTEXT",
)
cluster = _first_non_empty(
f"init_cluster.{kind}_cluster_name",
f"env_setup.{kind.upper()}_CLUSTER_NAME",
)
region = _first_non_empty(f"init_cluster.{kind}_cluster_region")
zone = _first_non_empty(f"init_cluster.{kind}_cluster_zone")
if not context:
continue
targets.append(
{
"kind": kind,
"context": context,
"cluster": cluster,
"region": region,
"zone": zone,
"project": project,
}
)
deduped: list[dict] = []
seen: set[str] = set()
for target in targets:
context = str(target.get("context") or "")
if context and context not in seen:
deduped.append(target)
seen.add(context)
return deduped
def _get_kube_contexts(self) -> set[str] | None:
kube_env = inst_config._augment_env_for_dependency_backend(os.environ.copy())
try:
res = subprocess.run(
["kubectl", "config", "get-contexts", "-o", "name"],
capture_output=True,
text=True,
env=kube_env,
timeout=20,
)
except Exception as exc:
self.logger.error("Failed to query kube contexts: %s", exc)
return None
if res.returncode != 0:
self.logger.error(
"Failed to list kube contexts (code %s): %s",
res.returncode,
(res.stderr or "").strip(),
)
return None
return {
line.strip()
for line in (res.stdout or "").splitlines()
if line.strip()
}
def _ensure_gke_contexts_for_k8s(self, state: InstallerState) -> bool:
if not self._is_gke_mode(state):
return True
targets = self._gke_context_targets(state)
if not targets:
return True
kube_contexts = self._get_kube_contexts()
if kube_contexts is None:
return False
missing_targets = [
target for target in targets if str(target.get("context")) not in kube_contexts
]
if not missing_targets:
return True
gcloud_bin = shutil.which("gcloud")
if not gcloud_bin:
self.logger.error(
"Missing gcloud; cannot acquire required GKE contexts: %s",
", ".join(str(t.get("context") or "") for t in missing_targets),
)
return False
for target in missing_targets:
context = str(target.get("context") or "")
cluster = str(target.get("cluster") or "")
if not cluster:
self.logger.error(
"Cannot acquire missing context '%s': cluster name is not configured.",
context,
)
return False
cmd = [gcloud_bin, "container", "clusters", "get-credentials", cluster]
region = str(target.get("region") or "")
zone = str(target.get("zone") or "")
project = str(target.get("project") or "")
if region:
cmd += ["--region", region]
elif zone:
cmd += ["--zone", zone]
if project:
cmd += ["--project", project]
self.logger.info(
"[ACTION] Acquiring GKE context '%s' via: %s",
context,
" ".join(shlex.quote(part) for part in cmd),
)
rc = self._run_cmd(cmd)
if rc != 0:
self.logger.error(
"Failed to acquire GKE context '%s' for cluster '%s' (code %s)",
context,
cluster,
rc,
)
return False
refreshed_contexts = self._get_kube_contexts()
if refreshed_contexts is None:
return False
still_missing = [
str(target.get("context") or "")
for target in targets
if str(target.get("context") or "") not in refreshed_contexts
]
if still_missing:
self.logger.error(
"GKE contexts still missing after credential acquisition: %s",
", ".join(still_missing),
)
return False
return True
def _ensure_gcloud_auth_for_k8s(self, state: InstallerState) -> bool:
cluster_env = str(state.inputs.get("init_cluster.cluster_env", "")).strip().lower()
if cluster_env not in {"prod", "production", "k8s"}:
if not self._is_gke_mode(state):
return True
gcloud_env = inst_config._augment_env_for_dependency_backend(os.environ.copy())

30
tests/installer/test_config_helpers.py
View File

@ -24,6 +24,8 @@ from knoe.config import (
detect_dependency_platform,
get_dependency_milestone_title,
get_platform_dependencies,
get_required_dependencies,
get_required_dependency_ids,
_validate_dependency_backend,
load_dependencies,
get_ui_icon_image_path,
@ -385,6 +387,34 @@ def test_get_platform_dependencies_linux_apt_includes_opentofu_without_brew_depe
assert "apt-get install -y opentofu" in str(opentofu.get("install_cmd") or "")
def test_get_required_dependency_ids_gke_excludes_k3d_and_opentofu():
required = get_required_dependency_ids(
{
"init_cluster.cluster_env": "prod",
"build.deploy_env": "Prod",
}
)
assert "gcloud" in required
assert "k3d" not in required
assert "opentofu" not in required
def test_get_required_dependencies_filters_platform_dependencies_for_gke():
deps = [
{"id": "python", "name": "python"},
{"id": "gcloud", "name": "gcloud"},
{"id": "k3d", "name": "k3d"},
{"id": "opentofu", "name": "OpenTofu"},
]
with patch("knoe.config.get_platform_dependencies", return_value=deps):
required = get_required_dependencies({"init_cluster.cluster_env": "prod"})
required_ids = [dep["id"] for dep in required]
assert required_ids == ["python", "gcloud"]
def test_validate_dependency_backend_rejects_linux_brew_commands():
with patch(
"knoe.config.detect_dependency_platform",

17
tests/installer/test_milestones.py
View File

@ -43,11 +43,13 @@ class TestMilestones(unittest.TestCase):
def test_dependencies_milestone_install_missing(
self, mock_popen, mock_get_dep_info
):
# Mock one missing dependency
# Mock one missing required dependency
dep = {"id": "python", "name": "python", "install_cmd": "apt install python3"}
def side_effect(dep):
if dep["id"] == "brew":
if dep["id"] == "python":
if mock_popen.called:
return (True, "/usr/local/bin/brew", "4.0.0")
return (True, "/usr/bin/python3", "3.10.0")
return (False, None, None)
return (True, "/usr/bin/python", "3.10.0")
@ -62,17 +64,18 @@ class TestMilestones(unittest.TestCase):
mock_process.wait.return_value = 0
mock_popen.return_value = mock_process
# Ensure brew install is enabled in state
self.state.inputs["dependencies.brew.install"] = "True"
self.state.inputs["dependencies.python.install"] = "True"
milestone = DependenciesMilestone()
milestone.execute(self.state)
with patch("knoe.config.get_required_dependencies", return_value=[dep]), \
patch("knoe.config.get_required_dependency_ids", return_value={"python"}):
milestone.execute(self.state)
self.assertEqual(
self.state.config_data.get("Dependencies", {}).get("STATUS"),
"All installed",
)
# Check that it tried to install brew
# Check that it tried to install python
mock_popen.assert_called()
@patch("knoe.config.get_resource_path")

117
tests/installer/test_milestones_extended.py
View File

@ -583,8 +583,11 @@ 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("knoe.config.get_dep_info", return_value=(False, None, None)):
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"
@ -595,8 +598,9 @@ def test_dependencies_no_install_cmd():
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.DEPENDENCIES",
[{"id": "mytool", "name": "MyTool"}]):
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"
@ -605,12 +609,14 @@ 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",
"dependencies.python.install": "False",
})
milestone = DependenciesMilestone()
dep = {"id": "brew", "name": "Homebrew", "install_cmd": "brew-install"}
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.DEPENDENCIES", [dep]):
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"
@ -619,10 +625,12 @@ 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"}
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.DEPENDENCIES", [dep]), \
patch.object(milestone, "_run_cmd", return_value=1):
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"
@ -631,7 +639,7 @@ 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"}
dep = {"id": "python", "name": "python", "install_cmd": "apt install python3"}
call_count = {"n": 0}
def _dep_info(d):
@ -639,10 +647,11 @@ def test_dependencies_install_succeeds_all():
# 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")
return (True, "/usr/bin/python3", "3.10")
with patch("knoe.config.get_dep_info", side_effect=_dep_info), \
patch("knoe.config.DEPENDENCIES", [dep]), \
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"
@ -656,7 +665,7 @@ def test_dependencies_k8s_fails_without_gcloud_auth_session():
}
)
milestone = DependenciesMilestone()
dep = {"id": "gcloud", "name": "gcloud", "install_cmd": "brew install gcloud"}
dep = {"id": "gcloud", "name": "gcloud", "install_cmd": "apt install google-cloud-cli"}
no_token = MagicMock(returncode=1, stdout="", stderr="not logged in")
@ -664,11 +673,13 @@ def test_dependencies_k8s_fails_without_gcloud_auth_session():
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.DEPENDENCIES", [dep]), \
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):
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"
@ -682,7 +693,7 @@ def test_dependencies_k8s_interactive_login_recovers_auth_session():
}
)
milestone = DependenciesMilestone()
dep = {"id": "gcloud", "name": "gcloud", "install_cmd": "brew install gcloud"}
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="")
@ -691,11 +702,13 @@ def test_dependencies_k8s_interactive_login_recovers_auth_session():
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.DEPENDENCIES", [dep]), \
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_contexts_for_k8s", return_value=True), \
patch.object(milestone, "_run_cmd", return_value=0) as run_cmd_mock:
milestone.execute(state)
@ -703,6 +716,74 @@ def test_dependencies_k8s_interactive_login_recovers_auth_session():
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_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", return_value=None), \
patch(
"knoe.core.milestones.subprocess.run",
side_effect=[token_ok, contexts_before],
), \
pytest.raises(RuntimeError, match="required kube contexts are missing"):
milestone.execute(state)
assert state.config_data.get("Dependencies", {}).get("STATUS") == "Missing"
# ---------------------------------------------------------------------------
# InitializationScriptsMilestone._regenerate_port_mapping_cfg
# ---------------------------------------------------------------------------