"""Tests for installer.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 # --------------------------------------------------------------------------- # env.py helpers # --------------------------------------------------------------------------- from installer.core.env import ( _bool_str, _build_required_port_forwards, _clean_yaml_value, _cluster_env_radio_value, _collect_images_from_files, _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_prole_secret, _k3d_prole_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, _resolve_host_ip, _collect_cfg_vars as env_collect_cfg_vars, get_resource_path, is_apple_silicon, get_docker_build_platform_args, ) # --------------------------------------------------------------------------- # config.py helpers # --------------------------------------------------------------------------- from installer.config import ( _parse_bool as cfg_parse_bool, _expand_path as cfg_expand_path, _expand_cfg_value as cfg_expand_cfg_value, _collect_cfg_vars as cfg_collect_cfg_vars, _extract_yaml_scalar_from_text as cfg_extract_yaml, _extract_inline_vault_block as cfg_extract_vault, _load_properties, _write_k3s_kubeconfig, normalize_version, get_resource_path as cfg_get_resource_path, is_apple_silicon as cfg_is_apple_silicon, get_docker_build_platform_args as cfg_get_docker_args, _is_prole_secret as cfg_is_prole_secret, _is_openbao_ref as cfg_is_openbao_ref, _encrypt_cfg_secret, _resolve_secret_value, _update_prole_cfg_value, get_properties, get_config_value, setup_logging, ) # ===== _is_prole_secret / _is_openbao_ref ===== class TestSecretDetection: def test_is_prole_secret_true(self): assert _is_prole_secret("${PROLE_SECRET:abc}") is True def test_is_prole_secret_false(self): assert _is_prole_secret("plain") is False assert _is_prole_secret("") is False assert _is_prole_secret(None) is False def test_is_openbao_ref_true(self): assert _is_openbao_ref("${OPENBAO:kv/prole/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_prole_secret("${PROLE_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'), ('prole-dev-cluster', 'dev'), ('k3d-prole-dev-cluster', 'dev'), ('k3d-custom', 'dev'), ('service', 'service'), ('k3s', 'service'), ('k3s-service', 'service'), ('prole-service-cluster', 'service'), ('prole-service-x', 'service'), ('prod', 'prod'), ('production', 'prod'), ('k8s', 'prod'), ('prole-prod-cluster', 'prod'), ('prole-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') == 'prole-dev-cluster' assert _deployment_target_label('k3s') == 'prole-service-cluster' assert _deployment_target_label('k8s') == 'prole-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_prole_secret(self): assert _normalize_k3s_token('${PROLE_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/prole/myns/db#password}' def test_without_key(self): r = _openbao_placeholder('myns', 'db') assert r == '${OPENBAO:kv/prole/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 # ===== _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'] = 'prole-test' result = env_collect_cfg_vars(cfg) assert result['NS'] == 'prole-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_cfg_extract_vault(self): text = "secret: !vault |\n $ANSIBLE_VAULT;1.1;AES256\n data\nother: x" result = cfg_extract_vault(text, 'secret') assert '$ANSIBLE_VAULT' in result 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( "prole_domain: example.com\n" "prole_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'] # ===== _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_prole_data_volume_args ===== class TestK3dVolumeArgs: def test_empty(self): assert _k3d_prole_data_volume_args('') == [] assert _k3d_prole_data_volume_args(None) == [] def test_valid(self, tmp_path): result = _k3d_prole_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_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('installer.config.PROJECT_ROOT', tmp_path): 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('installer.config.PROJECT_ROOT', tmp_path): result = _write_k3s_kubeconfig('server:6443', 'tok') content = result.read_text() assert 'server: https://server:6443' in content # ===== _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_prole_cfg_value ===== class TestUpdateProleCfgValue: def test_creates_section(self, tmp_path): cfg_file = tmp_path / 'conf' / 'prole.cfg' cfg_file.parent.mkdir(parents=True) cfg_file.write_text("[Global]\nFOO = bar\n") with mock.patch('installer.config.PROJECT_ROOT', tmp_path): _update_prole_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('installer.config.PROJECT_ROOT', tmp_path): _update_prole_cfg_value('Sec', 'K', 'V') # should not raise # ===== 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_prole_cfg ===== class TestRenderProleCfg: def test_minimal(self): from installer.core.env import _render_prole_cfg result = _render_prole_cfg( inputs={'env_setup.PROLE_HOME': '/opt/prole', 'env_setup.NAMESPACE': 'test-ns'}, globals_to_save={'PROLE_HOME': '/opt/prole', 'NAMESPACE': 'test-ns'}, sections={}, generated_at='2025-01-01 00:00:00', ) assert '[Global]' in result assert 'PROLE_HOME' in result assert 'test-ns' in result or 'NAMESPACE' in result def test_with_sections(self): from installer.core.env import _render_prole_cfg result = _render_prole_cfg( inputs={ 'env_setup.PROLE_HOME': '/opt/prole', 'env_setup.PROLE_CONF': '/opt/prole/conf', 'env_setup.PROLE_DATA': '/opt/prole/data', 'env_setup.PROLE_LOGS': '/opt/prole/logs', 'env_setup.PROLE_SERVICE': '/opt/prole/etc', 'env_setup.NAMESPACE': 'myns', 'init_password.db_namespace': 'myns', 'init_password.db_username': 'prole', '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={'PROLE_HOME': '/opt/prole', 'NAMESPACE': 'myns'}, sections={ 'Database Creation': {'DB_USER': 'prole', 'NAMESPACE': 'myns'}, 'Docker Build': {}, 'Kerberos Authentication': {'ENABLED': 'false'}, }, ) assert '[Database Creation]' in result assert '[Inputs]' in result