diff --git a/tests/hostcfgd/hostcfgd_test.py b/tests/hostcfgd/hostcfgd_test.py index 21f80bb7..3858a226 100644 --- a/tests/hostcfgd/hostcfgd_test.py +++ b/tests/hostcfgd/hostcfgd_test.py @@ -373,6 +373,367 @@ def test_banner_message(self, mock_run_cmd): mock_run_cmd.assert_has_calls([call(['systemctl', 'restart', 'banner-config'], True, True)]) +# class TestMemoryStatisticsCfgd(TestCase): +# """Test MemoryStatisticsCfg functionalities.""" + +# def setUp(self): +# # Initial configuration for Memory Statistics +# MockConfigDb.CONFIG_DB['MEMORY_STATISTICS'] = { +# 'enabled': 'false', +# 'sampling_interval': '5', +# 'retention_period': '15' +# } +# self.mem_stat_cfg = hostcfgd.MemoryStatisticsCfg(MockConfigDb.CONFIG_DB) + +# def tearDown(self): +# MockConfigDb.CONFIG_DB = {} + +# # Test initial loading +# def test_load_with_invalid_key(self): +# """Test loading configuration with invalid key""" +# config = {'invalid_key': 'value', 'enabled': 'true'} +# with mock.patch('hostcfgd.syslog.syslog') as mock_syslog: +# self.mem_stat_cfg.load(config) +# mock_syslog.assert_any_call(mock.ANY, "MemoryStatisticsCfg: Invalid key 'invalid_key' in initial configuration.") + +# def test_load_with_empty_config(self): +# """Test loading empty configuration""" +# with mock.patch('hostcfgd.syslog.syslog') as mock_syslog: +# self.mem_stat_cfg.load(None) +# mock_syslog.assert_any_call(mock.ANY, "MemoryStatisticsCfg: Loading initial configuration") + +# # Test validation +# def test_memory_statistics_update_invalid_key(self): +# """Test update with invalid key""" +# with mock.patch('hostcfgd.syslog.syslog') as mock_syslog: +# self.mem_stat_cfg.memory_statistics_update('invalid_key', 'value') +# mock_syslog.assert_any_call(mock.ANY, "MemoryStatisticsCfg: Invalid key 'invalid_key' received.") + +# def test_memory_statistics_update_invalid_numeric_value(self): +# """Test update with invalid numeric value""" +# with mock.patch('hostcfgd.syslog.syslog') as mock_syslog: +# self.mem_stat_cfg.memory_statistics_update('sampling_interval', '-1') +# mock_syslog.assert_any_call(mock.ANY, "MemoryStatisticsCfg: Invalid value '-1' for key 'sampling_interval'. Must be a positive integer.") + +# # Test daemon management +# @mock.patch('hostcfgd.subprocess.Popen') +# @mock.patch('hostcfgd.os.kill') +# def test_restart_memory_statistics_success(self, mock_kill, mock_popen): +# """Test successful daemon restart""" +# with mock.patch('hostcfgd.syslog.syslog') as mock_syslog: +# with mock.patch.object(self.mem_stat_cfg, 'get_memory_statistics_pid', return_value=123): +# self.mem_stat_cfg.restart_memory_statistics() +# mock_kill.assert_called_with(123, signal.SIGTERM) +# mock_popen.assert_called_once() + +# @mock.patch('hostcfgd.subprocess.Popen') +# def test_restart_memory_statistics_failure(self, mock_popen): +# """Test failed daemon restart""" +# mock_popen.side_effect = Exception("Failed to start") +# with mock.patch('hostcfgd.syslog.syslog') as mock_syslog: +# self.mem_stat_cfg.restart_memory_statistics() +# mock_syslog.assert_any_call(mock.ANY, "MemoryStatisticsCfg: Failed to start MemoryStatisticsDaemon: Failed to start") + +# # Test PID management +# def test_get_memory_statistics_pid_file_not_found(self): +# """Test PID retrieval when file doesn't exist""" +# with mock.patch('builtins.open', side_effect=FileNotFoundError): +# with mock.patch('hostcfgd.syslog.syslog') as mock_syslog: +# pid = self.mem_stat_cfg.get_memory_statistics_pid() +# self.assertIsNone(pid) +# mock_syslog.assert_any_call(mock.ANY, "MemoryStatisticsCfg: PID file not found. Daemon might not be running.") + +# def test_get_memory_statistics_pid_invalid_content(self): +# """Test PID retrieval with invalid file content""" +# mock_open = mock.mock_open(read_data="invalid") +# with mock.patch('builtins.open', mock_open): +# with mock.patch('hostcfgd.syslog.syslog') as mock_syslog: +# pid = self.mem_stat_cfg.get_memory_statistics_pid() +# self.assertIsNone(pid) +# mock_syslog.assert_any_call(mock.ANY, "MemoryStatisticsCfg: PID file contents invalid.") + +# @mock.patch('hostcfgd.psutil.pid_exists', return_value=True) +# @mock.patch('hostcfgd.psutil.Process') +# def test_get_memory_statistics_pid_wrong_process(self, mock_process, mock_pid_exists): +# """Test PID retrieval when process name doesn't match""" +# mock_process_instance = mock.Mock() +# mock_process_instance.name.return_value = "wrong_process" +# mock_process.return_value = mock_process_instance + +# mock_open = mock.mock_open(read_data="123") +# with mock.patch('builtins.open', mock_open): +# with mock.patch('hostcfgd.syslog.syslog') as mock_syslog: +# pid = self.mem_stat_cfg.get_memory_statistics_pid() +# self.assertIsNone(pid) +# mock_syslog.assert_any_call(mock.ANY, "MemoryStatisticsCfg: PID 123 does not correspond to memory_statistics_service.py.") + +# # Test daemon shutdown +# @mock.patch('hostcfgd.psutil.Process') +# def test_wait_for_shutdown_timeout(self, mock_process): +# """Test shutdown waiting timeout""" +# mock_process_instance = mock.Mock() +# mock_process_instance.wait.side_effect = psutil.TimeoutExpired(123, 10) +# mock_process.return_value = mock_process_instance + +# with mock.patch('hostcfgd.syslog.syslog') as mock_syslog: +# self.mem_stat_cfg.wait_for_shutdown(123) +# mock_syslog.assert_any_call(mock.ANY, "MemoryStatisticsCfg: Timed out while waiting for daemon (PID 123) to shut down.") + +# @mock.patch('hostcfgd.psutil.Process') +# def test_wait_for_shutdown_no_process(self, mock_process): +# """Test shutdown waiting when process doesn't exist""" +# mock_process.side_effect = psutil.NoSuchProcess(123) + +# with mock.patch('hostcfgd.syslog.syslog') as mock_syslog: +# self.mem_stat_cfg.wait_for_shutdown(123) +# mock_syslog.assert_any_call(mock.ANY, "MemoryStatisticsCfg: MemoryStatisticsDaemon process not found.") + +# # Test enable/disable functionality +# def test_memory_statistics_enable(self): +# """Test enabling memory statistics""" +# with mock.patch.object(self.mem_stat_cfg, 'restart_memory_statistics') as mock_restart: +# self.mem_stat_cfg.memory_statistics_update('enabled', 'true') +# mock_restart.assert_called_once() +# self.assertEqual(self.mem_stat_cfg.cache['enabled'], 'true') + +# def test_memory_statistics_update_same_value(self): +# """Test update with same value (should not trigger apply_setting)""" +# with mock.patch.object(self.mem_stat_cfg, 'apply_setting') as mock_apply: +# self.mem_stat_cfg.memory_statistics_update('sampling_interval', '5') +# mock_apply.assert_not_called() + +# def test_memory_statistics_update_exception_handling(self): +# """Test exception handling in memory_statistics_update""" +# with mock.patch.object(self.mem_stat_cfg, 'apply_setting', side_effect=Exception("Test error")): +# with mock.patch('hostcfgd.syslog.syslog') as mock_syslog: +# self.mem_stat_cfg.memory_statistics_update('enabled', 'true') +# mock_syslog.assert_any_call(mock.ANY, "MemoryStatisticsCfg: Failed to manage MemoryStatisticsDaemon: Test error") + +# def test_apply_setting_with_non_enabled_key(self): +# """Test apply_setting with sampling_interval or retention_period""" +# with mock.patch.object(self.mem_stat_cfg, 'reload_memory_statistics') as mock_reload: +# self.mem_stat_cfg.apply_setting('sampling_interval', '10') +# mock_reload.assert_called_once() + +# def test_apply_setting_with_enabled_false(self): +# """Test apply_setting with enabled=false""" +# with mock.patch.object(self.mem_stat_cfg, 'shutdown_memory_statistics') as mock_shutdown: +# self.mem_stat_cfg.apply_setting('enabled', 'false') +# mock_shutdown.assert_called_once() + +# def test_apply_setting_exception(self): +# """Test exception handling in apply_setting""" +# with mock.patch.object(self.mem_stat_cfg, 'restart_memory_statistics', +# side_effect=Exception("Test error")): +# with mock.patch('hostcfgd.syslog.syslog') as mock_syslog: +# self.mem_stat_cfg.apply_setting('enabled', 'true') +# mock_syslog.assert_any_call(mock.ANY, +# "MemoryStatisticsCfg: Exception in apply_setting() for key 'enabled': Test error") + +# @mock.patch('hostcfgd.psutil.pid_exists', return_value=False) +# def test_get_memory_statistics_pid_nonexistent(self, mock_pid_exists): +# """Test get_memory_statistics_pid when PID doesn't exist""" +# mock_open = mock.mock_open(read_data="123") +# with mock.patch('builtins.open', mock_open): +# with mock.patch('hostcfgd.syslog.syslog') as mock_syslog: +# pid = self.mem_stat_cfg.get_memory_statistics_pid() +# self.assertIsNone(pid) +# mock_syslog.assert_any_call(mock.ANY, "MemoryStatisticsCfg: PID does not exist.") + +# @mock.patch('hostcfgd.psutil.Process') +# def test_get_memory_statistics_pid_exception(self, mock_process): +# """Test general exception handling in get_memory_statistics_pid""" +# mock_process.side_effect = Exception("Unexpected error") +# mock_open = mock.mock_open(read_data="123") + +# with mock.patch('hostcfgd.psutil.pid_exists', return_value=True): +# with mock.patch('builtins.open', mock_open): +# with mock.patch('hostcfgd.syslog.syslog') as mock_syslog: +# pid = self.mem_stat_cfg.get_memory_statistics_pid() +# self.assertIsNone(pid) +# mock_syslog.assert_any_call(mock.ANY, +# "MemoryStatisticsCfg: Exception failed to retrieve MemoryStatisticsDaemon PID: Unexpected error") + +# def test_memory_statistics_handler(self): +# """Test memory_statistics_handler in HostConfigDaemon""" +# daemon = hostcfgd.HostConfigDaemon() +# with mock.patch.object(daemon.memorystatisticscfg, 'memory_statistics_update') as mock_update: +# daemon.memory_statistics_handler('enabled', None, 'true') +# mock_update.assert_called_once_with('enabled', 'true') + +# def test_memory_statistics_handler_exception(self): +# """Test exception handling in memory_statistics_handler""" +# daemon = hostcfgd.HostConfigDaemon() +# with mock.patch.object(daemon.memorystatisticscfg, 'memory_statistics_update', +# side_effect=Exception("Handler error")): +# with mock.patch('hostcfgd.syslog.syslog') as mock_syslog: +# daemon.memory_statistics_handler('enabled', None, 'true') +# mock_syslog.assert_any_call(mock.ANY, +# "MemoryStatisticsCfg: Error while handling memory statistics update: Handler error") + +# @mock.patch('hostcfgd.psutil.Process') +# def test_wait_for_shutdown_general_exception(self, mock_process): +# """Test general exception handling in wait_for_shutdown""" +# mock_process.side_effect = Exception("Unexpected shutdown error") +# with mock.patch('hostcfgd.syslog.syslog') as mock_syslog: +# self.mem_stat_cfg.wait_for_shutdown(123) +# mock_syslog.assert_any_call(mock.ANY, +# "MemoryStatisticsCfg: Exception in wait_for_shutdown(): Unexpected shutdown error") + +# def test_memory_statistics_disable(self): +# """Test disabling memory statistics""" +# # First set the initial state to enabled +# self.mem_stat_cfg.cache['enabled'] = 'true' + +# # Mock the apply_setting method to avoid actual system calls +# with mock.patch.object(self.mem_stat_cfg, 'apply_setting') as mock_apply: +# self.mem_stat_cfg.memory_statistics_update('enabled', 'false') + +# # Verify apply_setting was called with correct parameters +# mock_apply.assert_called_once_with('enabled', 'false') + +# # Verify the cache was updated +# self.assertEqual(self.mem_stat_cfg.cache['enabled'], 'false') + +# def test_memory_statistics_disable_with_shutdown(self): +# """Test disabling memory statistics with full shutdown chain""" +# # First set the initial state to enabled +# self.mem_stat_cfg.cache['enabled'] = 'true' + +# # Mock both get_memory_statistics_pid and os.kill to simulate full shutdown +# with mock.patch.object(self.mem_stat_cfg, 'get_memory_statistics_pid', return_value=123) as mock_get_pid, \ +# mock.patch('hostcfgd.os.kill') as mock_kill, \ +# mock.patch.object(self.mem_stat_cfg, 'wait_for_shutdown') as mock_wait: + +# self.mem_stat_cfg.memory_statistics_update('enabled', 'false') + +# # Verify the shutdown sequence +# mock_get_pid.assert_called_once() +# mock_kill.assert_called_once_with(123, signal.SIGTERM) +# mock_wait.assert_called_once_with(123) + +# # Verify the cache was updated +# self.assertEqual(self.mem_stat_cfg.cache['enabled'], 'false') + +# def test_memory_statistics_disable_no_running_daemon(self): +# """Test disabling memory statistics when daemon is not running""" +# # First set the initial state to enabled +# self.mem_stat_cfg.cache['enabled'] = 'true' + +# # Mock get_memory_statistics_pid to return None (no running daemon) +# with mock.patch.object(self.mem_stat_cfg, 'get_memory_statistics_pid', return_value=None) as mock_get_pid: +# self.mem_stat_cfg.memory_statistics_update('enabled', 'false') + +# # Verify get_pid was called +# mock_get_pid.assert_called_once() + +# # Verify the cache was updated despite no running daemon +# self.assertEqual(self.mem_stat_cfg.cache['enabled'], 'false') + + + + + +# def test_reload_memory_statistics_success(self): +# """Test successful reload of memory statistics daemon""" +# with mock.patch.object(self.mem_stat_cfg, 'get_memory_statistics_pid', return_value=123) as mock_get_pid, \ +# mock.patch('hostcfgd.os.kill') as mock_kill, \ +# mock.patch('hostcfgd.syslog.syslog') as mock_syslog: + +# self.mem_stat_cfg.reload_memory_statistics() + +# # Verify the method calls +# mock_get_pid.assert_called_once() +# mock_kill.assert_called_once_with(123, signal.SIGHUP) +# mock_syslog.assert_any_call(mock.ANY, "MemoryStatisticsCfg: Sent SIGHUP to reload daemon configuration") + +# def test_reload_memory_statistics_no_pid(self): +# """Test reload when no daemon is running""" +# with mock.patch.object(self.mem_stat_cfg, 'get_memory_statistics_pid', return_value=None) as mock_get_pid, \ +# mock.patch('hostcfgd.os.kill') as mock_kill: + +# self.mem_stat_cfg.reload_memory_statistics() + +# # Verify the method calls +# mock_get_pid.assert_called_once() +# mock_kill.assert_not_called() + +# def test_reload_memory_statistics_failure(self): +# """Test reload failure with exception""" +# with mock.patch.object(self.mem_stat_cfg, 'get_memory_statistics_pid', return_value=123) as mock_get_pid, \ +# mock.patch('hostcfgd.os.kill', side_effect=Exception("Test error")), \ +# mock.patch('hostcfgd.syslog.syslog') as mock_syslog: + +# self.mem_stat_cfg.reload_memory_statistics() + +# mock_get_pid.assert_called_once() +# mock_syslog.assert_any_call(mock.ANY, "MemoryStatisticsCfg: Failed to reload MemoryStatisticsDaemon: Test error") + +# def test_shutdown_memory_statistics_success(self): +# """Test successful shutdown of memory statistics daemon""" +# with mock.patch.object(self.mem_stat_cfg, 'get_memory_statistics_pid', return_value=123) as mock_get_pid, \ +# mock.patch('hostcfgd.os.kill') as mock_kill, \ +# mock.patch.object(self.mem_stat_cfg, 'wait_for_shutdown') as mock_wait, \ +# mock.patch('hostcfgd.syslog.syslog') as mock_syslog: + +# self.mem_stat_cfg.shutdown_memory_statistics() + +# mock_get_pid.assert_called_once() +# mock_kill.assert_called_once_with(123, signal.SIGTERM) +# mock_wait.assert_called_once_with(123) +# mock_syslog.assert_any_call(mock.ANY, "MemoryStatisticsCfg: Sent SIGTERM to stop MemoryStatisticsDaemon") + +# def test_shutdown_memory_statistics_failure(self): +# """Test shutdown failure with exception""" +# with mock.patch.object(self.mem_stat_cfg, 'get_memory_statistics_pid', return_value=123) as mock_get_pid, \ +# mock.patch('hostcfgd.os.kill', side_effect=Exception("Test error")), \ +# mock.patch('hostcfgd.syslog.syslog') as mock_syslog: + +# self.mem_stat_cfg.shutdown_memory_statistics() + +# mock_get_pid.assert_called_once() +# mock_syslog.assert_any_call(mock.ANY, "MemoryStatisticsCfg: Failed to shutdown MemoryStatisticsDaemon: Test error") + +# def test_wait_for_shutdown_success(self): +# """Test successful wait for shutdown""" +# mock_process = mock.Mock() +# with mock.patch('hostcfgd.psutil.Process', return_value=mock_process) as mock_process_class, \ +# mock.patch('hostcfgd.syslog.syslog') as mock_syslog: + +# self.mem_stat_cfg.wait_for_shutdown(123) + +# mock_process_class.assert_called_once_with(123) +# mock_process.wait.assert_called_once_with(timeout=10) +# mock_syslog.assert_any_call(mock.ANY, "MemoryStatisticsCfg: MemoryStatisticsDaemon stopped gracefully") + +# def test_main_function(self): +# """Test the main function""" +# with mock.patch('hostcfgd.HostConfigDaemon') as mock_daemon: +# hostcfgd.main() + +# # Verify daemon was instantiated and started +# mock_daemon.assert_called_once() +# mock_instance = mock_daemon.return_value +# mock_instance.register_callbacks.assert_called_once() +# mock_instance.start.assert_called_once() + +# def test_process_name_mismatch(self): +# """Test when process name doesn't match expected daemon name""" +# mock_process = mock.Mock() +# mock_process.name.return_value = "wrong_process_name" + +# with mock.patch('builtins.open', mock.mock_open(read_data="123")), \ +# mock.patch('hostcfgd.psutil.pid_exists', return_value=True), \ +# mock.patch('hostcfgd.psutil.Process', return_value=mock_process), \ +# mock.patch('hostcfgd.syslog.syslog') as mock_syslog: + +# pid = self.mem_stat_cfg.get_memory_statistics_pid() + +# self.assertIsNone(pid) +# mock_syslog.assert_any_call(mock.ANY, "MemoryStatisticsCfg: PID 123 does not correspond to memory_statistics_service.py.") + + + class TestMemoryStatisticsCfgd(TestCase): """Test MemoryStatisticsCfg functionalities.""" @@ -388,346 +749,151 @@ def setUp(self): def tearDown(self): MockConfigDb.CONFIG_DB = {} - # Test initial loading - def test_load_with_invalid_key(self): - """Test loading configuration with invalid key""" + # Test configuration loading and validation + def test_load_configurations(self): + """Test various configuration loading scenarios""" + # Test loading with invalid key config = {'invalid_key': 'value', 'enabled': 'true'} with mock.patch('hostcfgd.syslog.syslog') as mock_syslog: self.mem_stat_cfg.load(config) mock_syslog.assert_any_call(mock.ANY, "MemoryStatisticsCfg: Invalid key 'invalid_key' in initial configuration.") - def test_load_with_empty_config(self): - """Test loading empty configuration""" + # Test loading empty configuration with mock.patch('hostcfgd.syslog.syslog') as mock_syslog: self.mem_stat_cfg.load(None) mock_syslog.assert_any_call(mock.ANY, "MemoryStatisticsCfg: Loading initial configuration") - # Test validation - def test_memory_statistics_update_invalid_key(self): - """Test update with invalid key""" + def test_configuration_updates(self): + """Test various configuration update scenarios""" with mock.patch('hostcfgd.syslog.syslog') as mock_syslog: + # Test invalid key self.mem_stat_cfg.memory_statistics_update('invalid_key', 'value') mock_syslog.assert_any_call(mock.ANY, "MemoryStatisticsCfg: Invalid key 'invalid_key' received.") - def test_memory_statistics_update_invalid_numeric_value(self): - """Test update with invalid numeric value""" - with mock.patch('hostcfgd.syslog.syslog') as mock_syslog: + # Test invalid numeric value self.mem_stat_cfg.memory_statistics_update('sampling_interval', '-1') mock_syslog.assert_any_call(mock.ANY, "MemoryStatisticsCfg: Invalid value '-1' for key 'sampling_interval'. Must be a positive integer.") + # Test update with same value (should not trigger apply_setting) + with mock.patch.object(self.mem_stat_cfg, 'apply_setting') as mock_apply: + self.mem_stat_cfg.memory_statistics_update('sampling_interval', '5') + mock_apply.assert_not_called() + # Test daemon management - @mock.patch('hostcfgd.subprocess.Popen') - @mock.patch('hostcfgd.os.kill') - def test_restart_memory_statistics_success(self, mock_kill, mock_popen): - """Test successful daemon restart""" - with mock.patch('hostcfgd.syslog.syslog') as mock_syslog: - with mock.patch.object(self.mem_stat_cfg, 'get_memory_statistics_pid', return_value=123): - self.mem_stat_cfg.restart_memory_statistics() - mock_kill.assert_called_with(123, signal.SIGTERM) - mock_popen.assert_called_once() - - @mock.patch('hostcfgd.subprocess.Popen') - def test_restart_memory_statistics_failure(self, mock_popen): - """Test failed daemon restart""" - mock_popen.side_effect = Exception("Failed to start") - with mock.patch('hostcfgd.syslog.syslog') as mock_syslog: + def test_daemon_lifecycle_management(self): + """Test daemon start, restart, and shutdown operations""" + # Test successful restart + with mock.patch('hostcfgd.subprocess.Popen') as mock_popen, \ + mock.patch('hostcfgd.os.kill') as mock_kill, \ + mock.patch.object(self.mem_stat_cfg, 'get_memory_statistics_pid', return_value=123): + self.mem_stat_cfg.restart_memory_statistics() + mock_kill.assert_called_with(123, signal.SIGTERM) + mock_popen.assert_called_once() + + # Test restart failure + with mock.patch('hostcfgd.subprocess.Popen', side_effect=Exception("Failed to start")), \ + mock.patch('hostcfgd.syslog.syslog') as mock_syslog: self.mem_stat_cfg.restart_memory_statistics() mock_syslog.assert_any_call(mock.ANY, "MemoryStatisticsCfg: Failed to start MemoryStatisticsDaemon: Failed to start") + # Test reload functionality + with mock.patch.object(self.mem_stat_cfg, 'get_memory_statistics_pid', return_value=123), \ + mock.patch('hostcfgd.os.kill') as mock_kill, \ + mock.patch('hostcfgd.syslog.syslog') as mock_syslog: + self.mem_stat_cfg.reload_memory_statistics() + mock_kill.assert_called_with(123, signal.SIGHUP) + mock_syslog.assert_any_call(mock.ANY, "MemoryStatisticsCfg: Sent SIGHUP to reload daemon configuration") + # Test PID management - def test_get_memory_statistics_pid_file_not_found(self): - """Test PID retrieval when file doesn't exist""" - with mock.patch('builtins.open', side_effect=FileNotFoundError): - with mock.patch('hostcfgd.syslog.syslog') as mock_syslog: - pid = self.mem_stat_cfg.get_memory_statistics_pid() - self.assertIsNone(pid) - mock_syslog.assert_any_call(mock.ANY, "MemoryStatisticsCfg: PID file not found. Daemon might not be running.") - - def test_get_memory_statistics_pid_invalid_content(self): - """Test PID retrieval with invalid file content""" - mock_open = mock.mock_open(read_data="invalid") - with mock.patch('builtins.open', mock_open): - with mock.patch('hostcfgd.syslog.syslog') as mock_syslog: - pid = self.mem_stat_cfg.get_memory_statistics_pid() - self.assertIsNone(pid) - mock_syslog.assert_any_call(mock.ANY, "MemoryStatisticsCfg: PID file contents invalid.") - - @mock.patch('hostcfgd.psutil.pid_exists', return_value=True) - @mock.patch('hostcfgd.psutil.Process') - def test_get_memory_statistics_pid_wrong_process(self, mock_process, mock_pid_exists): - """Test PID retrieval when process name doesn't match""" - mock_process_instance = mock.Mock() - mock_process_instance.name.return_value = "wrong_process" - mock_process.return_value = mock_process_instance - - mock_open = mock.mock_open(read_data="123") - with mock.patch('builtins.open', mock_open): - with mock.patch('hostcfgd.syslog.syslog') as mock_syslog: - pid = self.mem_stat_cfg.get_memory_statistics_pid() - self.assertIsNone(pid) - mock_syslog.assert_any_call(mock.ANY, "MemoryStatisticsCfg: PID 123 does not correspond to memory_statistics_service.py.") - - # Test daemon shutdown - @mock.patch('hostcfgd.psutil.Process') - def test_wait_for_shutdown_timeout(self, mock_process): - """Test shutdown waiting timeout""" - mock_process_instance = mock.Mock() - mock_process_instance.wait.side_effect = psutil.TimeoutExpired(123, 10) - mock_process.return_value = mock_process_instance - - with mock.patch('hostcfgd.syslog.syslog') as mock_syslog: - self.mem_stat_cfg.wait_for_shutdown(123) - mock_syslog.assert_any_call(mock.ANY, "MemoryStatisticsCfg: Timed out while waiting for daemon (PID 123) to shut down.") + def test_pid_management(self): + """Test various PID management scenarios""" + # Test PID file not found + with mock.patch('builtins.open', side_effect=FileNotFoundError), \ + mock.patch('hostcfgd.syslog.syslog') as mock_syslog: + pid = self.mem_stat_cfg.get_memory_statistics_pid() + self.assertIsNone(pid) + mock_syslog.assert_any_call(mock.ANY, "MemoryStatisticsCfg: PID file not found. Daemon might not be running.") - @mock.patch('hostcfgd.psutil.Process') - def test_wait_for_shutdown_no_process(self, mock_process): - """Test shutdown waiting when process doesn't exist""" - mock_process.side_effect = psutil.NoSuchProcess(123) - - with mock.patch('hostcfgd.syslog.syslog') as mock_syslog: - self.mem_stat_cfg.wait_for_shutdown(123) - mock_syslog.assert_any_call(mock.ANY, "MemoryStatisticsCfg: MemoryStatisticsDaemon process not found.") + # Test invalid PID file content + with mock.patch('builtins.open', mock.mock_open(read_data="invalid")), \ + mock.patch('hostcfgd.syslog.syslog') as mock_syslog: + pid = self.mem_stat_cfg.get_memory_statistics_pid() + self.assertIsNone(pid) + mock_syslog.assert_any_call(mock.ANY, "MemoryStatisticsCfg: PID file contents invalid.") + + # Test process name mismatch + mock_process = mock.Mock() + mock_process.name.return_value = "wrong_process" + with mock.patch('hostcfgd.psutil.pid_exists', return_value=True), \ + mock.patch('hostcfgd.psutil.Process', return_value=mock_process), \ + mock.patch('builtins.open', mock.mock_open(read_data="123")), \ + mock.patch('hostcfgd.syslog.syslog') as mock_syslog: + pid = self.mem_stat_cfg.get_memory_statistics_pid() + self.assertIsNone(pid) + mock_syslog.assert_any_call(mock.ANY, "MemoryStatisticsCfg: PID 123 does not correspond to memory_statistics_service.py.") # Test enable/disable functionality - def test_memory_statistics_enable(self): - """Test enabling memory statistics""" + def test_enable_disable_functionality(self): + """Test enabling and disabling memory statistics""" + # Test enabling with mock.patch.object(self.mem_stat_cfg, 'restart_memory_statistics') as mock_restart: self.mem_stat_cfg.memory_statistics_update('enabled', 'true') mock_restart.assert_called_once() self.assertEqual(self.mem_stat_cfg.cache['enabled'], 'true') - def test_memory_statistics_update_same_value(self): - """Test update with same value (should not trigger apply_setting)""" - with mock.patch.object(self.mem_stat_cfg, 'apply_setting') as mock_apply: - self.mem_stat_cfg.memory_statistics_update('sampling_interval', '5') - mock_apply.assert_not_called() - - def test_memory_statistics_update_exception_handling(self): - """Test exception handling in memory_statistics_update""" - with mock.patch.object(self.mem_stat_cfg, 'apply_setting', side_effect=Exception("Test error")): - with mock.patch('hostcfgd.syslog.syslog') as mock_syslog: - self.mem_stat_cfg.memory_statistics_update('enabled', 'true') - mock_syslog.assert_any_call(mock.ANY, "MemoryStatisticsCfg: Failed to manage MemoryStatisticsDaemon: Test error") - - def test_apply_setting_with_non_enabled_key(self): - """Test apply_setting with sampling_interval or retention_period""" - with mock.patch.object(self.mem_stat_cfg, 'reload_memory_statistics') as mock_reload: - self.mem_stat_cfg.apply_setting('sampling_interval', '10') - mock_reload.assert_called_once() - - def test_apply_setting_with_enabled_false(self): - """Test apply_setting with enabled=false""" - with mock.patch.object(self.mem_stat_cfg, 'shutdown_memory_statistics') as mock_shutdown: - self.mem_stat_cfg.apply_setting('enabled', 'false') - mock_shutdown.assert_called_once() - - def test_apply_setting_exception(self): - """Test exception handling in apply_setting""" - with mock.patch.object(self.mem_stat_cfg, 'restart_memory_statistics', - side_effect=Exception("Test error")): - with mock.patch('hostcfgd.syslog.syslog') as mock_syslog: - self.mem_stat_cfg.apply_setting('enabled', 'true') - mock_syslog.assert_any_call(mock.ANY, - "MemoryStatisticsCfg: Exception in apply_setting() for key 'enabled': Test error") - - @mock.patch('hostcfgd.psutil.pid_exists', return_value=False) - def test_get_memory_statistics_pid_nonexistent(self, mock_pid_exists): - """Test get_memory_statistics_pid when PID doesn't exist""" - mock_open = mock.mock_open(read_data="123") - with mock.patch('builtins.open', mock_open): - with mock.patch('hostcfgd.syslog.syslog') as mock_syslog: - pid = self.mem_stat_cfg.get_memory_statistics_pid() - self.assertIsNone(pid) - mock_syslog.assert_any_call(mock.ANY, "MemoryStatisticsCfg: PID does not exist.") - - @mock.patch('hostcfgd.psutil.Process') - def test_get_memory_statistics_pid_exception(self, mock_process): - """Test general exception handling in get_memory_statistics_pid""" - mock_process.side_effect = Exception("Unexpected error") - mock_open = mock.mock_open(read_data="123") - - with mock.patch('hostcfgd.psutil.pid_exists', return_value=True): - with mock.patch('builtins.open', mock_open): - with mock.patch('hostcfgd.syslog.syslog') as mock_syslog: - pid = self.mem_stat_cfg.get_memory_statistics_pid() - self.assertIsNone(pid) - mock_syslog.assert_any_call(mock.ANY, - "MemoryStatisticsCfg: Exception failed to retrieve MemoryStatisticsDaemon PID: Unexpected error") - - def test_memory_statistics_handler(self): - """Test memory_statistics_handler in HostConfigDaemon""" - daemon = hostcfgd.HostConfigDaemon() - with mock.patch.object(daemon.memorystatisticscfg, 'memory_statistics_update') as mock_update: - daemon.memory_statistics_handler('enabled', None, 'true') - mock_update.assert_called_once_with('enabled', 'true') - - def test_memory_statistics_handler_exception(self): - """Test exception handling in memory_statistics_handler""" - daemon = hostcfgd.HostConfigDaemon() - with mock.patch.object(daemon.memorystatisticscfg, 'memory_statistics_update', - side_effect=Exception("Handler error")): - with mock.patch('hostcfgd.syslog.syslog') as mock_syslog: - daemon.memory_statistics_handler('enabled', None, 'true') - mock_syslog.assert_any_call(mock.ANY, - "MemoryStatisticsCfg: Error while handling memory statistics update: Handler error") - - @mock.patch('hostcfgd.psutil.Process') - def test_wait_for_shutdown_general_exception(self, mock_process): - """Test general exception handling in wait_for_shutdown""" - mock_process.side_effect = Exception("Unexpected shutdown error") - with mock.patch('hostcfgd.syslog.syslog') as mock_syslog: - self.mem_stat_cfg.wait_for_shutdown(123) - mock_syslog.assert_any_call(mock.ANY, - "MemoryStatisticsCfg: Exception in wait_for_shutdown(): Unexpected shutdown error") - - def test_memory_statistics_disable(self): - """Test disabling memory statistics""" - # First set the initial state to enabled - self.mem_stat_cfg.cache['enabled'] = 'true' - - # Mock the apply_setting method to avoid actual system calls - with mock.patch.object(self.mem_stat_cfg, 'apply_setting') as mock_apply: - self.mem_stat_cfg.memory_statistics_update('enabled', 'false') - - # Verify apply_setting was called with correct parameters - mock_apply.assert_called_once_with('enabled', 'false') - - # Verify the cache was updated - self.assertEqual(self.mem_stat_cfg.cache['enabled'], 'false') - - def test_memory_statistics_disable_with_shutdown(self): - """Test disabling memory statistics with full shutdown chain""" - # First set the initial state to enabled + # Test disabling with running daemon self.mem_stat_cfg.cache['enabled'] = 'true' - - # Mock both get_memory_statistics_pid and os.kill to simulate full shutdown - with mock.patch.object(self.mem_stat_cfg, 'get_memory_statistics_pid', return_value=123) as mock_get_pid, \ + with mock.patch.object(self.mem_stat_cfg, 'get_memory_statistics_pid', return_value=123), \ mock.patch('hostcfgd.os.kill') as mock_kill, \ mock.patch.object(self.mem_stat_cfg, 'wait_for_shutdown') as mock_wait: - self.mem_stat_cfg.memory_statistics_update('enabled', 'false') - - # Verify the shutdown sequence - mock_get_pid.assert_called_once() mock_kill.assert_called_once_with(123, signal.SIGTERM) mock_wait.assert_called_once_with(123) - - # Verify the cache was updated self.assertEqual(self.mem_stat_cfg.cache['enabled'], 'false') - def test_memory_statistics_disable_no_running_daemon(self): - """Test disabling memory statistics when daemon is not running""" - # First set the initial state to enabled + # Test disabling with no running daemon self.mem_stat_cfg.cache['enabled'] = 'true' - - # Mock get_memory_statistics_pid to return None (no running daemon) with mock.patch.object(self.mem_stat_cfg, 'get_memory_statistics_pid', return_value=None) as mock_get_pid: self.mem_stat_cfg.memory_statistics_update('enabled', 'false') - - # Verify get_pid was called mock_get_pid.assert_called_once() - - # Verify the cache was updated despite no running daemon self.assertEqual(self.mem_stat_cfg.cache['enabled'], 'false') + # Test error handling + def test_error_handling(self): + """Test various error handling scenarios""" + # Test memory_statistics_update exception handling + with mock.patch.object(self.mem_stat_cfg, 'apply_setting', side_effect=Exception("Test error")), \ + mock.patch('hostcfgd.syslog.syslog') as mock_syslog: + self.mem_stat_cfg.memory_statistics_update('enabled', 'true') + mock_syslog.assert_any_call(mock.ANY, "MemoryStatisticsCfg: Failed to manage MemoryStatisticsDaemon: Test error") - - - - def test_reload_memory_statistics_success(self): - """Test successful reload of memory statistics daemon""" - with mock.patch.object(self.mem_stat_cfg, 'get_memory_statistics_pid', return_value=123) as mock_get_pid, \ - mock.patch('hostcfgd.os.kill') as mock_kill, \ - mock.patch('hostcfgd.syslog.syslog') as mock_syslog: - - self.mem_stat_cfg.reload_memory_statistics() - - # Verify the method calls - mock_get_pid.assert_called_once() - mock_kill.assert_called_once_with(123, signal.SIGHUP) - mock_syslog.assert_any_call(mock.ANY, "MemoryStatisticsCfg: Sent SIGHUP to reload daemon configuration") - - def test_reload_memory_statistics_no_pid(self): - """Test reload when no daemon is running""" - with mock.patch.object(self.mem_stat_cfg, 'get_memory_statistics_pid', return_value=None) as mock_get_pid, \ - mock.patch('hostcfgd.os.kill') as mock_kill: - - self.mem_stat_cfg.reload_memory_statistics() - - # Verify the method calls - mock_get_pid.assert_called_once() - mock_kill.assert_not_called() - - def test_reload_memory_statistics_failure(self): - """Test reload failure with exception""" - with mock.patch.object(self.mem_stat_cfg, 'get_memory_statistics_pid', return_value=123) as mock_get_pid, \ - mock.patch('hostcfgd.os.kill', side_effect=Exception("Test error")), \ - mock.patch('hostcfgd.syslog.syslog') as mock_syslog: - - self.mem_stat_cfg.reload_memory_statistics() - - mock_get_pid.assert_called_once() - mock_syslog.assert_any_call(mock.ANY, "MemoryStatisticsCfg: Failed to reload MemoryStatisticsDaemon: Test error") - - def test_shutdown_memory_statistics_success(self): - """Test successful shutdown of memory statistics daemon""" - with mock.patch.object(self.mem_stat_cfg, 'get_memory_statistics_pid', return_value=123) as mock_get_pid, \ - mock.patch('hostcfgd.os.kill') as mock_kill, \ - mock.patch.object(self.mem_stat_cfg, 'wait_for_shutdown') as mock_wait, \ - mock.patch('hostcfgd.syslog.syslog') as mock_syslog: - - self.mem_stat_cfg.shutdown_memory_statistics() - - mock_get_pid.assert_called_once() - mock_kill.assert_called_once_with(123, signal.SIGTERM) - mock_wait.assert_called_once_with(123) - mock_syslog.assert_any_call(mock.ANY, "MemoryStatisticsCfg: Sent SIGTERM to stop MemoryStatisticsDaemon") - - def test_shutdown_memory_statistics_failure(self): - """Test shutdown failure with exception""" - with mock.patch.object(self.mem_stat_cfg, 'get_memory_statistics_pid', return_value=123) as mock_get_pid, \ - mock.patch('hostcfgd.os.kill', side_effect=Exception("Test error")), \ - mock.patch('hostcfgd.syslog.syslog') as mock_syslog: - - self.mem_stat_cfg.shutdown_memory_statistics() - - mock_get_pid.assert_called_once() - mock_syslog.assert_any_call(mock.ANY, "MemoryStatisticsCfg: Failed to shutdown MemoryStatisticsDaemon: Test error") - - def test_wait_for_shutdown_success(self): - """Test successful wait for shutdown""" + # Test shutdown timeout mock_process = mock.Mock() - with mock.patch('hostcfgd.psutil.Process', return_value=mock_process) as mock_process_class, \ - mock.patch('hostcfgd.syslog.syslog') as mock_syslog: - + mock_process.wait.side_effect = psutil.TimeoutExpired(123, 10) + with mock.patch('hostcfgd.psutil.Process', return_value=mock_process), \ + mock.patch('hostcfgd.syslog.syslog') as mock_syslog: self.mem_stat_cfg.wait_for_shutdown(123) - - mock_process_class.assert_called_once_with(123) - mock_process.wait.assert_called_once_with(timeout=10) - mock_syslog.assert_any_call(mock.ANY, "MemoryStatisticsCfg: MemoryStatisticsDaemon stopped gracefully") - - def test_main_function(self): - """Test the main function""" - with mock.patch('hostcfgd.HostConfigDaemon') as mock_daemon: - hostcfgd.main() - - # Verify daemon was instantiated and started - mock_daemon.assert_called_once() - mock_instance = mock_daemon.return_value - mock_instance.register_callbacks.assert_called_once() - mock_instance.start.assert_called_once() - - def test_process_name_mismatch(self): - """Test when process name doesn't match expected daemon name""" - mock_process = mock.Mock() - mock_process.name.return_value = "wrong_process_name" + mock_syslog.assert_any_call(mock.ANY, "MemoryStatisticsCfg: Timed out while waiting for daemon (PID 123) to shut down.") + + # Test process not found during shutdown + with mock.patch('hostcfgd.psutil.Process', side_effect=psutil.NoSuchProcess(123)), \ + mock.patch('hostcfgd.syslog.syslog') as mock_syslog: + self.mem_stat_cfg.wait_for_shutdown(123) + mock_syslog.assert_any_call(mock.ANY, "MemoryStatisticsCfg: MemoryStatisticsDaemon process not found.") + + def test_handler_functionality(self): + """Test HostConfigDaemon handler functionality""" + daemon = hostcfgd.HostConfigDaemon() - with mock.patch('builtins.open', mock.mock_open(read_data="123")), \ - mock.patch('hostcfgd.psutil.pid_exists', return_value=True), \ - mock.patch('hostcfgd.psutil.Process', return_value=mock_process), \ - mock.patch('hostcfgd.syslog.syslog') as mock_syslog: - - pid = self.mem_stat_cfg.get_memory_statistics_pid() - - self.assertIsNone(pid) - mock_syslog.assert_any_call(mock.ANY, "MemoryStatisticsCfg: PID 123 does not correspond to memory_statistics_service.py.") \ No newline at end of file + # Test normal handler operation + with mock.patch.object(daemon.memorystatisticscfg, 'memory_statistics_update') as mock_update: + daemon.memory_statistics_handler('enabled', None, 'true') + mock_update.assert_called_once_with('enabled', 'true') + + # Test handler exception + with mock.patch.object(daemon.memorystatisticscfg, 'memory_statistics_update', + side_effect=Exception("Handler error")), \ + mock.patch('hostcfgd.syslog.syslog') as mock_syslog: + daemon.memory_statistics_handler('enabled', None, 'true') + mock_syslog.assert_any_call(mock.ANY, + "MemoryStatisticsCfg: Error while handling memory statistics update: Handler error") \ No newline at end of file