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
b421f49af1
prole/tests/installer/test_env_helpers.py

1196 lines
39 KiB
Python
Raw Blame History

"""Tests for knoe.core.env pure utility functions."""
from __future__ import annotations
import configparser
import os
import tempfile
from pathlib import Path
from unittest import mock
import pytest
# ---------------------------------------------------------------------------
# config.py helpers
# ---------------------------------------------------------------------------
from knoe.config import (
_parse_bool as cfg_parse_bool,
_expand_path as cfg_expand_path,
_expand_cfg_value as cfg_expand_cfg_value,
_preserve_cfg_expr_for_persistence,
_collect_cfg_vars as cfg_collect_cfg_vars,
_filter_cfg_values_for_persistence,
_normalize_cfg_value_for_persistence,
_extract_yaml_scalar_from_text as cfg_extract_yaml,
_load_properties,
_write_k3s_kubeconfig,
_merge_kubeconfig,
normalize_version,
get_resource_path as cfg_get_resource_path,
get_docker_build_platform_args as cfg_get_docker_args,
_is_knoe_secret as cfg_is_knoe_secret,
_is_openbao_ref as cfg_is_openbao_ref,
_encrypt_cfg_secret,
_resolve_secret_value,
_update_knoe_cfg_value,
get_properties,
get_config_value,
setup_logging,
)
# ---------------------------------------------------------------------------
# env.py helpers
# ---------------------------------------------------------------------------
from knoe.core.env import (
_bool_str,
_build_required_port_forwards,
_clean_yaml_value,
_cluster_env_radio_value,
_collect_images_from_files,
_detect_ansible_k3s_settings,
_detect_ansible_storage_mounts,
_detect_ansible_node_labels,
_deployment_mode_from_env,
_deployment_target_label,
_expand_cfg_value,
_expand_path,
_extract_inline_vault_block,
_extract_yaml_scalar_from_text,
_find_kubeconfig_file,
_format_ollama_host,
_is_openbao_ref,
_is_knoe_secret,
_k3d_knoe_data_volume_args,
_local_registry_enabled,
_looks_like_k8s_bearer_token,
_normalize_cluster_env,
_normalize_k3s_token,
_openbao_placeholder,
_parse_ansible_inventory_hosts,
_parse_bool,
_parse_internal_a_records,
_parse_ollama_host,
_parse_yaml_scalar_values,
_pf_extract_id,
_pf_mapping_str,
_pf_upsert_mapping,
_registry_image_ref_exists,
_resolve_host_ip,
_collect_cfg_vars as env_collect_cfg_vars,
get_resource_path,
is_apple_silicon,
)
# ===== _is_knoe_secret / _is_openbao_ref =====
class TestSecretDetection:
def test_is_knoe_secret_true(self):
assert _is_knoe_secret("${KNOE_SECRET:abc}") is True
def test_is_knoe_secret_false(self):
assert _is_knoe_secret("plain") is False
assert _is_knoe_secret("") is False
assert _is_knoe_secret(None) is False
def test_is_openbao_ref_true(self):
assert _is_openbao_ref("${OPENBAO:kv/knoe/ns/db#password}") is True
def test_is_openbao_ref_false(self):
assert _is_openbao_ref("plain") is False
assert _is_openbao_ref("") is False
assert _is_openbao_ref(None) is False
def test_cfg_variants(self):
assert cfg_is_knoe_secret("${KNOE_SECRET:x}") is True
assert cfg_is_openbao_ref("${OPENBAO:x}") is True
# ===== _bool_str / _parse_bool =====
class TestBoolHelpers:
def test_bool_str(self):
assert _bool_str(True) == "true"
assert _bool_str(False) == "false"
assert _bool_str(1) == "true"
assert _bool_str(0) == "false"
@pytest.mark.parametrize(
"val,expected",
[
("true", True),
("True", True),
("1", True),
("yes", True),
("y", True),
("on", True),
("false", False),
("False", False),
("0", False),
("no", False),
("n", False),
("off", False),
(None, False),
("", False),
("maybe", False),
],
)
def test_parse_bool(self, val, expected):
assert _parse_bool(val) == expected
def test_parse_bool_default(self):
assert _parse_bool(None, default=True) is True
assert _parse_bool("junk", default=True) is True
def test_cfg_parse_bool(self):
assert cfg_parse_bool(True) is True
assert cfg_parse_bool(False) is False
assert cfg_parse_bool("true") is True
assert cfg_parse_bool("false") is False
assert cfg_parse_bool(None, default=True) is True
# ===== _normalize_cluster_env family =====
class TestClusterEnvNormalization:
@pytest.mark.parametrize(
"inp,expected",
[
("dev", "dev"),
("k3d", "dev"),
("k3d-dev", "dev"),
("knoe-dev-cluster", "dev"),
("k3d-knoe-dev-cluster", "dev"),
("k3d-custom", "dev"),
("service", "service"),
("k3s", "service"),
("k3s-service", "service"),
("knoe-service-cluster", "service"),
("knoe-service-x", "service"),
("prod", "prod"),
("production", "prod"),
("k8s", "prod"),
("knoe-prod-cluster", "prod"),
("knoe-prod-x", "prod"),
("", ""),
(None, ""),
("unknown", "unknown"),
],
)
def test_normalize_cluster_env(self, inp, expected):
assert _normalize_cluster_env(inp) == expected
def test_cluster_env_radio_value(self):
assert _cluster_env_radio_value("k3d") == "dev"
assert _cluster_env_radio_value("k3s") == "service"
assert _cluster_env_radio_value("k8s") == "prod"
assert _cluster_env_radio_value("unknown") == "unknown"
assert _cluster_env_radio_value(None) == ""
def test_deployment_target_label(self):
assert _deployment_target_label("dev") == "knoe-dev-cluster"
assert _deployment_target_label("k3s") == "knoe-service-cluster"
assert _deployment_target_label("k8s") == "knoe-prod-cluster"
assert _deployment_target_label(None) == ""
def test_deployment_mode_from_env(self):
assert _deployment_mode_from_env("dev") == "k3d"
assert _deployment_mode_from_env("service") == "k3s"
assert _deployment_mode_from_env("prod") == "k8s"
assert _deployment_mode_from_env("") == ""
assert _deployment_mode_from_env(None) == ""
# ===== _normalize_k3s_token / _looks_like_k8s_bearer_token =====
class TestTokenHelpers:
def test_normalize_k3s_token_plain(self):
assert _normalize_k3s_token("abc123") == "abc123"
def test_normalize_k3s_token_empty(self):
assert _normalize_k3s_token("") == ""
assert _normalize_k3s_token(None) == ""
assert _normalize_k3s_token(" ") == ""
def test_normalize_k3s_token_openbao(self):
assert _normalize_k3s_token("${OPENBAO:kv/x}") == ""
def test_normalize_k3s_token_knoe_secret(self):
assert _normalize_k3s_token("${KNOE_SECRET:junk}") == ""
def test_looks_like_bearer_jwt(self):
assert _looks_like_k8s_bearer_token("a.b.c") is True
def test_looks_like_bearer_bootstrap(self):
assert _looks_like_k8s_bearer_token("abcdef.1234567890abcdef") is True
def test_not_bearer_node_token(self):
assert _looks_like_k8s_bearer_token("K1::node::xyz") is False
def test_not_bearer_empty(self):
assert _looks_like_k8s_bearer_token("") is False
assert _looks_like_k8s_bearer_token(None) is False
def test_not_bearer_short_parts(self):
assert _looks_like_k8s_bearer_token("a.b") is False
# ===== _parse_ollama_host / _format_ollama_host =====
class TestOllamaHelpers:
def test_parse_ollama_host_basic(self):
host, port = _parse_ollama_host("http://myhost:1234")
assert host == "myhost"
assert port == "1234"
def test_parse_ollama_host_no_scheme(self):
host, port = _parse_ollama_host("myhost:5555")
assert host == "myhost"
assert port == "5555"
def test_parse_ollama_host_default_port(self):
host, port = _parse_ollama_host("http://myhost")
assert host == "myhost"
assert port == "11434"
def test_parse_ollama_host_empty(self):
assert _parse_ollama_host("") == ("", "")
def test_format_ollama_host(self):
assert _format_ollama_host("myhost", "1234") == "http://myhost:1234"
def test_format_ollama_host_empty(self):
assert _format_ollama_host("", "1234") == ""
def test_format_ollama_host_https(self):
assert _format_ollama_host("https://myhost", "443") == "https://myhost:443"
def test_format_ollama_host_no_port(self):
assert _format_ollama_host("myhost", "") == "http://myhost"
# ===== _openbao_placeholder =====
class TestOpenbaoPlaceholder:
def test_with_key(self):
r = _openbao_placeholder("myns", "db", "password")
assert r == "${OPENBAO:kv/knoe/myns/db#password}"
def test_without_key(self):
r = _openbao_placeholder("myns", "db")
assert r == "${OPENBAO:kv/knoe/myns/db}"
def test_default_namespace(self):
r = _openbao_placeholder("", "db", "pw")
assert "default" in r
# ===== _expand_path / _expand_cfg_value =====
class TestExpandHelpers:
def test_expand_path_none(self):
assert _expand_path(None) == ""
def test_expand_path_tilde(self):
result = _expand_path("~/test")
assert "~" not in result
def test_expand_cfg_value_none(self):
assert _expand_cfg_value(None) == ""
def test_expand_cfg_value_openbao(self):
val = "${OPENBAO:x}"
assert _expand_cfg_value(val) == val
def test_expand_cfg_value_vars(self):
assert (
_expand_cfg_value("${HOME}/data", cfg_vars={"HOME": "/opt"}) == "/opt/data"
)
def test_expand_cfg_value_recursive(self):
assert _expand_cfg_value("$A", cfg_vars={"A": "$B", "B": "final"}) == "final"
def test_cfg_expand_path(self):
assert cfg_expand_path(None) == ""
assert cfg_expand_path("") == ""
def test_cfg_expand_cfg_value(self):
assert cfg_expand_cfg_value("${X}", {"X": "hello"}) == "hello"
assert cfg_expand_cfg_value("", {}) == ""
assert cfg_expand_cfg_value(None, {}) is None
def test_preserve_cfg_expr_for_persistence_keeps_home_token(self):
value = _preserve_cfg_expr_for_persistence(
"${HOME}/dev/knoe/data/staging",
"/Users/chrisfu/dev/knoe/data/staging",
)
assert value == "${HOME}/dev/knoe/data/staging"
def test_preserve_cfg_expr_for_persistence_uses_expanded_without_tokens(self):
value = _preserve_cfg_expr_for_persistence(
"/Users/chrisfu/dev/knoe/data/staging",
"/Users/chrisfu/dev/knoe/data/staging",
)
assert value == "/Users/chrisfu/dev/knoe/data/staging"
# ===== _collect_cfg_vars =====
class TestCollectCfgVars:
def test_env_collect(self):
cfg = configparser.ConfigParser(interpolation=None)
cfg.optionxform = str
cfg.add_section("Global")
cfg["Global"]["NS"] = "knoe-test"
result = env_collect_cfg_vars(cfg)
assert result["NS"] == "knoe-test"
def test_cfg_collect(self):
cfg = configparser.ConfigParser(interpolation=None)
cfg.optionxform = str
cfg.add_section("Global")
cfg["Global"]["FOO"] = "bar"
cfg.add_section("System Environment")
cfg["System Environment"]["BAZ"] = "qux"
result = cfg_collect_cfg_vars(cfg)
assert "FOO" in result or "foo" in result
assert "BAZ" in result or "baz" in result
# ===== _clean_yaml_value =====
class TestCleanYamlValue:
def test_plain(self):
assert _clean_yaml_value(" hello ") == "hello"
def test_quoted(self):
assert _clean_yaml_value('"hello"') == "hello"
assert _clean_yaml_value("'hello'") == "hello"
def test_comment(self):
assert _clean_yaml_value("val # comment") == "val"
def test_none(self):
assert _clean_yaml_value(None) == ""
# ===== YAML helpers =====
class TestYamlHelpers:
def test_extract_yaml_scalar(self):
text = "name: test\nport: 8080\n# comment\n"
assert _extract_yaml_scalar_from_text(text, "name") == "test"
assert _extract_yaml_scalar_from_text(text, "port") == "8080"
assert _extract_yaml_scalar_from_text(text, "missing") == ""
def test_cfg_extract_yaml(self):
text = 'key: "value"\nother: plain'
assert cfg_extract_yaml(text, "key") == "value"
assert cfg_extract_yaml(text, "missing") == ""
def test_extract_inline_vault_block(self):
text = "secret: !vault |\n $ANSIBLE_VAULT;1.1;AES256\n 6162\nnext_key: val"
result = _extract_inline_vault_block(text, "secret")
assert "$ANSIBLE_VAULT" in result
def test_extract_inline_vault_no_match(self):
assert _extract_inline_vault_block("key: value", "key") == ""
assert _extract_inline_vault_block("key: value", "missing") == ""
def test_parse_yaml_scalar_values(self, tmp_path):
f = tmp_path / "test.yaml"
f.write_text("---\nname: test\nport: 8080\n# comment\n- item\n")
result = _parse_yaml_scalar_values(f, {"name", "port"})
assert result["name"] == "test"
assert result["port"] == "8080"
def test_parse_yaml_scalar_missing_file(self, tmp_path):
f = tmp_path / "missing.yaml"
assert _parse_yaml_scalar_values(f, {"key"}) == {}
# ===== _parse_internal_a_records =====
class TestParseInternalARecords:
def test_basic(self, tmp_path):
f = tmp_path / "dns.yml"
f.write_text(
"knoe_domain: example.com\n"
"knoe_internal_a_records:\n"
" - fqdn: host1.example.com\n"
" ipv4: 10.0.0.1\n"
" - fqdn: host2.example.com\n"
" ipv4: 10.0.0.2\n"
)
domain, ip_map, fqdn_map = _parse_internal_a_records(f)
assert domain == "example.com"
assert fqdn_map["host1.example.com"] == "10.0.0.1"
assert "host1" in ip_map
def test_missing_file(self, tmp_path):
f = tmp_path / "missing.yml"
domain, ip_map, fqdn_map = _parse_internal_a_records(f)
assert domain == ""
assert ip_map == {}
# ===== _parse_ansible_inventory_hosts =====
class TestParseInventory:
def test_basic(self, tmp_path):
f = tmp_path / "hosts"
f.write_text("[servers]\nhost1\nhost2\n\n[workers]\nhost3\n")
groups = _parse_ansible_inventory_hosts(f)
assert "servers" in groups
assert "host1" in groups["servers"]
assert "host3" in groups["workers"]
def test_missing(self, tmp_path):
f = tmp_path / "missing"
assert _parse_ansible_inventory_hosts(f) == {}
def test_comments(self, tmp_path):
f = tmp_path / "hosts"
f.write_text("# comment\n[grp]\n; comment\nhost1\n")
groups = _parse_ansible_inventory_hosts(f)
assert groups["grp"] == ["host1"]
class TestDetectAnsibleK3sSettings:
def test_children_groups_fallback_to_servers(self, tmp_path):
inventory_path = tmp_path / "inventory"
(inventory_path / "host_vars").mkdir(parents=True)
# Host vars are intentionally empty; the detection should still infer a
# server host from the k3s_servers group.
(inventory_path / "host_vars" / "myrddin.knoe.org.yml").write_text("\n")
(inventory_path / "host_vars" / "pi.knoe.org.yml").write_text("\n")
groups = {
"k3s_hosts:children": ["k3s_servers", "k3s_agents"],
"k3s_servers": ["myrddin.knoe.org"],
"k3s_agents": ["pi.knoe.org"],
}
res = _detect_ansible_k3s_settings(
inventory_path=inventory_path, groups=groups, domain="knoe.org", ip_map={}
)
assert res["server_host"] == "myrddin.knoe.org"
assert res["server_url"] == "https://myrddin.knoe.org:6443"
class TestDetectAnsibleStorageMounts:
def test_parses_iscsi_mounts_from_host_vars(self, tmp_path):
inventory = tmp_path / "inventory"
hv = inventory / "host_vars"
hv.mkdir(parents=True)
(hv / "myrddin.knoe.org.yml").write_text(
"""
iscsi_targets:
- name: syno
mounts:
- name: d001
path: /synology/d001
k3s_service_node_labels:
- knoe.org/node-role=db
- knoe.org/storage=synology
""".lstrip()
)
(hv / "pi.knoe.org.yml").write_text(
"""
iscsi_targets:
- name: syno
mounts:
- name: d003
path: /synology/d003
""".lstrip()
)
mounts = _detect_ansible_storage_mounts(inventory)
assert mounts["d001"]["host"] == "myrddin.knoe.org"
assert mounts["d001"]["path"] == "/synology/d001"
assert mounts["d003"]["host"] == "pi.knoe.org"
assert mounts["d003"]["path"] == "/synology/d003"
labels = _detect_ansible_node_labels(inventory)
assert labels["myrddin.knoe.org"]["knoe.org/node-role"] == "db"
assert labels["myrddin.knoe.org"]["knoe.org/storage"] == "synology"
# ===== _resolve_host_ip =====
class TestResolveHostIp:
def test_direct(self):
assert _resolve_host_ip("h1", "example.com", {"h1": "1.2.3.4"}) == "1.2.3.4"
def test_fqdn(self):
assert (
_resolve_host_ip("h1", "example.com", {"h1.example.com": "1.2.3.4"})
== "1.2.3.4"
)
def test_short_from_fqdn(self):
assert (
_resolve_host_ip("h1.example.com", "example.com", {"h1": "1.2.3.4"})
== "1.2.3.4"
)
def test_empty(self):
assert _resolve_host_ip("", "", {}) == ""
# ===== Port-forward helpers =====
class TestPortForwardHelpers:
def test_pf_extract_id(self):
assert _pf_extract_id("id=grafana;namespace=monitoring") == "grafana"
assert _pf_extract_id("namespace=x") == ""
assert _pf_extract_id("") == ""
def test_pf_mapping_str(self):
s = _pf_mapping_str("test", "ns", "svc/mysvc", 8080, 80)
assert "id=test" in s
assert "namespace=ns" in s
assert "hostPort=8080" in s
def test_pf_upsert_new(self):
section = {}
prefix = "PF_"
mapping = "id=grafana;namespace=monitoring;hostPort=3000;servicePort=80"
assert _pf_upsert_mapping(section, prefix, mapping) is True
assert "PF_1" in section
def test_pf_upsert_update(self):
section = {
"PF_1": "id=grafana;namespace=monitoring;hostPort=3000;servicePort=80"
}
new = "id=grafana;namespace=monitoring;hostPort=3001;servicePort=80"
assert _pf_upsert_mapping(section, "PF_", new) is True
assert section["PF_1"] == new
def test_pf_upsert_duplicate(self):
mapping = "id=grafana;namespace=monitoring;hostPort=3000;servicePort=80"
section = {"PF_1": mapping}
assert _pf_upsert_mapping(section, "PF_", mapping) is False
def test_pf_upsert_none_section(self):
assert _pf_upsert_mapping(None, "PF_", "id=x") is False
def test_pf_upsert_no_id(self):
assert _pf_upsert_mapping({}, "PF_", "no-id-here") is False
# ===== _build_required_port_forwards =====
class TestBuildPortForwards:
def test_basic_k3d(self):
mappings = _build_required_port_forwards(
"k3d", "default", "argocd", "myns", "5432", False, "supabase"
)
assert len(mappings) >= 8
ids = [_pf_extract_id(m) for m in mappings]
assert "argocd" in ids
assert "grafana" in ids
assert "postgres" in ids
assert "openbao" in ids
def test_with_supabase(self):
mappings = _build_required_port_forwards(
"k3d", "default", "argocd", "myns", "5432", True, "supabase"
)
ids = [_pf_extract_id(m) for m in mappings]
assert "supabase-kong" in ids
assert "supabase-studio" in ids
def test_defaults(self):
mappings = _build_required_port_forwards("k3d", "", "", "", "", False, "")
assert len(mappings) >= 8
# ===== _k3d_knoe_data_volume_args =====
class TestK3dVolumeArgs:
def test_empty(self):
assert _k3d_knoe_data_volume_args("") == []
assert _k3d_knoe_data_volume_args(None) == []
def test_valid(self, tmp_path):
result = _k3d_knoe_data_volume_args(str(tmp_path / "data"))
assert result[0] == "--volume"
assert "storage@all" in result[1]
# ===== _find_kubeconfig_file =====
class TestFindKubeconfig:
def test_from_env(self, tmp_path):
kc = tmp_path / "kubeconfig"
kc.write_text("test")
env = {"KUBECONFIG": str(kc)}
assert _find_kubeconfig_file(env) == str(kc)
def test_prefers_project_root_knoe_k3s_over_etc_secrets(
self, tmp_path, monkeypatch
):
# If both exist, prefer the Ansible-fetched kubeconfig in project root.
import knoe.core.env as env_mod
fake_home = tmp_path / "home"
fake_home.mkdir(parents=True, exist_ok=True)
monkeypatch.setattr(Path, "home", classmethod(lambda cls: fake_home))
fake_root = tmp_path / "root"
(fake_root / "etc" / "secrets").mkdir(parents=True, exist_ok=True)
etc_kc = fake_root / "etc" / "secrets" / "k3s.kubeconfig"
etc_kc.write_text("etc")
knoe_kc = fake_root / "knoe-k3s.kubeconfig"
knoe_kc.write_text("root")
monkeypatch.setattr(env_mod, "PROJECT_ROOT", fake_root)
# Must pass a non-empty mapping; `_find_kubeconfig_file({})` falls back
# to `os.environ` because an empty dict is falsy.
assert _find_kubeconfig_file({"KUBECONFIG": ""}) == str(knoe_kc)
def test_none_found(self):
env = {"KUBECONFIG": "/nonexistent/path"}
result = _find_kubeconfig_file(env)
# May return '' or another found path
assert isinstance(result, str)
# ===== _local_registry_enabled =====
class TestLocalRegistryEnabled:
def test_explicit_env(self):
with mock.patch.dict(
os.environ, {"PROLE_ENABLE_LOCAL_REGISTRY": "true"}, clear=False
):
assert _local_registry_enabled() is True
def test_mode_hint(self):
with mock.patch.dict(os.environ, {}, clear=False):
os.environ.pop("PROLE_ENABLE_LOCAL_REGISTRY", None)
os.environ.pop("ENABLE_LOCAL_REGISTRY", None)
assert _local_registry_enabled("k3d") is True
assert _local_registry_enabled("k3s") is True
def test_explicit_false(self):
with mock.patch.dict(
os.environ, {"PROLE_ENABLE_LOCAL_REGISTRY": "false"}, clear=False
):
assert _local_registry_enabled() is False
# ===== _collect_images_from_files =====
class TestCollectImages:
def test_basic(self, tmp_path):
f = tmp_path / "deploy.yaml"
f.write_text("spec:\n image: nginx:latest\n image: postgres:16\n")
result = _collect_images_from_files([f])
assert "nginx:latest" in result
assert "postgres:16" in result
def test_skip_vars(self, tmp_path):
f = tmp_path / "deploy.yaml"
f.write_text(" image: ${REGISTRY}/app:latest\n")
result = _collect_images_from_files([f])
assert len(result) == 0
def test_missing_file(self, tmp_path):
result = _collect_images_from_files([tmp_path / "missing.yaml"])
assert result == set()
# ===== get_resource_path =====
class TestGetResourcePath:
def test_returns_path(self):
p = get_resource_path("img/test.png")
assert isinstance(p, Path)
def test_cfg_variant(self):
p = cfg_get_resource_path("img/test.png")
assert isinstance(p, Path)
# ===== normalize_version =====
class TestNormalizeVersion:
def test_semver(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_short(self):
assert normalize_version("5") == "05.00.00"
def test_empty(self):
assert normalize_version("") == ""
def test_no_numbers(self):
assert normalize_version("abc") == "abc"
# ===== _load_properties =====
class TestLoadProperties:
def test_basic(self, tmp_path):
f = tmp_path / "test.properties"
f.write_text("# comment\nkey1=val1\nkey2 = val2\n\n")
result = _load_properties(f)
assert result["key1"] == "val1"
assert result["key2"] == "val2"
def test_missing(self, tmp_path):
assert _load_properties(tmp_path / "missing") == {}
# ===== _write_k3s_kubeconfig =====
class TestWriteK3sKubeconfig:
def test_basic(self, tmp_path):
with mock.patch("knoe.config.PROJECT_ROOT", tmp_path), mock.patch.dict(
os.environ, {"KNOE_SERVICE": "", "KNOE_HOME": ""}, clear=False
):
result = _write_k3s_kubeconfig("https://server:6443", "mytoken")
assert result.exists()
content = result.read_text()
assert "server: https://server:6443" in content
assert "mytoken" in content
def test_no_scheme(self, tmp_path):
with mock.patch("knoe.config.PROJECT_ROOT", tmp_path), mock.patch.dict(
os.environ, {"KNOE_SERVICE": "", "KNOE_HOME": ""}, clear=False
):
result = _write_k3s_kubeconfig("server:6443", "tok")
content = result.read_text()
assert "server: https://server:6443" in content
# ===== _merge_kubeconfig =====
class TestMergeKubeconfig:
def test_merge_into_empty_home(self, tmp_path):
"""Merge a standalone kubeconfig when no ~/.kube/config exists yet."""
kube_dir = tmp_path / ".kube"
standalone = tmp_path / "standalone.yaml"
standalone.write_text(
"apiVersion: v1\nkind: Config\n"
"clusters:\n- cluster:\n server: https://srv:6443\n"
" insecure-skip-tls-verify: true\n name: knoe-k3s\n"
"contexts:\n- context:\n cluster: knoe-k3s\n user: knoe-k3s\n"
" name: knoe-k3s\ncurrent-context: knoe-k3s\n"
"users:\n- name: knoe-k3s\n user:\n token: tok123\n"
)
with mock.patch("knoe.config.Path.home", return_value=tmp_path):
result = _merge_kubeconfig(str(standalone))
if result:
merged = (kube_dir / "config").read_text()
assert "knoe-k3s" in merged
assert "tok123" in merged
def test_merge_adds_context_to_existing(self, tmp_path):
"""Merge adds the knoe-k3s context alongside an existing context."""
kube_dir = tmp_path / ".kube"
kube_dir.mkdir()
existing = kube_dir / "config"
existing.write_text(
"apiVersion: v1\nkind: Config\n"
"clusters:\n- cluster:\n server: https://other:6443\n name: other-cluster\n"
"contexts:\n- context:\n cluster: other-cluster\n user: other-user\n"
" name: other-ctx\ncurrent-context: other-ctx\n"
"users:\n- name: other-user\n user:\n token: othertoken\n"
)
standalone = tmp_path / "standalone.yaml"
standalone.write_text(
"apiVersion: v1\nkind: Config\n"
"clusters:\n- cluster:\n server: https://srv:6443\n"
" insecure-skip-tls-verify: true\n name: knoe-k3s\n"
"contexts:\n- context:\n cluster: knoe-k3s\n user: knoe-k3s\n"
" name: knoe-k3s\ncurrent-context: knoe-k3s\n"
"users:\n- name: knoe-k3s\n user:\n token: tok123\n"
)
with mock.patch("knoe.config.Path.home", return_value=tmp_path):
result = _merge_kubeconfig(str(standalone))
if result:
merged = existing.read_text()
assert "knoe-k3s" in merged
assert "other-ctx" in merged
def test_merge_returns_false_when_kubectl_missing(self, tmp_path):
"""Returns False when kubectl is not available."""
standalone = tmp_path / "standalone.yaml"
standalone.write_text("apiVersion: v1\nkind: Config\n")
with mock.patch(
"knoe.config.Path.home", return_value=tmp_path
), mock.patch("subprocess.run", side_effect=FileNotFoundError):
result = _merge_kubeconfig(str(standalone))
assert result is False
# ===== _encrypt_cfg_secret / _resolve_secret_value =====
class TestSecretEncryption:
def test_encrypt_empty(self):
assert _encrypt_cfg_secret("") == ""
assert _encrypt_cfg_secret(None) == ""
def test_encrypt_openbao_passthrough(self):
val = "${OPENBAO:kv/x}"
assert _encrypt_cfg_secret(val) == val
def test_resolve_plain(self):
assert _resolve_secret_value("hello") == "hello"
# ===== setup_logging =====
class TestSetupLogging:
def test_default(self):
setup_logging()
def test_verbose(self):
setup_logging(verbose=True)
def test_debug(self):
setup_logging(debug=True)
# ===== get_properties / get_config_value =====
class TestProperties:
def test_get_properties(self):
props = get_properties()
assert isinstance(props, dict)
def test_get_config_value_default(self):
assert get_config_value("nonexistent_key_xyz", "fallback") == "fallback"
# ===== _update_knoe_cfg_value =====
class TestUpdateKnoeCfgValue:
def test_creates_section(self, tmp_path):
cfg_file = tmp_path / "conf" / "knoe.cfg"
cfg_file.parent.mkdir(parents=True)
cfg_file.write_text("[Global]\nFOO = bar\n")
with mock.patch("knoe.config.PROJECT_ROOT", tmp_path):
_update_knoe_cfg_value("NewSection", "KEY", "VALUE")
content = cfg_file.read_text()
assert "NewSection" in content
assert "KEY" in content
def test_missing_cfg(self, tmp_path):
with mock.patch("knoe.config.PROJECT_ROOT", tmp_path):
_update_knoe_cfg_value("Sec", "K", "V") # should not raise
def test_writes_through_knoe_conf_symlink(self, tmp_path):
conf_dir = tmp_path / "knoe-conf"
dev_dir = conf_dir / "dev"
svc_dir = conf_dir / "service"
dev_dir.mkdir(parents=True)
svc_dir.mkdir(parents=True)
dev_base = dev_dir / "knoe.cfg"
svc_base = svc_dir / "knoe.cfg"
dev_base.write_text("[Global]\nFOO = base\n")
svc_base.write_text("[Global]\nFOO = service\n")
entry = conf_dir / "knoe.cfg"
os.symlink(str(dev_base), str(entry))
with mock.patch.dict(os.environ, {"KNOE_CONF": str(conf_dir)}):
_update_knoe_cfg_value("Global", "FOO", "updated")
assert "FOO = updated" in dev_base.read_text()
assert svc_base.read_text() == "[Global]\nFOO = service\n"
def test_service_mode_drops_placeholder_endpoint_value(self, tmp_path):
cfg_file = tmp_path / "conf" / "knoe.cfg"
cfg_file.parent.mkdir(parents=True)
cfg_file.write_text("[Global]\nSERVICE_HOSTNAME = old-host\n")
with mock.patch("knoe.config.PROJECT_ROOT", tmp_path):
_update_knoe_cfg_value(
"Global",
"SERVICE_HOSTNAME",
"${SERVICE_HOSTNAME}",
mode="k3s",
)
content = cfg_file.read_text()
assert "SERVICE_HOSTNAME" not in content
class TestPersistenceFilters:
def test_service_mode_drops_generated_endpoints_and_localhost(self):
values = {
"API_SERVICE_ENDPOINT": "http://localhost:8081",
"SERVICE_HOSTNAME": "k3d.localhost",
"PROLE_K3S_SERVER": "https://myrddin.knoe.org:6443",
"SERVICE_NAMESPACE": "knoe-system",
}
filtered = _filter_cfg_values_for_persistence(
"Global",
values,
mode="k3s",
explicit_keys={"PROLE_K3S_SERVER"},
)
assert filtered == {"PROLE_K3S_SERVER": "https://myrddin.knoe.org:6443"}
def test_service_mode_keeps_explicit_service_namespace(self):
filtered = _filter_cfg_values_for_persistence(
"Global",
{"SERVICE_NAMESPACE": "knoe-system"},
mode="k3s",
explicit_keys={"SERVICE_NAMESPACE"},
)
assert filtered == {"SERVICE_NAMESPACE": "knoe-system"}
def test_dev_mode_keeps_localhost_endpoint(self):
value = _normalize_cfg_value_for_persistence(
"Global",
"API_SERVICE_ENDPOINT",
"http://localhost:8081",
mode="k3d",
)
assert value == "http://localhost:8081"
# ===== is_apple_silicon / get_docker_build_platform_args =====
class TestPlatformHelpers:
def test_is_apple_silicon(self):
result = is_apple_silicon()
assert isinstance(result, bool)
def test_docker_args_env_override(self):
with mock.patch.dict(os.environ, {"PROLE_DOCKER_PLATFORM": "linux/arm64"}):
args = cfg_get_docker_args()
assert args == ["--platform", "linux/arm64"]
def test_docker_args_dev(self):
with mock.patch.dict(os.environ, {}, clear=False):
os.environ.pop("PROLE_DOCKER_PLATFORM", None)
args = cfg_get_docker_args("dev")
assert isinstance(args, list)
# ===== _render_knoe_cfg =====
class TestRenderKnoeCfg:
def test_minimal(self):
from knoe.core.env import _render_knoe_cfg
result = _render_knoe_cfg(
inputs={
"env_setup.KNOE_HOME": "/opt/knoe",
"env_setup.NAMESPACE": "test-ns",
},
globals_to_save={"KNOE_HOME": "/opt/knoe", "NAMESPACE": "test-ns"},
sections={},
generated_at="2025-01-01 00:00:00",
)
assert "[Global]" in result
assert "KNOE_HOME" in result
assert "test-ns" in result or "NAMESPACE" in result
def test_with_sections(self):
from knoe.core.env import _render_knoe_cfg
result = _render_knoe_cfg(
inputs={
"env_setup.KNOE_HOME": "/opt/knoe",
"env_setup.KNOE_CONF": "/opt/knoe/conf",
"env_setup.PROLE_DATA": "/opt/knoe/data",
"env_setup.PROLE_LOGS": "/opt/knoe/logs",
"env_setup.KNOE_SERVICE": "/opt/knoe/etc",
"env_setup.NAMESPACE": "myns",
"init_password.db_namespace": "myns",
"init_password.db_username": "knoe",
"init_cluster.cluster_env": "dev",
"init_cluster.supabase_enabled": "false",
"init_cluster.kerberos_enabled": "false",
"init_cluster.at_rest_encryption_enabled": "false",
},
globals_to_save={"KNOE_HOME": "/opt/knoe", "NAMESPACE": "myns"},
sections={
"Database Creation": {"DB_USER": "knoe", "NAMESPACE": "myns"},
"Docker Build": {},
"Kerberos Authentication": {"ENABLED": "false"},
},
)
assert "[Database Creation]" in result
assert "[Inputs]" in result
def test_does_not_inject_placeholder_filesystem_paths(self):
"""New policy: knoe.cfg should not introduce `${...}` placeholders for filesystem paths."""
from knoe.core.env import _render_knoe_cfg
result = _render_knoe_cfg(
inputs={
# Simulate an operator/session where KNOE_HOME is known, but we
# do not want the renderer to normalize filesystem paths into
# placeholders.
"env_setup.KNOE_HOME": "",
"env_setup.KNOE_CONF": "",
"env_setup.PROLE_DATA": "",
"env_setup.PROLE_LOGS": "",
"env_setup.KNOE_SERVICE": "",
"env_setup.NAMESPACE": "test-ns",
},
globals_to_save={"KNOE_HOME": "/opt/knoe", "NAMESPACE": "test-ns"},
sections={},
generated_at="2025-01-01 00:00:00",
)
# Must not emit placeholder filesystem paths like `${KNOE_HOME}`.
assert "${KNOE_HOME}" not in result
assert "${KNOE_CONF}" not in result
assert "${PROLE_DATA}" not in result
assert "${PROLE_LOGS}" not in result
assert "${KNOE_SERVICE}" not in result
# Must not inject derived filesystem keys that rely on placeholder expansion.
assert "ANSIBLE_INFRASTRUCTURE" not in result
assert "ANSIBLE_INVENTORY" not in result
class TestRegistryManifestProbe:
def test_invalid_ref_returns_false(self):
assert _registry_image_ref_exists("") is False
assert _registry_image_ref_exists("alpine:latest") is False
def test_head_200_true_and_no_get(self, monkeypatch):
import http.client
requests: list[tuple[str, str]] = []
class _Resp:
def __init__(self, status: int):
self.status = status
def read(self):
return b""
class _Conn:
def __init__(self, host, port, timeout=None):
self.host = host
self.port = port
self.timeout = timeout
self._method = None
self._path = None
def request(self, method, path, headers=None):
requests.append((method, path))
self._method = method
self._path = path
def getresponse(self):
assert self._method is not None
return _Resp(200)
def close(self):
return None
monkeypatch.setattr(http.client, "HTTPConnection", _Conn)
assert _registry_image_ref_exists("myrddin.knoe.org:5000/knoe-db:latest") is True
assert requests and requests[0][0] == "HEAD"
assert all(m != "GET" for m, _ in requests)
def test_head_404_false(self, monkeypatch):
import http.client
class _Resp:
def __init__(self, status: int):
self.status = status
def read(self):
return b""
class _Conn:
def __init__(self, host, port, timeout=None):
self._method = None
def request(self, method, path, headers=None):
self._method = method
def getresponse(self):
return _Resp(404)
def close(self):
return None
monkeypatch.setattr(http.client, "HTTPConnection", _Conn)
assert _registry_image_ref_exists("myrddin.knoe.org:5000/knoe-db:latest") is False
def test_head_405_falls_back_to_get(self, monkeypatch):
import http.client
requests: list[str] = []
class _Resp:
def __init__(self, status: int):
self.status = status
def read(self):
return b""
class _Conn:
def __init__(self, host, port, timeout=None):
self._method = None
def request(self, method, path, headers=None):
self._method = method
requests.append(method)
def getresponse(self):
assert self._method is not None
if self._method == "HEAD":
return _Resp(405)
return _Resp(200)
def close(self):
return None
monkeypatch.setattr(http.client, "HTTPConnection", _Conn)
assert _registry_image_ref_exists("myrddin.knoe.org:5000/knoe-db:latest") is True
assert requests == ["HEAD", "GET"]
Reference in New Issue View Git Blame Copy Permalink