diff --git a/tests/hostcfgd/hostcfgd_test.py b/tests/hostcfgd/hostcfgd_test.py index 68739c3b..25f3bf0c 100644 --- a/tests/hostcfgd/hostcfgd_test.py +++ b/tests/hostcfgd/hostcfgd_test.py @@ -373,372 +373,6 @@ 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'] = { -# '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.""" @@ -758,27 +392,27 @@ def tearDown(self): # Test initial loading def test_load_with_invalid_key(self): - """Test loading configuration with invalid key.""" + """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.""" + """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.") + 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.""" + """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.""" + """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.") @@ -787,7 +421,7 @@ def test_memory_statistics_update_invalid_numeric_value(self): @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.""" + """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() @@ -796,15 +430,15 @@ def test_restart_memory_statistics_success(self, mock_kill, mock_popen): @mock.patch('hostcfgd.subprocess.Popen') def test_restart_memory_statistics_failure(self, mock_popen): - """Test failed daemon restart.""" + """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.") + 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.""" + """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() @@ -812,7 +446,7 @@ def test_get_memory_statistics_pid_file_not_found(self): 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.""" + """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: @@ -823,11 +457,11 @@ def test_get_memory_statistics_pid_invalid_content(self): @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.""" + """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: @@ -838,51 +472,170 @@ def test_get_memory_statistics_pid_wrong_process(self, mock_process, mock_pid_ex # Test daemon shutdown @mock.patch('hostcfgd.psutil.Process') def test_wait_for_shutdown_timeout(self, mock_process): - """Test shutdown waiting timeout.""" + """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.""" + """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.""" + """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.""" + """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_no_running_daemon(self): - """Test disabling memory statistics when no daemon is running.""" + 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, \ @@ -980,3 +733,250 @@ def test_process_name_mismatch(self): 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.""" + +# def setUp(self): +# # Initial configuration for Memory Statistics +# MockConfigDb.CONFIG_DB['MEMORY_STATISTICS'] = { +# '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.") + +# 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_disable(self): +# """Test disabling memory statistics.""" +# self.mem_stat_cfg.cache['enabled'] = 'true' + +# with mock.patch.object(self.mem_stat_cfg, 'apply_setting') as mock_apply: +# self.mem_stat_cfg.memory_statistics_update('enabled', 'false') +# mock_apply.assert_called_once_with('enabled', 'false') +# self.assertEqual(self.mem_stat_cfg.cache['enabled'], 'false') + +# def test_memory_statistics_disable_no_running_daemon(self): +# """Test disabling memory statistics when no daemon is running.""" +# self.mem_stat_cfg.cache['enabled'] = 'true' + +# 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') +# mock_get_pid.assert_called_once() +# 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.")