Coverage: Increase code coverage - TimerManager and ServiceManager

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__':

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__':