"""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"]