diff --git a/src/core/tests/Test_ServiceManager.py b/src/core/tests/Test_ServiceManager.py index 7de69a68..0eb406ff 100644 --- a/src/core/tests/Test_ServiceManager.py +++ b/src/core/tests/Test_ServiceManager.py @@ -21,10 +21,13 @@ from core.src.core_logic.ServiceManager import ServiceManager from core.src.core_logic.ServiceManager import ServiceInfo + class TestServiceManager(unittest.TestCase): def setUp(self): self.runtime = RuntimeCompositor(ArgumentComposer().get_composed_arguments(), legacy_mode=True) - self.container = self.runtime.container + self.service_manager = ServiceManager(self.runtime.env_layer, self.runtime.execution_config, self.runtime.composite_logger, self.runtime.telemetry_writer,ServiceInfo("AutoAssessment", "Auto assessment service", "path")) + self.service_manager.service_name = "test_service" + self.mock_systemd_service_unit_path = "/etc/systemd/system/{0}.service" def tearDown(self): self.runtime.stop() @@ -33,17 +36,136 @@ def mock_run_command_to_set_service_file_permission(self, cmd, no_output=False, parts = cmd.split() if parts[1] == "chmod" and parts[2] == "644": return 0, "permissions set" - else: - raise Exception - + def mock_write_with_retry_valid(self, file_path_or_handle, data, mode='a+'): return - + + def mock_invoke_systemctl(self, command, description): + self.service_manager.invoke_systemctl_called = True + + if "start" in command and "restart" not in command: + return 0, "Service started" + elif "stop" in command: + return 0, "Service stopped" + elif "reload-or-restart" in command: + return 0, "Reloading the service" + elif "status" in command: + return 0, "Getting the service status" + elif "enable" in command or "is-enabled" in command: + return 0, "Enabling the service or Checking Service is enabled" + elif "disable" in command: + return 0, "Disabling the service" + elif "is-active" in command: + return 0, "Checking if service is active" + def test_create_service_unit_file(self): - service_manager = ServiceManager(self.runtime.env_layer, self.runtime.execution_config, self.runtime.composite_logger, self.runtime.telemetry_writer, ServiceInfo("AutoAssessment","Auto assessment service","path")) - service_manager.env_layer.run_command_output = self.mock_run_command_to_set_service_file_permission - service_manager.env_layer.file_system.write_with_retry = self.mock_write_with_retry_valid - service_manager.create_service_unit_file(exec_start="/bin/bash " + service_manager.service_exec_path, desc="Microsoft Azure Linux Patch Extension - Auto Assessment") + self.service_manager.env_layer.run_command_output = self.mock_run_command_to_set_service_file_permission + self.service_manager.env_layer.file_system.write_with_retry = self.mock_write_with_retry_valid + self.service_manager.create_service_unit_file(exec_start="/bin/bash " + self.service_manager.service_exec_path, desc="Microsoft Azure Linux Patch Extension - Auto Assessment") + + def test_start_service(self): + # Set method calls + self.service_manager.invoke_systemctl_called = False + + self.service_manager.invoke_systemctl = self.mock_invoke_systemctl + + # Act + result = self.service_manager.start_service() + + # Assert + self.assertTrue(result, "Service should be started") + self.assertTrue(self.service_manager.invoke_systemctl_called) + + def test_reload_service(self): + # Set method calls + self.service_manager.invoke_systemctl_called = False + + self.service_manager.invoke_systemctl = self.mock_invoke_systemctl + + # Act + result = self.service_manager.reload_service() + + # Assert + self.assertTrue(result, "Reloading the service.") + self.assertTrue(self.service_manager.invoke_systemctl_called) + + def test_stop_service(self): + # Set method calls + self.service_manager.invoke_systemctl_called = False + + self.service_manager.invoke_systemctl = self.mock_invoke_systemctl + + # Act + result = self.service_manager.stop_service() + + # Assert + self.assertTrue(result, "Service should not be started") + self.assertTrue(self.service_manager.invoke_systemctl_called) + + def test_get_service_status(self): + # Set method calls + self.service_manager.invoke_systemctl_called = False + + self.service_manager.invoke_systemctl = self.mock_invoke_systemctl + + # Act + result = self.service_manager.get_service_status() + + # Assert + self.assertTrue(result, "Getting the service status") + self.assertTrue(self.service_manager.invoke_systemctl_called) + + def test_enable_service(self): + # Set method calls + self.service_manager.invoke_systemctl_called = False + + self.service_manager.invoke_systemctl = self.mock_invoke_systemctl + + # Act + result = self.service_manager.enable_service() + + # Assert + self.assertTrue(result, "Enabling the service") + self.assertTrue(self.service_manager.invoke_systemctl_called) + + def test_disable_service(self): + # Set method calls + self.service_manager.invoke_systemctl_called = False + + self.service_manager.invoke_systemctl = self.mock_invoke_systemctl + + # Act + result = self.service_manager.disable_service() + + # Assert + self.assertTrue(result, "Disabling the service") + self.assertTrue(self.service_manager.invoke_systemctl_called) + + def test_is_service_active(self): + # Set method calls + self.service_manager.invoke_systemctl_called = False + + self.service_manager.invoke_systemctl = self.mock_invoke_systemctl + + # Act + result = self.service_manager.is_service_active() + + # Assert + self.assertTrue(result, "Checking if service is active") + self.assertTrue(self.service_manager.invoke_systemctl_called) + + def test_is_service_enabled(self): + # Set method calls + self.service_manager.invoke_systemctl_called = False + + self.service_manager.invoke_systemctl = self.mock_invoke_systemctl + + # Act + result = self.service_manager.is_service_enabled() + + # Assert + self.assertTrue(result, "Checking if service is enabled") + self.assertTrue(self.service_manager.invoke_systemctl_called) if __name__ == '__main__': diff --git a/src/core/tests/Test_TimerManager.py b/src/core/tests/Test_TimerManager.py index 8dcf49b4..4034e57f 100644 --- a/src/core/tests/Test_TimerManager.py +++ b/src/core/tests/Test_TimerManager.py @@ -24,7 +24,9 @@ class TestTimerManager(unittest.TestCase): def setUp(self): self.runtime = RuntimeCompositor(ArgumentComposer().get_composed_arguments(), legacy_mode=True) - self.container = self.runtime.container + self.service_manager = TimerManager(self.runtime.env_layer, self.runtime.execution_config, self.runtime.composite_logger, self.runtime.telemetry_writer, ServiceInfo("AutoAssessment","Auto assessment service","path")) + self.service_manager.service_name = "test_service" + self.mock_systemd_timer_unit_path = "/etc/systemd/system/{0}.timer" def tearDown(self): self.runtime.stop() @@ -33,17 +35,127 @@ def mock_run_command_to_set_service_file_permission(self, cmd, no_output=False, parts = cmd.split() if parts[1] == "chmod" and parts[2] == "644": return 0, "permissions set" - else: - raise Exception - + def mock_write_with_retry_valid(self, file_path_or_handle, data, mode='a+'): return - + + def mock_invoke_systemctl(self, command, description): + self.service_manager.invoke_systemctl_called = True + if "start" in command and "restart" not in command: + return 0, "Timer started" + elif "stop" in command: + return 0, "Timer stopped" + elif "reload-or-restart" in command: + return 0, "Reloading the timer" + elif "status" in command: + return 0, "Getting the timer status" + elif "enable" in command or "is-enable" in command: + return 0, "Enabling the timer or checking timer is enabled" + elif "disable" in command: + return 0, "Disable the timer" + elif "is-active" in command: + return 0, "Checking if timer is active" + def test_create_timer_unit_file(self): - service_manager = TimerManager(self.runtime.env_layer, self.runtime.execution_config, self.runtime.composite_logger, self.runtime.telemetry_writer, ServiceInfo("AutoAssessment","Auto assessment service","path")) - service_manager.env_layer.run_command_output = self.mock_run_command_to_set_service_file_permission - service_manager.env_layer.file_system.write_with_retry = self.mock_write_with_retry_valid - service_manager.create_timer_unit_file(desc="Microsoft Azure Linux Patch Extension - Auto Assessment Timer") + self.service_manager.env_layer.run_command_output = self.mock_run_command_to_set_service_file_permission + self.service_manager.env_layer.file_system.write_with_retry = self.mock_write_with_retry_valid + self.service_manager.create_timer_unit_file(desc="Microsoft Azure Linux Patch Extension - Auto Assessment Timer") + + def test_start_timer(self): + # Set method calls + self.service_manager.invoke_systemctl_called = False + self.service_manager.invoke_systemctl = self.mock_invoke_systemctl + + # Act + result = self.service_manager.start_timer() + + # Assert + self.assertTrue(result, "Timer should be started") + self.assertTrue(self.service_manager.invoke_systemctl_called) + + def test_stop_timer(self): + # Set method calls + self.service_manager.invoke_systemctl_called = False + self.service_manager.invoke_systemctl = self.mock_invoke_systemctl + + # Act + result = self.service_manager.stop_timer() + + # Assert + self.assertTrue(result, "Timer should be stopped") + self.assertTrue(self.service_manager.invoke_systemctl_called) + + def test_reload_timer(self): + # Set method calls + self.service_manager.invoke_systemctl_called = False + self.service_manager.invoke_systemctl = self.mock_invoke_systemctl + + # Act + result = self.service_manager.reload_timer() + + # Assert + self.assertTrue(result, "Reload timer") + self.assertTrue(self.service_manager.invoke_systemctl_called) + + def test_enable_timer(self): + # Set method calls + self.service_manager.invoke_systemctl_called = False + self.service_manager.invoke_systemctl = self.mock_invoke_systemctl + + # Act + result = self.service_manager.enable_timer() + + # Assert + self.assertTrue(result, "Timer should be enabled") + self.assertTrue(self.service_manager.invoke_systemctl_called) + + def test_disable_timer(self): + # Set method calls + self.service_manager.invoke_systemctl_called = False + self.service_manager.invoke_systemctl = self.mock_invoke_systemctl + + # Act + result = self.service_manager.disable_timer() + + # Assert + self.assertTrue(result, "Timer should be disabled") + self.assertTrue(self.service_manager.invoke_systemctl_called) + + def test_get_timer_status(self): + # Set method calls + self.service_manager.invoke_systemctl_called = False + self.service_manager.invoke_systemctl = self.mock_invoke_systemctl + + # Act + result = self.service_manager.get_timer_status() + + # Assert + self.assertTrue(result, "Get timer status") + self.assertTrue(self.service_manager.invoke_systemctl_called) + + def test_timer_is_enable(self): + # Set method calls + self.service_manager.invoke_systemctl_called = False + self.service_manager.invoke_systemctl = self.mock_invoke_systemctl + + # Act + result = self.service_manager.is_timer_enabled() + + # Assert + self.assertTrue(result, "Check if timer is enable") + self.assertTrue(self.service_manager.invoke_systemctl_called) + + def test_timer_is_active(self): + # Set method calls + self.service_manager.invoke_systemctl_called = False + self.service_manager.invoke_systemctl = self.mock_invoke_systemctl + + # Act + result = self.service_manager.is_timer_active() + + # Assert + self.assertTrue(result, "Check if timer is active") + self.assertTrue(self.service_manager.invoke_systemctl_called) if __name__ == '__main__':