mirror of
https://github.com/dredx/prole.git
synced 2026-09-23 11:03:59 +00:00
521 lines
17 KiB
Python
521 lines
17 KiB
Python
"""
|
|
Unit tests for previously untested installer/config.py helpers.
|
|
|
|
Covers:
|
|
_get_secret_key_file, _get_file_key, normalize_version,
|
|
get_dep_info (check_cmd/version_cmd paths), load_dependencies,
|
|
get_ui_icon_image_path, get_ui_background_image_path, get_properties
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from knoe.config import (
|
|
_get_secret_key_file,
|
|
_get_file_key,
|
|
normalize_version,
|
|
_extract_yaml_scalar_from_text,
|
|
get_dep_info,
|
|
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,
|
|
get_ui_background_image_path,
|
|
get_properties,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _get_secret_key_file
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_get_secret_key_file_returns_path():
|
|
result = _get_secret_key_file()
|
|
assert isinstance(result, Path)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _get_file_key
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_get_file_key_generates_new_key(tmp_path):
|
|
key_path = tmp_path / "subdir" / "key.b64"
|
|
key = _get_file_key(key_path)
|
|
assert isinstance(key, bytes)
|
|
assert len(key) == 32
|
|
assert key_path.exists()
|
|
|
|
|
|
def test_get_file_key_reads_existing_key(tmp_path):
|
|
import base64
|
|
key_path = tmp_path / "key.b64"
|
|
raw_key = os.urandom(32)
|
|
key_path.write_text(base64.urlsafe_b64encode(raw_key).decode("utf-8"))
|
|
key = _get_file_key(key_path)
|
|
assert key == raw_key
|
|
|
|
|
|
def test_get_file_key_with_wrong_length_content(tmp_path):
|
|
"""If existing key file has content that decodes to non-32 bytes, still returns bytes."""
|
|
key_path = tmp_path / "key.b64"
|
|
# 'dGVzdA==' is valid base64 but decodes to only 4 bytes ('test')
|
|
key_path.write_text("dGVzdA==")
|
|
key = _get_file_key(key_path)
|
|
assert isinstance(key, bytes)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# normalize_version — edge cases
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_normalize_version_empty():
|
|
assert normalize_version("") == ""
|
|
|
|
|
|
def test_normalize_version_simple():
|
|
assert normalize_version("1.2.3") == "01.02.03"
|
|
|
|
|
|
def test_normalize_version_no_regex_match():
|
|
"""Text with no digit sequence matching the pattern falls back to findall."""
|
|
# A string with digits but no contiguous group matching pattern
|
|
result = normalize_version("v5")
|
|
assert "05" in result
|
|
|
|
|
|
def test_normalize_version_no_digits():
|
|
"""Text with no digits at all returns stripped text."""
|
|
result = normalize_version("nodigits")
|
|
assert result == "nodigits"
|
|
|
|
|
|
def test_normalize_version_prefix():
|
|
result = normalize_version("v1.2.3-beta")
|
|
assert result == "01.02.03"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _extract_yaml_scalar_from_text
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_extract_yaml_scalar_simple():
|
|
text = 'key: somevalue\nother: x\n'
|
|
assert _extract_yaml_scalar_from_text(text, "key") == "somevalue"
|
|
|
|
|
|
def test_extract_yaml_scalar_missing_key():
|
|
assert _extract_yaml_scalar_from_text("a: b\n", "missing") == ""
|
|
|
|
|
|
def test_extract_yaml_scalar_plain_value():
|
|
"""Plain YAML scalar extraction works for config files."""
|
|
text = "db_password: supersecret\n"
|
|
result = _extract_yaml_scalar_from_text(text, "db_password")
|
|
assert result == "supersecret"
|
|
|
|
|
|
def test_extract_yaml_scalar_missing_key():
|
|
text = "other_key: value\n"
|
|
result = _extract_yaml_scalar_from_text(text, "db_password")
|
|
assert result == ""
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# get_dep_info — check_cmd and version_cmd paths
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _mock_run_results(*return_codes_and_outputs):
|
|
"""Build a side_effect list for subprocess.run."""
|
|
results = []
|
|
for rc, out in return_codes_and_outputs:
|
|
m = MagicMock()
|
|
m.returncode = rc
|
|
m.stdout = out
|
|
results.append(m)
|
|
return results
|
|
|
|
|
|
def test_get_dep_info_bin_not_found():
|
|
dep = {"id": "mytool", "name": "mytool", "bin": "mytool"}
|
|
with patch("knoe.config.shutil.which", return_value=None):
|
|
ok, loc, ver = get_dep_info(dep)
|
|
assert ok is False
|
|
assert loc is None
|
|
|
|
|
|
def test_get_dep_info_bin_found():
|
|
dep = {"id": "mytool", "name": "mytool", "bin": "mytool"}
|
|
with patch("knoe.config.shutil.which", return_value="/usr/bin/mytool"):
|
|
ok, loc, ver = get_dep_info(dep)
|
|
assert ok is True
|
|
assert loc == "/usr/bin/mytool"
|
|
|
|
|
|
def test_get_dep_info_check_cmd_success():
|
|
dep = {"id": "docker", "name": "docker", "bin": "docker",
|
|
"check_cmd": "docker info"}
|
|
run_result = MagicMock(returncode=0, stdout="Docker version 24.0")
|
|
with patch("knoe.config.shutil.which", return_value="/usr/bin/docker"), \
|
|
patch("knoe.config.subprocess.run", return_value=run_result):
|
|
ok, loc, ver = get_dep_info(dep)
|
|
assert ok is True
|
|
assert "Docker" in (ver or "")
|
|
|
|
|
|
def test_get_dep_info_prepends_linuxbrew_to_path_when_present():
|
|
dep = {
|
|
"id": "brew",
|
|
"name": "Homebrew",
|
|
"bin": "brew",
|
|
"check_cmd": "brew --version",
|
|
}
|
|
|
|
def _which(cmd, mode=os.F_OK | os.X_OK, path=None):
|
|
assert cmd == "brew"
|
|
assert path is not None
|
|
path_parts = [p for p in (path or "").split(os.pathsep) if p]
|
|
assert path_parts[0] == "/home/linuxbrew/.linuxbrew/bin"
|
|
assert "/home/linuxbrew/.linuxbrew/sbin" in path_parts[:2]
|
|
return "/home/linuxbrew/.linuxbrew/bin/brew"
|
|
|
|
def _run(args, capture_output=False, text=False, env=None, **kwargs):
|
|
assert env is not None
|
|
path_parts = (env.get("PATH") or "").split(os.pathsep)
|
|
assert path_parts[0] == "/home/linuxbrew/.linuxbrew/bin"
|
|
assert "/home/linuxbrew/.linuxbrew/sbin" in path_parts[:2]
|
|
|
|
assert args[:2] == ["bash", "-c"]
|
|
|
|
m = MagicMock()
|
|
m.returncode = 0
|
|
m.stdout = "Homebrew 5.0.0\n"
|
|
return m
|
|
|
|
with patch.dict(os.environ, {"PATH": "/usr/bin:/bin"}, clear=False), \
|
|
patch(
|
|
"knoe.config.detect_dependency_platform",
|
|
return_value={
|
|
"os": "macos",
|
|
"arch": "arm64",
|
|
"package_manager": "brew",
|
|
"package_family": "brew",
|
|
"supported": True,
|
|
"reason": None,
|
|
},
|
|
), \
|
|
patch("knoe.config.os.path.isfile", side_effect=lambda p: p == "/home/linuxbrew/.linuxbrew/bin/brew"), \
|
|
patch("knoe.config.os.access", return_value=True), \
|
|
patch("knoe.config.os.path.isdir", side_effect=lambda p: p == "/home/linuxbrew/.linuxbrew/sbin"), \
|
|
patch("knoe.config.shutil.which", side_effect=_which), \
|
|
patch("knoe.config.subprocess.run", side_effect=_run):
|
|
ok, loc, ver = get_dep_info(dep)
|
|
|
|
assert ok is True
|
|
assert loc == "/home/linuxbrew/.linuxbrew/bin/brew"
|
|
assert ver == "Homebrew 5.0.0"
|
|
|
|
|
|
def test_get_dep_info_version_cmd_success():
|
|
dep = {"id": "kubectl", "name": "kubectl", "bin": "kubectl",
|
|
"version_cmd": "kubectl version --client --short"}
|
|
run_result = MagicMock(returncode=0, stdout="Client Version: v1.28.0")
|
|
with patch("knoe.config.shutil.which", return_value="/usr/bin/kubectl"), \
|
|
patch("knoe.config.subprocess.run", return_value=run_result):
|
|
ok, loc, ver = get_dep_info(dep)
|
|
assert ok is True
|
|
assert ver == "Client Version: v1.28.0"
|
|
|
|
|
|
def test_detect_dependency_platform_linux_apt():
|
|
with patch("knoe.config._DEP_PLATFORM_CACHE", None), \
|
|
patch("knoe.config.platform.system", return_value="Linux"), \
|
|
patch("knoe.config.platform.machine", return_value="aarch64"), \
|
|
patch(
|
|
"knoe.config.shutil.which",
|
|
side_effect=lambda cmd: {
|
|
"apt-get": "/usr/bin/apt-get",
|
|
"dpkg": "/usr/bin/dpkg",
|
|
}.get(cmd),
|
|
):
|
|
info = detect_dependency_platform(refresh=True)
|
|
|
|
assert info["os"] == "linux"
|
|
assert info["package_family"] == "apt/deb"
|
|
assert info["package_manager"] == "apt"
|
|
assert info["supported"] is True
|
|
|
|
|
|
def test_detect_dependency_platform_linux_rejects_brew_fallback_by_default():
|
|
with patch("knoe.config._DEP_PLATFORM_CACHE", None), \
|
|
patch("knoe.config.platform.system", return_value="Linux"), \
|
|
patch("knoe.config.platform.machine", return_value="aarch64"), \
|
|
patch("knoe.config.shutil.which", side_effect=lambda cmd: "/home/linuxbrew/.linuxbrew/bin/brew" if cmd == "brew" else None), \
|
|
patch.dict(os.environ, {"PROLE_LINUX_ALLOW_BREW": ""}, clear=False):
|
|
info = detect_dependency_platform(refresh=True)
|
|
|
|
assert info["os"] == "linux"
|
|
assert info["supported"] is False
|
|
assert "Refusing brew fallback" in str(info["reason"])
|
|
|
|
|
|
def test_get_platform_dependencies_linux_apt_includes_opentofu_without_brew_dependency():
|
|
with patch(
|
|
"knoe.config.detect_dependency_platform",
|
|
return_value={
|
|
"os": "linux",
|
|
"arch": "aarch64",
|
|
"package_manager": "apt",
|
|
"package_family": "apt/deb",
|
|
"supported": True,
|
|
"reason": None,
|
|
},
|
|
):
|
|
deps = get_platform_dependencies()
|
|
|
|
ids = {d["id"] for d in deps}
|
|
assert "brew" not in ids
|
|
opentofu = next(d for d in deps if d["id"] == "opentofu")
|
|
assert "apt-cache show opentofu" in str(opentofu.get("install_cmd") or "")
|
|
|
|
|
|
def test_get_platform_dependencies_linux_apt_gcloud_bootstraps_vendor_repo_when_missing():
|
|
with patch(
|
|
"knoe.config.detect_dependency_platform",
|
|
return_value={
|
|
"os": "linux",
|
|
"arch": "aarch64",
|
|
"package_manager": "apt",
|
|
"package_family": "apt/deb",
|
|
"supported": True,
|
|
"reason": None,
|
|
},
|
|
):
|
|
deps = get_platform_dependencies()
|
|
|
|
gcloud = next(d for d in deps if d["id"] == "gcloud")
|
|
cmd = str(gcloud.get("install_cmd") or "")
|
|
|
|
assert "apt-cache show google-cloud-cli" in cmd
|
|
assert "packages.cloud.google.com/apt" in cmd
|
|
assert "google-cloud.gpg" in cmd
|
|
assert "apt-get update" in cmd
|
|
assert "apt-get install -y google-cloud-cli" in cmd
|
|
assert "adding vendor repository 'Google Cloud SDK'" in cmd
|
|
assert "Added vendor apt repository 'Google Cloud SDK'" in cmd
|
|
assert "[ERROR] Failed to install apt package google-cloud-cli after repository bootstrap." in cmd
|
|
|
|
|
|
def test_get_platform_dependencies_linux_apt_includes_gke_auth_plugin_package():
|
|
with patch(
|
|
"knoe.config.detect_dependency_platform",
|
|
return_value={
|
|
"os": "linux",
|
|
"arch": "aarch64",
|
|
"package_manager": "apt",
|
|
"package_family": "apt/deb",
|
|
"supported": True,
|
|
"reason": None,
|
|
},
|
|
):
|
|
deps = get_platform_dependencies()
|
|
|
|
plugin_dep = next(d for d in deps if d["id"] == "gke-gcloud-auth-plugin")
|
|
cmd = str(plugin_dep.get("install_cmd") or "")
|
|
assert "google-cloud-cli-gke-gcloud-auth-plugin" in cmd
|
|
assert "apt-get install -y google-cloud-cli-gke-gcloud-auth-plugin" in cmd
|
|
|
|
|
|
def test_get_platform_dependencies_linux_apt_opentofu_bootstraps_vendor_repo_when_missing():
|
|
with patch(
|
|
"knoe.config.detect_dependency_platform",
|
|
return_value={
|
|
"os": "linux",
|
|
"arch": "aarch64",
|
|
"package_manager": "apt",
|
|
"package_family": "apt/deb",
|
|
"supported": True,
|
|
"reason": None,
|
|
},
|
|
):
|
|
deps = get_platform_dependencies()
|
|
|
|
opentofu = next(d for d in deps if d["id"] == "opentofu")
|
|
cmd = str(opentofu.get("install_cmd") or "")
|
|
|
|
assert "apt-cache show opentofu" in cmd
|
|
assert "packages.opentofu.org" in cmd
|
|
assert "opentofu.gpg" in cmd
|
|
assert "apt-get update" in cmd
|
|
assert "apt-get install -y opentofu" in cmd
|
|
assert "adding vendor repository 'OpenTofu'" in cmd
|
|
assert "Added vendor apt repository 'OpenTofu'" in cmd
|
|
assert "[ERROR] Failed to install apt package opentofu after repository bootstrap." in cmd
|
|
|
|
|
|
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_dependency_ids_gke_linux_apt_includes_gke_auth_plugin():
|
|
with patch(
|
|
"knoe.config.detect_dependency_platform",
|
|
return_value={
|
|
"os": "linux",
|
|
"arch": "x86_64",
|
|
"package_manager": "apt",
|
|
"package_family": "apt/deb",
|
|
"supported": True,
|
|
"reason": None,
|
|
},
|
|
):
|
|
required = get_required_dependency_ids({"init_cluster.cluster_env": "prod"})
|
|
|
|
assert "gcloud" in required
|
|
assert "gke-gcloud-auth-plugin" 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",
|
|
return_value={
|
|
"os": "linux",
|
|
"arch": "aarch64",
|
|
"package_manager": "apt",
|
|
"package_family": "apt/deb",
|
|
"supported": True,
|
|
"reason": None,
|
|
},
|
|
):
|
|
ok, reason = _validate_dependency_backend(
|
|
[{"id": "opentofu", "install_cmd": "brew install opentofu"}]
|
|
)
|
|
|
|
assert ok is False
|
|
assert "Homebrew commands" in reason
|
|
|
|
|
|
def test_get_dependency_milestone_title_linux_and_macos():
|
|
with patch(
|
|
"knoe.config.detect_dependency_platform",
|
|
return_value={"os": "linux"},
|
|
):
|
|
assert get_dependency_milestone_title() == "Linux Dependencies"
|
|
|
|
with patch(
|
|
"knoe.config.detect_dependency_platform",
|
|
return_value={"os": "macos"},
|
|
):
|
|
assert get_dependency_milestone_title() == "Mac OS Dependencies"
|
|
|
|
|
|
def test_get_dep_info_exception_returns_false():
|
|
dep = {"id": "broken", "name": "broken", "bin": "broken"}
|
|
with patch("knoe.config.shutil.which", side_effect=OSError("broken")):
|
|
ok, loc, ver = get_dep_info(dep)
|
|
assert ok is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# load_dependencies
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_load_dependencies_no_refresh():
|
|
deps = load_dependencies(refresh=False)
|
|
assert isinstance(deps, list)
|
|
assert all(isinstance(d, dict) for d in deps)
|
|
|
|
|
|
def test_load_dependencies_with_refresh():
|
|
with patch("knoe.config.get_dep_info",
|
|
return_value=(True, "/usr/bin/test", "1.0")):
|
|
deps = load_dependencies(refresh=True)
|
|
assert isinstance(deps, list)
|
|
for dep in deps:
|
|
assert dep.get("installed") is True
|
|
|
|
|
|
def test_load_dependencies_refresh_exception_handled():
|
|
"""If get_dep_info raises, dep.installed should be False."""
|
|
with patch("knoe.config.get_dep_info", side_effect=RuntimeError("oops")):
|
|
deps = load_dependencies(refresh=True)
|
|
for dep in deps:
|
|
assert dep.get("installed") is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# get_ui_icon_image_path / get_ui_background_image_path
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_get_ui_icon_image_path_returns_path():
|
|
result = get_ui_icon_image_path()
|
|
assert isinstance(result, Path)
|
|
|
|
|
|
def test_get_ui_icon_image_path_existing_file(tmp_path):
|
|
img = tmp_path / "myicon.png"
|
|
img.write_bytes(b"PNG")
|
|
with patch("knoe.config.get_config_value", return_value="myicon.png"), \
|
|
patch("knoe.config.PROJECT_ROOT", tmp_path):
|
|
result = get_ui_icon_image_path()
|
|
assert result == img.resolve()
|
|
|
|
|
|
def test_get_ui_background_image_path_returns_path():
|
|
result = get_ui_background_image_path()
|
|
assert isinstance(result, Path)
|
|
|
|
|
|
def test_get_ui_background_image_path_existing_file(tmp_path):
|
|
img = tmp_path / "bg.png"
|
|
img.write_bytes(b"PNG")
|
|
with patch("knoe.config.get_config_value", return_value="bg.png"), \
|
|
patch("knoe.config.PROJECT_ROOT", tmp_path):
|
|
result = get_ui_background_image_path()
|
|
assert result == img.resolve()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# get_properties
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_get_properties_returns_dict():
|
|
result = get_properties()
|
|
assert isinstance(result, dict)
|
|
|
|
|
|
def test_get_properties_consistent(tmp_path):
|
|
"""Calling get_properties twice returns consistent results."""
|
|
r1 = get_properties()
|
|
r2 = get_properties()
|
|
assert r1 == r2
|