mirror of
https://github.com/dredx/prole.git
synced 2026-09-23 12:03:59 +00:00
489 lines
16 KiB
Python
489 lines
16 KiB
Python
"""Extended tests for installer/config.py – covering encryption, properties, helpers."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import configparser
|
||
import os
|
||
from pathlib import Path
|
||
from unittest import mock
|
||
from unittest.mock import MagicMock, patch
|
||
|
||
from knoe.config import (
|
||
_is_knoe_secret,
|
||
_is_openbao_ref,
|
||
_encrypt_knoe_secret,
|
||
_decrypt_knoe_secret,
|
||
_encrypt_cfg_secret,
|
||
_resolve_secret_value,
|
||
_load_properties,
|
||
_expand_path,
|
||
_collect_cfg_vars,
|
||
_expand_cfg_value,
|
||
_parse_bool,
|
||
_extract_yaml_scalar_from_text,
|
||
_extract_inline_vault_block,
|
||
_write_k3s_kubeconfig,
|
||
_get_file_key,
|
||
normalize_version,
|
||
get_resource_path,
|
||
get_docker_build_platform_args,
|
||
setup_logging,
|
||
KNOE_SECRET_PREFIX,
|
||
KNOE_SECRET_SUFFIX,
|
||
OPENBAO_PREFIX,
|
||
OPENBAO_SUFFIX,
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _is_knoe_secret / _is_openbao_ref
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestSecretDetection:
|
||
def test_knoe_secret_valid(self):
|
||
val = f"{KNOE_SECRET_PREFIX}v1:nonce:ct{KNOE_SECRET_SUFFIX}"
|
||
assert _is_knoe_secret(val) is True
|
||
|
||
def test_knoe_secret_none(self):
|
||
assert _is_knoe_secret(None) is False
|
||
|
||
def test_knoe_secret_empty(self):
|
||
assert _is_knoe_secret("") is False
|
||
|
||
def test_knoe_secret_missing_prefix(self):
|
||
assert _is_knoe_secret(f"garbage{KNOE_SECRET_SUFFIX}") is False
|
||
|
||
def test_knoe_secret_missing_suffix(self):
|
||
assert _is_knoe_secret(f"{KNOE_SECRET_PREFIX}stuff") is False
|
||
|
||
def test_openbao_ref_valid(self):
|
||
val = f"{OPENBAO_PREFIX}kv/path:key{OPENBAO_SUFFIX}"
|
||
assert _is_openbao_ref(val) is True
|
||
|
||
def test_openbao_ref_none(self):
|
||
assert _is_openbao_ref(None) is False
|
||
|
||
def test_openbao_ref_empty(self):
|
||
assert _is_openbao_ref("") is False
|
||
|
||
def test_openbao_ref_missing_suffix(self):
|
||
assert _is_openbao_ref(f"{OPENBAO_PREFIX}stuff") is False
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Encryption round-trip
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestEncryptionRoundTrip:
|
||
def test_encrypt_decrypt(self):
|
||
plaintext = "my-secret-password"
|
||
encrypted = _encrypt_knoe_secret(plaintext)
|
||
assert _is_knoe_secret(encrypted)
|
||
decrypted = _decrypt_knoe_secret(encrypted)
|
||
assert decrypted == plaintext
|
||
|
||
def test_encrypt_none(self):
|
||
assert _encrypt_knoe_secret(None) == ""
|
||
|
||
def test_encrypt_already_encrypted(self):
|
||
val = f"{KNOE_SECRET_PREFIX}v1:x:y{KNOE_SECRET_SUFFIX}"
|
||
assert _encrypt_knoe_secret(val) == val
|
||
|
||
def test_decrypt_non_secret(self):
|
||
assert _decrypt_knoe_secret("plaintext") == "plaintext"
|
||
|
||
def test_decrypt_malformed(self):
|
||
val = f"{KNOE_SECRET_PREFIX}bad{KNOE_SECRET_SUFFIX}"
|
||
assert _decrypt_knoe_secret(val) == val
|
||
|
||
def test_decrypt_wrong_version(self):
|
||
val = f"{KNOE_SECRET_PREFIX}v999:nonce:ct{KNOE_SECRET_SUFFIX}"
|
||
assert _decrypt_knoe_secret(val) == val
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _encrypt_cfg_secret
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestEncryptCfgSecret:
|
||
def test_empty(self):
|
||
assert _encrypt_cfg_secret("") == ""
|
||
assert _encrypt_cfg_secret(None) == ""
|
||
|
||
def test_already_knoe_secret(self):
|
||
val = f"{KNOE_SECRET_PREFIX}v1:x:y{KNOE_SECRET_SUFFIX}"
|
||
assert _encrypt_cfg_secret(val) == val
|
||
|
||
def test_already_openbao_ref(self):
|
||
val = f"{OPENBAO_PREFIX}kv/path:key{OPENBAO_SUFFIX}"
|
||
assert _encrypt_cfg_secret(val) == val
|
||
|
||
def test_encrypts_plaintext(self):
|
||
result = _encrypt_cfg_secret("my-password")
|
||
assert _is_knoe_secret(result)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _resolve_secret_value
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestResolveSecretValue:
|
||
def test_plain_value(self):
|
||
assert _resolve_secret_value("hello") == "hello"
|
||
|
||
def test_knoe_secret(self):
|
||
encrypted = _encrypt_knoe_secret("test123")
|
||
assert _resolve_secret_value(encrypted) == "test123"
|
||
|
||
def test_openbao_ref_no_server(self):
|
||
val = f"{OPENBAO_PREFIX}kv/path:key{OPENBAO_SUFFIX}"
|
||
with mock.patch.dict(os.environ, {}, clear=False):
|
||
os.environ.pop("OPENBAO_ROOT_TOKEN", None)
|
||
os.environ.pop("KNOE_SERVICE", None)
|
||
os.environ.pop("PROLE_OPENBAO_URL", None)
|
||
result = _resolve_secret_value(val)
|
||
# Without a token/server, should return the original ref
|
||
assert result == val
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _load_properties
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestLoadProperties:
|
||
def test_basic(self, tmp_path):
|
||
f = tmp_path / "test.properties"
|
||
f.write_text("key1=val1\nkey2=val2\n# comment\n\nkey3 = val3\n")
|
||
props = _load_properties(f)
|
||
assert props == {"key1": "val1", "key2": "val2", "key3": "val3"}
|
||
|
||
def test_empty_file(self, tmp_path):
|
||
f = tmp_path / "empty.properties"
|
||
f.write_text("")
|
||
assert _load_properties(f) == {}
|
||
|
||
def test_comments_only(self, tmp_path):
|
||
f = tmp_path / "comments.properties"
|
||
f.write_text("# comment 1\n# comment 2\n")
|
||
assert _load_properties(f) == {}
|
||
|
||
def test_file_not_found(self, tmp_path):
|
||
f = tmp_path / "nonexistent.properties"
|
||
assert _load_properties(f) == {}
|
||
|
||
def test_equals_in_value(self, tmp_path):
|
||
f = tmp_path / "eq.properties"
|
||
f.write_text("key=val=with=equals\n")
|
||
props = _load_properties(f)
|
||
assert props["key"] == "val=with=equals"
|
||
|
||
def test_empty_key_skipped(self, tmp_path):
|
||
f = tmp_path / "blank.properties"
|
||
f.write_text("=value\ngood=val\n")
|
||
props = _load_properties(f)
|
||
assert "good" in props
|
||
assert "" not in props
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _expand_path
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestExpandPath:
|
||
def test_none(self):
|
||
assert _expand_path(None) == ""
|
||
|
||
def test_empty(self):
|
||
assert _expand_path("") == ""
|
||
|
||
def test_tilde(self):
|
||
result = _expand_path("~/test")
|
||
assert "~" not in result
|
||
assert "test" in result
|
||
|
||
def test_absolute(self):
|
||
result = _expand_path("/tmp/test")
|
||
assert result == "/tmp/test"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _collect_cfg_vars / _expand_cfg_value
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestCfgExpansion:
|
||
def test_collect_vars(self):
|
||
cfg = configparser.ConfigParser(interpolation=None)
|
||
cfg.add_section("Global")
|
||
cfg.set("Global", "namespace", "my-ns")
|
||
cfg.add_section("System Environment")
|
||
cfg.set("System Environment", "knoe_home", "/opt/knoe")
|
||
variables = _collect_cfg_vars(cfg)
|
||
assert variables["namespace"] == "my-ns"
|
||
assert variables["knoe_home"] == "/opt/knoe"
|
||
|
||
def test_collect_vars_no_sections(self):
|
||
cfg = configparser.ConfigParser(interpolation=None)
|
||
assert _collect_cfg_vars(cfg) == {}
|
||
|
||
def test_expand_value(self):
|
||
result = _expand_cfg_value("ns=${namespace}", {"namespace": "my-ns"})
|
||
assert result == "ns=my-ns"
|
||
|
||
def test_expand_no_vars(self):
|
||
assert _expand_cfg_value("plain", {}) == "plain"
|
||
|
||
def test_expand_empty(self):
|
||
assert _expand_cfg_value("", {}) == ""
|
||
|
||
def test_expand_none(self):
|
||
assert _expand_cfg_value(None, {}) is None
|
||
|
||
def test_expand_missing_var(self):
|
||
result = _expand_cfg_value("${missing}", {})
|
||
assert result == "${missing}"
|
||
|
||
def test_expand_multiple_vars(self):
|
||
result = _expand_cfg_value("${a}-${b}", {"a": "X", "b": "Y"})
|
||
assert result == "X-Y"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _parse_bool (extended from test_config.py)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestParseBoolExtended:
|
||
def test_none_with_default(self):
|
||
assert _parse_bool(None, default=True) is True
|
||
assert _parse_bool(None, default=False) is False
|
||
|
||
def test_bool_passthrough(self):
|
||
assert _parse_bool(True) is True
|
||
assert _parse_bool(False) is False
|
||
|
||
def test_string_true_variants(self):
|
||
for v in ("true", "True", "TRUE", "1", "yes", "Yes", "on", "ON"):
|
||
assert _parse_bool(v) is True, f"Failed for {v!r}"
|
||
|
||
def test_string_false_variants(self):
|
||
for v in ("false", "False", "FALSE", "0", "no", "No", "off", "OFF"):
|
||
assert _parse_bool(v) is False, f"Failed for {v!r}"
|
||
|
||
def test_unrecognized_returns_default(self):
|
||
assert _parse_bool("maybe") is False
|
||
assert _parse_bool("maybe", default=True) is True
|
||
|
||
def test_whitespace_stripped(self):
|
||
assert _parse_bool(" true ") is True
|
||
assert _parse_bool(" false ") is False
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _extract_yaml_scalar_from_text
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestExtractYamlScalar:
|
||
def test_simple(self):
|
||
text = "key1: value1\nkey2: value2\n"
|
||
assert _extract_yaml_scalar_from_text(text, "key1") == "value1"
|
||
assert _extract_yaml_scalar_from_text(text, "key2") == "value2"
|
||
|
||
def test_quoted(self):
|
||
text = 'key: "quoted value"\n'
|
||
assert _extract_yaml_scalar_from_text(text, "key") == "quoted value"
|
||
|
||
def test_single_quoted(self):
|
||
text = "key: 'single quoted'\n"
|
||
assert _extract_yaml_scalar_from_text(text, "key") == "single quoted"
|
||
|
||
def test_not_found(self):
|
||
text = "key: value\n"
|
||
assert _extract_yaml_scalar_from_text(text, "missing") == ""
|
||
|
||
def test_empty_text(self):
|
||
assert _extract_yaml_scalar_from_text("", "key") == ""
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _extract_inline_vault_block
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestExtractInlineVaultBlock:
|
||
def test_vault_block(self):
|
||
text = (
|
||
"top_key: stuff\n"
|
||
"secret_key:\n"
|
||
" $ANSIBLE_VAULT;1.1;AES256\n"
|
||
" 6162636465\n"
|
||
"other_key: val\n"
|
||
)
|
||
result = _extract_inline_vault_block(text, "secret_key")
|
||
assert "$ANSIBLE_VAULT" in result
|
||
|
||
def test_no_vault_block(self):
|
||
text = "key:\n" " just plain text\n"
|
||
result = _extract_inline_vault_block(text, "key")
|
||
assert result == ""
|
||
|
||
def test_key_not_found(self):
|
||
text = "key: value\n"
|
||
assert _extract_inline_vault_block(text, "missing") == ""
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _write_k3s_kubeconfig
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestWriteK3sKubeconfig:
|
||
def test_creates_file(self):
|
||
path = _write_k3s_kubeconfig("https://10.0.0.1:6443", "mytoken")
|
||
assert path.exists()
|
||
content = path.read_text()
|
||
assert "knoe-k3s" in content
|
||
assert "https://10.0.0.1:6443" in content
|
||
assert "mytoken" in content
|
||
# Cleanup
|
||
path.unlink(missing_ok=True)
|
||
|
||
def test_adds_https(self):
|
||
path = _write_k3s_kubeconfig("10.0.0.1:6443", "tok")
|
||
content = path.read_text()
|
||
assert "https://10.0.0.1:6443" in content
|
||
path.unlink(missing_ok=True)
|
||
|
||
def test_already_https(self):
|
||
path = _write_k3s_kubeconfig("https://server:6443", "tok")
|
||
content = path.read_text()
|
||
# Should not double the https
|
||
assert "https://https://" not in content
|
||
path.unlink(missing_ok=True)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _get_file_key
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestGetFileKey:
|
||
def test_creates_key_file(self, tmp_path):
|
||
key_file = tmp_path / "secret.key"
|
||
key = _get_file_key(key_file)
|
||
assert len(key) == 32
|
||
assert key_file.exists()
|
||
|
||
def test_reads_existing_key(self, tmp_path):
|
||
key_file = tmp_path / "secret.key"
|
||
key1 = _get_file_key(key_file)
|
||
key2 = _get_file_key(key_file)
|
||
assert key1 == key2
|
||
|
||
def test_creates_parent_dirs(self, tmp_path):
|
||
key_file = tmp_path / "deep" / "nested" / "secret.key"
|
||
key = _get_file_key(key_file)
|
||
assert len(key) == 32
|
||
assert key_file.parent.exists()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# get_docker_build_platform_args (extended)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestDockerBuildPlatformExtended:
|
||
@patch.dict(os.environ, {"PROLE_DOCKER_PLATFORM": "linux/arm64"})
|
||
def test_env_override(self):
|
||
result = get_docker_build_platform_args()
|
||
assert result == ["--platform", "linux/arm64"]
|
||
|
||
@patch.dict(os.environ, {"PROLE_DOCKER_PLATFORM": ""}, clear=False)
|
||
@patch("knoe.config.is_apple_silicon", return_value=True)
|
||
def test_dev_on_apple_silicon(self, mock_as):
|
||
result = get_docker_build_platform_args("dev")
|
||
assert result == ["--platform", "linux/arm64"]
|
||
|
||
@patch.dict(os.environ, {"PROLE_DOCKER_PLATFORM": ""}, clear=False)
|
||
@patch("knoe.config.is_apple_silicon", return_value=False)
|
||
def test_service_env(self, mock_as):
|
||
result = get_docker_build_platform_args("service")
|
||
assert result == ["--platform", "linux/arm64"]
|
||
|
||
@patch.dict(os.environ, {"PROLE_DOCKER_PLATFORM": ""}, clear=False)
|
||
@patch("knoe.config.is_apple_silicon", return_value=False)
|
||
def test_k3s_env(self, mock_as):
|
||
result = get_docker_build_platform_args("k3s")
|
||
assert result == ["--platform", "linux/arm64"]
|
||
|
||
@patch.dict(os.environ, {"PROLE_DOCKER_PLATFORM": ""}, clear=False)
|
||
@patch("knoe.config.is_apple_silicon", return_value=False)
|
||
def test_no_env_no_silicon(self, mock_as):
|
||
result = get_docker_build_platform_args("prod")
|
||
assert result == []
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# normalize_version (extended)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestNormalizeVersionExtended:
|
||
def test_dotted(self):
|
||
assert normalize_version("1.2.3") == "01.02.03"
|
||
|
||
def test_with_prefix(self):
|
||
assert normalize_version("v1.2.3") == "01.02.03"
|
||
|
||
def test_single_digit(self):
|
||
assert normalize_version("5") == "05.00.00"
|
||
|
||
def test_two_digits(self):
|
||
assert normalize_version("3.7") == "03.07.00"
|
||
|
||
def test_hyphenated(self):
|
||
assert normalize_version("17.7-043") == "17.07.43"
|
||
|
||
def test_large_numbers(self):
|
||
assert normalize_version("100.200.300") == "100.200.300"
|
||
|
||
def test_no_numbers(self):
|
||
assert normalize_version("abc") == "abc"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# get_resource_path
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestGetResourcePath:
|
||
def test_basic(self):
|
||
result = get_resource_path("etc/test.sh")
|
||
assert result.name == "test.sh"
|
||
assert isinstance(result, Path)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# setup_logging
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestSetupLogging:
|
||
def test_default(self):
|
||
# Should not raise
|
||
setup_logging()
|
||
|
||
def test_verbose(self):
|
||
setup_logging(verbose=True)
|
||
|
||
def test_debug(self, tmp_path):
|
||
# Debug creates a log file
|
||
setup_logging(debug=True)
|