mirror of
https://github.com/dredx/prole.git
synced 2026-09-24 19:54:32 +00:00
c24ca728dd
- Deleted `prole.cfg`, `port-mapping.cfg`, and PostgreSQL version file as part of configuration cleanup. - Enhanced Homebrew dependency resolution with `_resolve_brew_bin` and `_augment_env_for_brew` methods to manage PATH automatically. - Updated subprocess calls to respect augmented environment configurations. - Added unit tests for Homebrew PATH augmentation and dependency validation logic.
402 lines
14 KiB
Python
402 lines
14 KiB
Python
"""
|
|
Unit tests for previously untested installer/config.py helpers.
|
|
|
|
Covers:
|
|
_get_secret_key_file, _get_file_key, normalize_version,
|
|
_extract_inline_vault_block, _try_read_ansible_vault_value,
|
|
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
|
|
import subprocess
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
import pytest
|
|
|
|
from installer.config import (
|
|
_get_secret_key_file,
|
|
_get_file_key,
|
|
normalize_version,
|
|
_extract_yaml_scalar_from_text,
|
|
_extract_inline_vault_block,
|
|
_try_read_ansible_vault_value,
|
|
get_dep_info,
|
|
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") == ""
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _extract_inline_vault_block
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_extract_inline_vault_block_not_found():
|
|
text = "key: value\nother: x\n"
|
|
result = _extract_inline_vault_block(text, "missing")
|
|
assert result == ""
|
|
|
|
|
|
def test_extract_inline_vault_block_plain_value():
|
|
"""Key found but block doesn't start with $ANSIBLE_VAULT → return ''."""
|
|
text = "mykey:\n just some plain text\n more text\n"
|
|
result = _extract_inline_vault_block(text, "mykey")
|
|
assert result == ""
|
|
|
|
|
|
def test_extract_inline_vault_block_ansible_vault():
|
|
text = (
|
|
"mykey:\n"
|
|
" $ANSIBLE_VAULT;1.1;AES256\n"
|
|
" 6162636465666768\n"
|
|
)
|
|
result = _extract_inline_vault_block(text, "mykey")
|
|
assert "$ANSIBLE_VAULT" in result
|
|
|
|
|
|
def test_extract_inline_vault_block_no_block_lines():
|
|
"""Key found but indented block is empty → return ''."""
|
|
text = "mykey:\nother: x\n"
|
|
result = _extract_inline_vault_block(text, "mykey")
|
|
assert result == ""
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _try_read_ansible_vault_value
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_try_read_ansible_vault_value_file_not_found(tmp_path):
|
|
result = _try_read_ansible_vault_value(tmp_path / "nonexistent.cfg", "mykey")
|
|
assert result == ""
|
|
|
|
|
|
def test_try_read_ansible_vault_value_plain_text(tmp_path):
|
|
vault = tmp_path / "prole.cfg"
|
|
vault.write_text("db_password: supersecret\n")
|
|
result = _try_read_ansible_vault_value(vault, "db_password")
|
|
assert result == "supersecret"
|
|
|
|
|
|
def test_try_read_ansible_vault_value_plain_text_vault_marker_skipped(tmp_path):
|
|
"""Plain value starting with $ANSIBLE_VAULT is treated as encrypted → skip."""
|
|
vault = tmp_path / "prole.cfg"
|
|
vault.write_text("db_password: $ANSIBLE_VAULT;1.1;AES256\n")
|
|
# No vault password file → returns ""
|
|
with patch.dict(os.environ, {"ANSIBLE_VAULT_PASSWORD_FILE": ""}, clear=False), \
|
|
patch("installer.config.shutil.which", return_value=None):
|
|
result = _try_read_ansible_vault_value(vault, "db_password")
|
|
assert result == ""
|
|
|
|
|
|
def test_try_read_ansible_vault_value_no_password_file(tmp_path):
|
|
vault = tmp_path / "prole.cfg"
|
|
vault.write_text("db_password: $ANSIBLE_VAULT;1.1;AES256\n")
|
|
with patch.dict(os.environ, {"ANSIBLE_VAULT_PASSWORD_FILE": ""}, clear=False), \
|
|
patch("installer.config.shutil.which", return_value=None):
|
|
result = _try_read_ansible_vault_value(vault, "db_password")
|
|
assert result == ""
|
|
|
|
|
|
def test_try_read_ansible_vault_value_with_vault_success(tmp_path):
|
|
vault = tmp_path / "prole.cfg"
|
|
vault.write_text("db_password: $ANSIBLE_VAULT;1.1;AES256\n")
|
|
pw_file = tmp_path / ".vault_pass"
|
|
pw_file.write_text("mypassword\n")
|
|
|
|
mock_result = MagicMock()
|
|
mock_result.returncode = 0
|
|
mock_result.stdout = "db_password: decryptedvalue\n"
|
|
|
|
with patch.dict(os.environ, {"ANSIBLE_VAULT_PASSWORD_FILE": str(pw_file)}, clear=False), \
|
|
patch("installer.config.shutil.which", return_value="/usr/bin/ansible-vault"), \
|
|
patch("installer.config.subprocess.run", return_value=mock_result):
|
|
result = _try_read_ansible_vault_value(vault, "db_password")
|
|
assert result == "decryptedvalue"
|
|
|
|
|
|
def test_try_read_ansible_vault_value_vault_fails_no_inline(tmp_path):
|
|
vault = tmp_path / "prole.cfg"
|
|
vault.write_text("db_password: $ANSIBLE_VAULT;1.1;AES256\n")
|
|
pw_file = tmp_path / ".vault_pass"
|
|
pw_file.write_text("mypassword\n")
|
|
|
|
mock_result = MagicMock()
|
|
mock_result.returncode = 1
|
|
mock_result.stdout = ""
|
|
|
|
with patch.dict(os.environ, {"ANSIBLE_VAULT_PASSWORD_FILE": str(pw_file)}, clear=False), \
|
|
patch("installer.config.shutil.which", return_value="/usr/bin/ansible-vault"), \
|
|
patch("installer.config.subprocess.run", return_value=mock_result):
|
|
result = _try_read_ansible_vault_value(vault, "db_password")
|
|
assert result == ""
|
|
|
|
|
|
def test_try_read_ansible_vault_value_uses_project_root_vault_pass(tmp_path):
|
|
"""Falls back to PROJECT_ROOT/.vault_pass if no env var set."""
|
|
vault = tmp_path / "prole.cfg"
|
|
vault.write_text("mykey: plainvalue\n")
|
|
with patch.dict(os.environ, {"ANSIBLE_VAULT_PASSWORD_FILE": ""}, clear=False):
|
|
result = _try_read_ansible_vault_value(vault, "mykey")
|
|
assert result == "plainvalue"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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("installer.config.subprocess.run",
|
|
return_value=MagicMock(returncode=1, stdout="")):
|
|
ok, loc, ver = get_dep_info(dep)
|
|
assert ok is False
|
|
|
|
|
|
def test_get_dep_info_bin_found():
|
|
dep = {"id": "mytool", "name": "mytool", "bin": "mytool"}
|
|
with patch("installer.config.subprocess.run",
|
|
return_value=MagicMock(returncode=0, stdout="/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_results = [
|
|
MagicMock(returncode=0, stdout="/usr/bin/docker"), # command -v
|
|
MagicMock(returncode=0, stdout="Docker version 24.0"), # check_cmd
|
|
]
|
|
with patch("installer.config.subprocess.run", side_effect=run_results):
|
|
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 _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]
|
|
|
|
m = MagicMock()
|
|
cmd = " ".join(args)
|
|
if "command -v" in cmd:
|
|
m.returncode = 0
|
|
m.stdout = "/home/linuxbrew/.linuxbrew/bin/brew\n"
|
|
else:
|
|
m.returncode = 0
|
|
m.stdout = "Homebrew 5.0.0\n"
|
|
return m
|
|
|
|
with patch.dict(os.environ, {"PATH": "/usr/bin:/bin"}, clear=False), \
|
|
patch("installer.config.os.path.isfile", side_effect=lambda p: p == "/home/linuxbrew/.linuxbrew/bin/brew"), \
|
|
patch("installer.config.os.access", return_value=True), \
|
|
patch("installer.config.os.path.isdir", side_effect=lambda p: p == "/home/linuxbrew/.linuxbrew/sbin"), \
|
|
patch("installer.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_results = [
|
|
MagicMock(returncode=0, stdout="/usr/bin/kubectl"), # command -v
|
|
MagicMock(returncode=0, stdout="Client Version: v1.28.0"), # version_cmd
|
|
]
|
|
with patch("installer.config.subprocess.run", side_effect=run_results):
|
|
ok, loc, ver = get_dep_info(dep)
|
|
assert ok is True
|
|
assert ver == "Client Version: v1.28.0"
|
|
|
|
|
|
def test_get_dep_info_exception_returns_false():
|
|
dep = {"id": "broken", "name": "broken", "bin": "broken"}
|
|
with patch("installer.config.subprocess.run", 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("installer.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("installer.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("installer.config.get_config_value", return_value="myicon.png"), \
|
|
patch("installer.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("installer.config.get_config_value", return_value="bg.png"), \
|
|
patch("installer.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
|