mirror of
https://github.com/dredx/prole.git
synced 2026-09-27 12:44:31 +00:00
1573bb59c2
- Include new `prole.spec` for build configurations and dependencies. - Add Terraform state handling for OpenTofu in `k3s` cluster. - Provision multiple Kubernetes resources in `prole-db` namespace: namespace, services, ConfigMaps, StatefulSets, Ingress rules, and PersistentVolumes. - Integrate deployment and configuration enhancements for `garage`, `prole`, and related components.
304 lines
12 KiB
Python
304 lines
12 KiB
Python
"""Tests for installer/core/actions.py – ProleInstallerBase helper methods."""
|
||
from __future__ import annotations
|
||
import os
|
||
import sys
|
||
from pathlib import Path
|
||
from unittest import mock
|
||
from unittest.mock import MagicMock, patch
|
||
|
||
import pytest
|
||
|
||
from installer.core.actions import ProleInstallerBase, ProleSilentInstaller, _configure_unbuffered_io
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Concrete subclass for testing the abstract base
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class _TestableInstaller(ProleInstallerBase):
|
||
"""Minimal concrete subclass for testing base-class helpers."""
|
||
|
||
def __init__(self, inputs=None, project_root=None):
|
||
self._inputs = inputs or {}
|
||
self.project_root = project_root or Path('/tmp/fake-prole')
|
||
self.controller = MagicMock()
|
||
self._init_shared_state()
|
||
|
||
def _get_input(self, key: str, default: str | None = None) -> str:
|
||
return self._inputs.get(key, default or '')
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _configure_unbuffered_io
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestConfigureUnbufferedIO:
|
||
def test_sets_env_var(self):
|
||
with mock.patch.dict(os.environ, {}, clear=False):
|
||
os.environ.pop('PYTHONUNBUFFERED', None)
|
||
_configure_unbuffered_io()
|
||
assert os.environ.get('PYTHONUNBUFFERED') == '1'
|
||
|
||
def test_does_not_overwrite_existing(self):
|
||
with mock.patch.dict(os.environ, {'PYTHONUNBUFFERED': 'already'}, clear=False):
|
||
_configure_unbuffered_io()
|
||
assert os.environ['PYTHONUNBUFFERED'] == 'already'
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _init_shared_state
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestInitSharedState:
|
||
def test_initial_sections(self):
|
||
inst = _TestableInstaller()
|
||
expected_sections = [
|
||
'Global', 'Welcome', 'Dependencies', 'Network',
|
||
'Port Forwards', 'System Environment', 'Kerberos Authentication',
|
||
'Ollama', 'Optional Features', 'Database Creation',
|
||
'Docker Build', 'Initialize Cluster', 'Initialization Scripts',
|
||
'Deployment', 'Dev Cluster (k3d)', 'Service Cluster (k3s)',
|
||
'Prod Cluster (k8s)', 'Install',
|
||
]
|
||
for section in expected_sections:
|
||
assert section in inst.prole_cfg_data
|
||
|
||
def test_initial_flags(self):
|
||
inst = _TestableInstaller()
|
||
assert inst._cfg_secret_cache == {}
|
||
assert inst._secrets_finalized is False
|
||
assert inst._managed_kubeconfig is None
|
||
assert inst._repair_ran is False
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _get_input_bool
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestGetInputBool:
|
||
def test_true_values(self):
|
||
for val in ('true', '1', 'yes', 'on', 'True', 'YES'):
|
||
inst = _TestableInstaller(inputs={'key': val})
|
||
assert inst._get_input_bool('key') is True
|
||
|
||
def test_false_values(self):
|
||
for val in ('false', '0', 'no', 'off', 'False', 'NO'):
|
||
inst = _TestableInstaller(inputs={'key': val})
|
||
assert inst._get_input_bool('key') is False
|
||
|
||
def test_default(self):
|
||
inst = _TestableInstaller()
|
||
assert inst._get_input_bool('missing_key', default=True) is True
|
||
assert inst._get_input_bool('missing_key', default=False) is False
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Logging
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestLogging:
|
||
def test_log_prints(self, capsys):
|
||
inst = _TestableInstaller()
|
||
inst.log("hello world")
|
||
captured = capsys.readouterr()
|
||
assert "hello world" in captured.out
|
||
|
||
def test_err_prints_to_stderr(self, capsys):
|
||
inst = _TestableInstaller()
|
||
inst.err("error msg")
|
||
captured = capsys.readouterr()
|
||
assert "error msg" in captured.err
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _deployment_mode
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestDeploymentMode:
|
||
def test_dev_mode(self):
|
||
inst = _TestableInstaller(inputs={'init_cluster.cluster_env': 'dev'})
|
||
assert inst._deployment_mode() == 'k3d'
|
||
|
||
def test_service_mode(self):
|
||
inst = _TestableInstaller(inputs={'init_cluster.cluster_env': 'service'})
|
||
assert inst._deployment_mode() == 'k3s'
|
||
|
||
def test_prod_mode(self):
|
||
inst = _TestableInstaller(inputs={'init_cluster.cluster_env': 'prod'})
|
||
assert inst._deployment_mode() == 'k8s'
|
||
|
||
def test_empty_mode(self):
|
||
inst = _TestableInstaller(inputs={'init_cluster.cluster_env': ''})
|
||
result = inst._deployment_mode()
|
||
assert result == '' or result is not None
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _secret_namespace
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestSecretNamespace:
|
||
def test_from_db_namespace(self):
|
||
inst = _TestableInstaller(inputs={'init_password.db_namespace': 'my-ns'})
|
||
assert inst._secret_namespace() == 'my-ns'
|
||
|
||
def test_from_env_namespace(self):
|
||
inst = _TestableInstaller(inputs={'env_setup.NAMESPACE': 'env-ns'})
|
||
assert inst._secret_namespace() == 'env-ns'
|
||
|
||
def test_from_cfg_data(self):
|
||
inst = _TestableInstaller()
|
||
inst.prole_cfg_data['Global']['NAMESPACE'] = 'cfg-ns'
|
||
assert inst._secret_namespace() == 'cfg-ns'
|
||
|
||
def test_default(self):
|
||
inst = _TestableInstaller()
|
||
assert inst._secret_namespace() == 'default'
|
||
|
||
def test_priority_db_over_env(self):
|
||
inst = _TestableInstaller(inputs={
|
||
'init_password.db_namespace': 'db-ns',
|
||
'env_setup.NAMESPACE': 'env-ns',
|
||
})
|
||
assert inst._secret_namespace() == 'db-ns'
|
||
|
||
def test_strips_whitespace(self):
|
||
inst = _TestableInstaller(inputs={'init_password.db_namespace': ' my-ns '})
|
||
assert inst._secret_namespace() == 'my-ns'
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _service_namespace
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestServiceNamespace:
|
||
def test_from_cfg_data(self):
|
||
inst = _TestableInstaller()
|
||
inst.prole_cfg_data['Global']['SERVICE_NAMESPACE'] = 'svc-ns'
|
||
assert inst._service_namespace() == 'svc-ns'
|
||
|
||
def test_from_env_var(self):
|
||
inst = _TestableInstaller()
|
||
with mock.patch.dict(os.environ, {'SERVICE_NAMESPACE': 'env-svc-ns'}):
|
||
assert inst._service_namespace() == 'env-svc-ns'
|
||
|
||
def test_default(self):
|
||
inst = _TestableInstaller()
|
||
with mock.patch.dict(os.environ, {}, clear=False):
|
||
os.environ.pop('SERVICE_NAMESPACE', None)
|
||
assert inst._service_namespace() == 'default'
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _argocd_namespace / _registry_namespace
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class _TestableSilentInstaller(ProleSilentInstaller):
|
||
"""Testable subclass of ProleSilentInstaller."""
|
||
|
||
def __init__(self, inputs=None, project_root=None):
|
||
self._inputs_dict = inputs or {}
|
||
self.project_root = project_root or Path('/tmp/fake-prole')
|
||
self.controller = MagicMock()
|
||
self.cfg_path = None
|
||
self.silent = True
|
||
self._init_shared_state()
|
||
|
||
def _get_input(self, key: str, default: str | None = None) -> str:
|
||
return self._inputs_dict.get(key, default or '')
|
||
|
||
|
||
class TestNamespaceHelpers:
|
||
def test_argocd_from_cfg(self):
|
||
inst = _TestableSilentInstaller()
|
||
inst.prole_cfg_data['Global']['ARGOCD_NAMESPACE'] = 'argo-ns'
|
||
assert inst._argocd_namespace() == 'argo-ns'
|
||
|
||
def test_argocd_from_env(self):
|
||
inst = _TestableSilentInstaller()
|
||
with mock.patch.dict(os.environ, {'ARGOCD_NAMESPACE': 'env-argo'}):
|
||
assert inst._argocd_namespace() == 'env-argo'
|
||
|
||
def test_argocd_default(self):
|
||
inst = _TestableSilentInstaller()
|
||
with mock.patch.dict(os.environ, {}, clear=False):
|
||
os.environ.pop('ARGOCD_NAMESPACE', None)
|
||
assert inst._argocd_namespace() == 'argocd'
|
||
|
||
def test_registry_from_cfg(self):
|
||
inst = _TestableSilentInstaller()
|
||
inst.prole_cfg_data['Global']['REGISTRY_NAMESPACE'] = 'reg-ns'
|
||
assert inst._registry_namespace() == 'reg-ns'
|
||
|
||
def test_registry_from_env(self):
|
||
inst = _TestableSilentInstaller()
|
||
with mock.patch.dict(os.environ, {'REGISTRY_NAMESPACE': 'env-reg'}):
|
||
assert inst._registry_namespace() == 'env-reg'
|
||
|
||
def test_registry_default(self):
|
||
inst = _TestableSilentInstaller()
|
||
with mock.patch.dict(os.environ, {}, clear=False):
|
||
os.environ.pop('REGISTRY_NAMESPACE', None)
|
||
assert inst._registry_namespace() == 'default'
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _read_k3s_cfg_values
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestReadK3sCfgValues:
|
||
def test_no_cfg_path(self):
|
||
inst = _TestableInstaller()
|
||
env, url, token = inst._read_k3s_cfg_values(None)
|
||
# With no cfg, should return empty/defaults
|
||
assert isinstance(env, str)
|
||
assert isinstance(url, str)
|
||
assert isinstance(token, str)
|
||
|
||
def test_with_cfg_path(self, tmp_path):
|
||
import configparser
|
||
cfg = configparser.ConfigParser(interpolation=None)
|
||
cfg.optionxform = str
|
||
cfg.add_section('Initialize Cluster')
|
||
cfg.set('Initialize Cluster', 'ENVIRONMENT', 'service')
|
||
cfg.set('Initialize Cluster', 'K3S_SERVER_URL', 'https://10.0.0.1:6443')
|
||
cfg.set('Initialize Cluster', 'K3S_TOKEN', 'mytoken')
|
||
cfg_path = tmp_path / 'prole.cfg'
|
||
with open(cfg_path, 'w') as f:
|
||
cfg.write(f)
|
||
inst = _TestableInstaller()
|
||
env, url, token = inst._read_k3s_cfg_values(cfg_path)
|
||
assert env == 'service'
|
||
assert url == 'https://10.0.0.1:6443'
|
||
assert token == 'mytoken'
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _run_script
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestRunScript:
|
||
def test_run_script_delegates(self):
|
||
inst = _TestableInstaller(inputs={'init_cluster.cluster_env': 'dev'})
|
||
inst.controller.run_script.return_value = 0
|
||
rc = inst._run_script("test_script.sh", args=["arg1"])
|
||
assert rc == 0
|
||
inst.controller.run_script.assert_called_once()
|
||
|
||
def test_run_script_adds_mode(self):
|
||
inst = _TestableInstaller(inputs={'init_cluster.cluster_env': 'dev'})
|
||
inst.controller.run_script.return_value = 0
|
||
inst._run_script("init_common_services.sh", args=["update"])
|
||
call_kwargs = inst.controller.run_script.call_args
|
||
args_passed = call_kwargs[1].get('args', call_kwargs[0][1] if len(call_kwargs[0]) > 1 else [])
|
||
assert '--mode' in args_passed
|
||
assert 'k3d' in args_passed
|
||
|
||
def test_run_script_no_mode_for_non_init(self):
|
||
inst = _TestableInstaller(inputs={'init_cluster.cluster_env': 'dev'})
|
||
inst.controller.run_script.return_value = 0
|
||
inst._run_script("build-a-bao.sh", args=[])
|
||
call_kwargs = inst.controller.run_script.call_args
|
||
args_passed = call_kwargs[1].get('args', call_kwargs[0][1] if len(call_kwargs[0]) > 1 else [])
|
||
assert '--mode' not in args_passed
|