import unittest from unittest.mock import MagicMock, patch, ANY from pathlib import Path from installer.core.milestones import ( DependenciesMilestone, NetworkScanMilestone, EnvSetupMilestone, SecretManagementMilestone, ClusterLifecycleMilestone, InitializationScriptsMilestone, DeploymentMilestone ) from installer.state import InstallerState from installer.core.controller import ProleController class TestMilestones(unittest.TestCase): def setUp(self): self.state = InstallerState() self.state.inputs['dependencies.auto_install_missing'] = "True" self.state.inputs['network_scan.run'] = "True" self.state.controller = MagicMock(spec=ProleController) @patch('installer.config.get_dep_info') @patch('subprocess.run') def test_dependencies_milestone_all_installed(self, mock_run, mock_get_dep_info): # Mock all dependencies as installed mock_get_dep_info.return_value = (True, "/usr/bin/foo", "1.0.0") milestone = DependenciesMilestone() progress_mock = MagicMock() milestone.execute(self.state, progress=progress_mock) self.assertEqual(self.state.config_data.get('Dependencies', {}).get('STATUS'), 'All installed') progress_mock.assert_called() @patch('installer.config.get_dep_info') @patch('subprocess.Popen') def test_dependencies_milestone_install_missing(self, mock_popen, mock_get_dep_info): # Mock one missing dependency def side_effect(dep): if dep['id'] == 'brew': if mock_popen.called: return (True, "/usr/local/bin/brew", "4.0.0") return (False, None, None) return (True, "/usr/bin/python", "3.10.0") mock_get_dep_info.side_effect = side_effect # Mock subprocess successful run mock_process = MagicMock() mock_process.returncode = 0 mock_process.stdout = MagicMock() mock_process.stdout.readline.side_effect = ["Installing...", ""] mock_process.poll.side_effect = [None, 0] mock_process.wait.return_value = 0 mock_popen.return_value = mock_process # Ensure brew install is enabled in state self.state.inputs['dependencies.brew.install'] = "True" milestone = DependenciesMilestone() milestone.execute(self.state) self.assertEqual(self.state.config_data.get('Dependencies', {}).get('STATUS'), 'All installed') # Check that it tried to install brew mock_popen.assert_called() @patch('installer.config.get_resource_path') @patch('subprocess.Popen') def test_network_scan_milestone(self, mock_popen, mock_get_resource): mock_binary = MagicMock() mock_binary.exists.return_value = True mock_binary.__str__.return_value = "/tmp/prole-agent" mock_get_resource.return_value = mock_binary mock_process = MagicMock() mock_process.returncode = 0 mock_process.stdout = MagicMock() mock_process.stdout.readline.side_effect = ["Scanning...", "KDC is: 10.0.0.1", ""] mock_process.poll.side_effect = [None, None, 0] mock_process.wait.return_value = 0 mock_popen.return_value = mock_process milestone = NetworkScanMilestone() milestone.execute(self.state) self.assertEqual(self.state.inputs.get('kerberos_config.kdc'), '10.0.0.1') self.assertEqual(self.state.config_data.get('Network', {}).get('KDC_AUTO_DETECTED'), '10.0.0.1') def test_env_setup_milestone(self): self.state.inputs['env_setup.PROLE_HOME'] = "/custom/home" self.state.inputs['env_setup.NAMESPACE'] = "test-ns" milestone = EnvSetupMilestone() milestone.execute(self.state) self.assertEqual(self.state.config_data.get('System Environment', {}).get('PROLE_HOME'), '/custom/home') self.assertEqual(self.state.config_data.get('System Environment', {}).get('NAMESPACE'), 'test-ns') @patch('installer.core.milestones.inst_config._resolve_secret_value') def test_secret_management_milestone(self, mock_resolve): mock_resolve.side_effect = lambda x: f"resolved-{x}" if "secret" in x else x self.state.inputs['init_password.db_password'] = "secret-pw" milestone = SecretManagementMilestone() milestone.execute(self.state) self.assertEqual(self.state.inputs.get('init_password.db_password'), 'resolved-secret-pw') self.assertEqual(self.state.inputs.get('init_password.db_password_confirm'), 'resolved-secret-pw') @patch('subprocess.run') def test_cluster_lifecycle_milestone_dev(self, mock_subprocess_run): self.state.inputs['init_cluster.cluster_env'] = 'dev' self.state.inputs['init_cluster.start_cluster'] = 'True' self.state.controller.run_script.return_value = 0 self.state.controller.check_docker_running.return_value = True self.state.controller.get_prole_db_version.return_value = "1.0.0" # Generic success result for k3d kubeconfig merge / misc calls mock_ok = MagicMock() mock_ok.returncode = 0 mock_ok.stdout = "" # Mock k3d cluster list showing cluster exists and running mock_list_result = MagicMock() mock_list_result.stdout = "prole-dev-cluster 1/1 0/2 running" mock_list_result.returncode = 0 # Mock docker image inspect (image not found locally) mock_inspect_result = MagicMock() mock_inspect_result.returncode = 1 # Mock kubectl config use-context mock_ctx_result = MagicMock() mock_ctx_result.returncode = 0 # Use a default return so extra calls (k3d kubeconfig merge, etc.) don't exhaust side_effect mock_subprocess_run.return_value = mock_ok mock_subprocess_run.side_effect = None def _side_effect(*args, **kwargs): cmd = args[0] if args else kwargs.get('args', []) if cmd and cmd[0] == 'k3d': if 'cluster' in cmd and 'list' in cmd: return mock_list_result # k3d kubeconfig merge, k3d cluster start, k3d image import, etc. return mock_ok if cmd and cmd[0] == 'kubectl': return mock_ctx_result if cmd and cmd[0] == 'docker': return mock_inspect_result return mock_ok mock_subprocess_run.side_effect = _side_effect milestone = ClusterLifecycleMilestone() milestone.execute(self.state) # Verify k3d cluster list was called mock_subprocess_run.assert_any_call( ['k3d', 'cluster', 'list', '--no-headers'], capture_output=True, text=True ) @patch('installer.core.milestones.InitializationScriptsMilestone._run_cmd') @patch('subprocess.run') def test_init_scripts_milestone(self, mock_subprocess_run, mock_run_cmd): self.state.controller.run_script.return_value = 0 # Mock _run_cmd to return 0 (secrets exist) mock_run_cmd.return_value = 0 milestone = InitializationScriptsMilestone() milestone.execute(self.state) # Verify current script list is called self.state.controller.run_script.assert_any_call( "init_common_services.sh", args=ANY, env=ANY, on_line=ANY ) self.state.controller.run_script.assert_any_call( "init_cloudnative_pg.sh", args=ANY, env=ANY, on_line=ANY ) self.state.controller.run_script.assert_any_call( "init_monitoring.sh", args=ANY, env=ANY, on_line=ANY ) def test_deployment_milestone(self): self.state.controller.run_script.return_value = 0 milestone = DeploymentMilestone() milestone.execute(self.state) self.state.controller.run_script.assert_any_call( "init_cloudnative_pg.sh", args=ANY, env=ANY, on_line=ANY ) if __name__ == '__main__': unittest.main()