From 312fb27df5c33ca25fc2dee29a1fb8665ca76f14 Mon Sep 17 00:00:00 2001 From: chrisfu Date: Fri, 20 Feb 2026 19:30:13 -0800 Subject: [PATCH] test: add unit tests for core helpers, milestone base, controller, and build - 141 tests for env.py/config.py pure functions (bool, cluster env, ollama, port-forwards, yaml, render_prole_cfg, etc.) - 48 tests for state, milestone base class, controller, build, and milestone subclasses - Combined coverage improved from 15% to 20% (2414/12349 stmts) --- tests/installer/test_core_classes.py | 422 +++++++++++++++ tests/installer/test_env_helpers.py | 741 +++++++++++++++++++++++++++ 2 files changed, 1163 insertions(+) create mode 100644 tests/installer/test_core_classes.py create mode 100644 tests/installer/test_env_helpers.py diff --git a/tests/installer/test_core_classes.py b/tests/installer/test_core_classes.py new file mode 100644 index 0000000..73eb14b --- /dev/null +++ b/tests/installer/test_core_classes.py @@ -0,0 +1,422 @@ +"""Tests for installer core classes: state, milestone, controller, build.""" +from __future__ import annotations + +import os +from pathlib import Path +from unittest import mock + +import pytest + +from installer.state import InstallerState +from installer.milestone import Milestone, ProgressCallback +from installer.build import get_build_command, BuildMilestone +from installer.core.controller import ProleController + + +# ===== InstallerState ===== + +class TestInstallerState: + def test_defaults(self): + s = InstallerState() + assert s.current_id is None + assert s.data == {} + assert s.inputs == {} + assert s.config_data == {} + assert s.errors == [] + assert s.progress == {} + assert s.history == [] + assert s.completed == set() + assert s.controller is None + + def test_mark_completed(self): + s = InstallerState() + s.mark_completed('step1') + s.mark_completed('step2') + assert 'step1' in s.completed + assert 'step2' in s.completed + + def test_mark_completed_idempotent(self): + s = InstallerState() + s.mark_completed('step1') + s.mark_completed('step1') + assert len(s.completed) == 1 + + def test_with_controller(self): + s = InstallerState(controller='fake') + assert s.controller == 'fake' + + def test_inputs_and_config(self): + s = InstallerState( + inputs={'key': 'val'}, + config_data={'section': {'k': 'v'}}, + ) + assert s.inputs['key'] == 'val' + assert s.config_data['section']['k'] == 'v' + + +# ===== Milestone base class ===== + +class _DummyMilestone(Milestone): + """Concrete subclass for testing the abstract base.""" + def execute(self, state, progress=None): + if progress: + progress("running", 0.5) + +class TestMilestoneBase: + def test_init(self): + m = _DummyMilestone('test_id', 'Test Title') + assert m.id == 'test_id' + assert m.title == 'Test Title' + + def test_validate_default(self): + m = _DummyMilestone('x', 'X') + assert m.validate(InstallerState()) is None + + def test_next_default(self): + m = _DummyMilestone('x', 'X') + assert m.next(InstallerState()) is None + + def test_execute_with_progress(self): + m = _DummyMilestone('x', 'X') + msgs = [] + m.execute(InstallerState(), progress=lambda msg, p: msgs.append((msg, p))) + assert ('running', 0.5) in msgs + + def test_parse_bool(self): + m = _DummyMilestone('x', 'X') + assert m._parse_bool('true') is True + assert m._parse_bool('false') is False + assert m._parse_bool(None) is False + assert m._parse_bool(None, default=True) is True + + def test_run_cmd_list(self): + m = _DummyMilestone('x', 'X') + rc = m._run_cmd(['echo', 'hello']) + assert rc == 0 + + def test_run_cmd_string(self): + m = _DummyMilestone('x', 'X') + rc = m._run_cmd('echo hello') + assert rc == 0 + + def test_run_cmd_with_stdout(self): + m = _DummyMilestone('x', 'X') + lines = [] + rc = m._run_cmd(['echo', 'test_output'], on_stdout=lambda l: lines.append(l.strip())) + assert rc == 0 + assert 'test_output' in lines + + def test_run_cmd_failure(self): + m = _DummyMilestone('x', 'X') + rc = m._run_cmd(['false']) + assert rc != 0 + + def test_run_cmd_bad_command(self): + m = _DummyMilestone('x', 'X') + rc = m._run_cmd(['/nonexistent/command']) + assert rc == 1 + + def test_get_script_env_basic(self): + m = _DummyMilestone('x', 'X') + state = InstallerState( + inputs={ + 'init_password.db_namespace': 'myns', + 'init_password.db_username': 'testuser', + 'init_password.db_password': 'secret123', + 'init_cluster.cluster_env': 'dev', + 'kerberos_config.realm': 'EXAMPLE.COM', + 'kerberos_config.kdc': '10.0.0.1', + 'kerberos_config.user': 'admin', + 'kerberos_config.password': 'krbpass', + 'kerberos_config.enabled': 'True', + 'init_cluster.at_rest_encryption_enabled': 'true', + }, + config_data={}, + ) + env = m._get_script_env(state) + assert env['NAMESPACE'] == 'myns' + assert env['DB_PASSWORD'] == 'secret123' + assert env['PROLE_DB_USER'] == 'testuser' + assert env['PROLE_MODE'] == 'k3d' + assert env['KRB5_REALM'] == 'EXAMPLE.COM' + assert env['KERBEROS_ENABLED'] == 'True' + assert env['AT_REST_ENCRYPTION_ENABLED'] == 'true' + assert env['REALM'] == 'EXAMPLE.COM' + assert env['DOMAIN'] == 'example.com' + + def test_get_script_env_defaults(self): + m = _DummyMilestone('x', 'X') + state = InstallerState(inputs={}, config_data={}) + env = m._get_script_env(state) + assert env['NAMESPACE'] == 'default' + assert env['PROLE_MODE'] == 'k3d' + assert env['KERBEROS_ENABLED'] == 'False' + + def test_get_script_env_grafana_from_config(self): + m = _DummyMilestone('x', 'X') + state = InstallerState( + inputs={'init_password.db_password': 'dbpw'}, + config_data={'Monitoring': {'GRAFANA_ADMIN_PASSWORD': 'grafanapw'}}, + ) + env = m._get_script_env(state) + assert env['GRAFANA_ADMIN_PASSWORD'] == 'grafanapw' + + def test_get_script_env_grafana_fallback(self): + m = _DummyMilestone('x', 'X') + state = InstallerState( + inputs={'init_password.db_password': 'dbpw'}, + config_data={}, + ) + env = m._get_script_env(state) + assert env['GRAFANA_ADMIN_PASSWORD'] == 'dbpw' + + def test_get_script_env_k3s_mode(self): + m = _DummyMilestone('x', 'X') + state = InstallerState( + inputs={ + 'init_cluster.cluster_env': 'service', + 'init_cluster.k3s_server_url': 'myserver:6443', + 'init_cluster.k3s_token': 'mytoken', + }, + config_data={}, + ) + env = m._get_script_env(state) + assert env['PROLE_MODE'] == 'k3s' + assert env['PROLE_K3S_SERVER'] == 'https://myserver:6443' + assert env['PROLE_K3S_TOKEN'] == 'mytoken' + + def test_get_script_env_argocd_from_config(self): + m = _DummyMilestone('x', 'X') + state = InstallerState( + inputs={}, + config_data={'Global': {'ARGOCD_NAMESPACE': 'custom-argocd', 'REGISTRY_NAMESPACE': 'custom-reg'}}, + ) + env = m._get_script_env(state) + assert env['ARGOCD_NAMESPACE'] == 'custom-argocd' + assert env['REGISTRY_NAMESPACE'] == 'custom-reg' + + +# ===== build.py ===== + +class TestBuildHelpers: + def test_get_build_command(self, tmp_path): + cmd = get_build_command(tmp_path, 'dev') + assert 'cd' in cmd + assert str(tmp_path) in cmd + assert '# Dev' in cmd + + def test_get_build_command_service(self, tmp_path): + cmd = get_build_command(tmp_path, 'service') + assert '# Service' in cmd + + def test_get_build_command_none_env(self, tmp_path): + cmd = get_build_command(tmp_path, None) + assert '# Dev' in cmd + + +class TestBuildMilestone: + def test_init(self, tmp_path): + m = BuildMilestone(tmp_path, 'dev', 'next_step') + assert m.id == 'build' + assert m.title == 'Build' + assert m._next_id == 'next_step' + + def test_validate_exists(self, tmp_path): + m = BuildMilestone(tmp_path) + assert m.validate(InstallerState()) is None + + def test_validate_missing(self): + m = BuildMilestone(Path('/nonexistent/path')) + result = m.validate(InstallerState()) + assert result is not None + assert len(result) > 0 + + def test_execute(self, tmp_path): + m = BuildMilestone(tmp_path, 'dev') + state = InstallerState() + msgs = [] + m.execute(state, progress=lambda msg, p: msgs.append(msg)) + assert 'build.command' in state.data + assert 'Build command prepared' in msgs + + def test_next(self, tmp_path): + m = BuildMilestone(tmp_path, next_id='deploy') + assert m.next(InstallerState()) == 'deploy' + + def test_next_none(self, tmp_path): + m = BuildMilestone(tmp_path) + assert m.next(InstallerState()) is None + + +# ===== ProleController ===== + +class TestProleController: + def test_init(self, tmp_path): + c = ProleController(tmp_path) + assert c.project_root == tmp_path + assert c.verbose is False + assert c.state is not None + + def test_init_verbose(self, tmp_path): + c = ProleController(tmp_path, verbose=True) + assert c.verbose is True + + def test_check_docker_running(self, tmp_path): + c = ProleController(tmp_path) + result = c.check_docker_running() + assert isinstance(result, bool) + + def test_get_prole_db_version_defaults(self, tmp_path): + c = ProleController(tmp_path) + version = c.get_prole_db_version() + assert '17.7' in version # default pg version + assert '-' in version + + def test_get_prole_db_version_custom(self, tmp_path): + pg_dir = tmp_path / 'conf' / 'postgresql' + pg_dir.mkdir(parents=True) + (pg_dir / '.version').write_text('16.2') + db_dir = tmp_path / 'prole-db' + db_dir.mkdir() + (db_dir / '.version').write_text('42') + c = ProleController(tmp_path) + version = c.get_prole_db_version() + assert version == '16.2-042' + + def test_run_milestones(self, tmp_path): + c = ProleController(tmp_path) + m1 = _DummyMilestone('step1', 'Step 1') + m2 = _DummyMilestone('step2', 'Step 2') + msgs = [] + c.run_milestones([m1, m2], progress_callback=lambda msg, p: msgs.append(msg)) + assert 'step1' in c.state.completed + assert 'step2' in c.state.completed + + def test_run_milestones_no_callback(self, tmp_path): + c = ProleController(tmp_path) + m = _DummyMilestone('s1', 'S1') + c.run_milestones([m]) + assert 's1' in c.state.completed + + def test_run_script_missing(self, tmp_path): + c = ProleController(tmp_path) + # Script doesn't exist but should not crash + rc = c.run_script('nonexistent_script.sh', args=[]) + # Will fail since the script doesn't exist + assert rc != 0 or rc is None + + +# ===== Milestone concrete subclasses (from core/milestones.py) ===== + +class TestDependenciesMilestone: + def test_init(self): + from installer.core.milestones import DependenciesMilestone + m = DependenciesMilestone() + assert m.id == 'dependencies' + assert m.title == 'Dependency Verification' + + +class TestNetworkScanMilestone: + def test_init(self): + from installer.core.milestones import NetworkScanMilestone + m = NetworkScanMilestone() + assert m.id == 'network_scan' + + +class TestEnvSetupMilestone: + def test_execute(self): + from installer.core.milestones import EnvSetupMilestone + m = EnvSetupMilestone() + state = InstallerState( + inputs={ + 'env_setup.PROLE_HOME': '/opt/prole', + 'env_setup.PROLE_CONF': '/opt/prole/conf', + 'env_setup.NAMESPACE': 'testns', + }, + config_data={}, + ) + msgs = [] + m.execute(state, progress=lambda msg, p: msgs.append(msg)) + assert state.config_data['System Environment']['PROLE_HOME'] == '/opt/prole' + assert state.config_data['System Environment']['NAMESPACE'] == 'testns' + assert 'Environment setup complete' in msgs + + +class TestSecretManagementMilestone: + def test_execute_plain(self): + from installer.core.milestones import SecretManagementMilestone + m = SecretManagementMilestone() + state = InstallerState( + inputs={ + 'init_password.db_password': 'plaintext123', + }, + ) + msgs = [] + m.execute(state, progress=lambda msg, p: msgs.append(msg)) + assert state.inputs['init_password.db_password_confirm'] == 'plaintext123' + + def test_execute_no_passwords(self): + from installer.core.milestones import SecretManagementMilestone + m = SecretManagementMilestone() + state = InstallerState(inputs={}) + m.execute(state) + + +class TestDatabaseCreationMilestone: + def test_execute_generates_password(self): + from installer.core.milestones import DatabaseCreationMilestone + m = DatabaseCreationMilestone() + state = InstallerState( + inputs={'env_setup.NAMESPACE': 'testns'}, + config_data={}, + ) + m.execute(state) + assert state.inputs.get('init_password.db_password') + assert len(state.inputs['init_password.db_password']) >= 20 + assert state.config_data['Database Creation']['DB_NAME'] == 'testns' + + def test_execute_with_password(self): + from installer.core.milestones import DatabaseCreationMilestone + m = DatabaseCreationMilestone() + state = InstallerState( + inputs={ + 'init_password.db_password': 'mypassword', + 'env_setup.NAMESPACE': 'ns1', + }, + config_data={}, + ) + m.execute(state) + assert state.inputs['init_password.db_password'] == 'mypassword' + + def test_execute_unresolvable_openbao(self): + from installer.core.milestones import DatabaseCreationMilestone + m = DatabaseCreationMilestone() + state = InstallerState( + inputs={ + 'init_password.db_password': '${OPENBAO:kv/prole/ns/db#password}', + 'env_setup.NAMESPACE': 'ns1', + }, + config_data={}, + ) + m.execute(state) + pw = state.inputs['init_password.db_password'] + assert not pw.startswith('${') + assert len(pw) >= 20 + + +class TestDockerBuildMilestone: + def test_disabled(self): + from installer.core.milestones import DockerBuildMilestone + m = DockerBuildMilestone() + state = InstallerState( + inputs={'init_db_build.run_build': 'False'}, + config_data={}, + ) + m.execute(state) # Should not raise + + +class TestDeploymentMilestone: + def test_init(self): + from installer.core.milestones import DeploymentMilestone + m = DeploymentMilestone() + assert m.id == 'deployment' diff --git a/tests/installer/test_env_helpers.py b/tests/installer/test_env_helpers.py new file mode 100644 index 0000000..8bb7799 --- /dev/null +++ b/tests/installer/test_env_helpers.py @@ -0,0 +1,741 @@ +"""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