Merge branch 'master' of https://github.com/Azure/LinuxPatchExtension …

…into fix/tele_ascii_key_test

pull master changes.

src/core/tests/Test_ConfigurePatchingProcessor.py

@@ -17,6 +17,9 @@
 import os
 import re
 import unittest
+import sys
+from io import StringIO
+
 from core.src.CoreMain import CoreMain
 from core.src.bootstrap.Constants import Constants
 from core.tests.library.ArgumentComposer import ArgumentComposer
@@ -42,6 +45,8 @@ def mock_package_manager_get_current_auto_os_patch_state_returns_unknown(self):
             return Constants.AutomaticOSPatchStates.DISABLED
         else:
             return Constants.AutomaticOSPatchStates.UNKNOWN
+    def mock_get_current_auto_os_patch_state(self):
+        raise Exception("Mocked Exception")
     #endregion Mocks
 
     def test_operation_success_for_configure_patching_request_for_apt_with_default_updates_config(self):
@@ -309,6 +314,63 @@ def test_configure_patching_with_patch_mode_and_assessment_mode_by_platform(self
         # stop test runtime
         runtime.stop()
 
+    def test_configure_patching_raise_exception_auto_os_patch_state(self):
+        # arrange capture std IO
+        captured_output = StringIO()
+        sys.stdout = captured_output
+
+        argument_composer = ArgumentComposer()
+        argument_composer.operation = Constants.CONFIGURE_PATCHING
+        argument_composer.patch_mode = Constants.PatchModes.AUTOMATIC_BY_PLATFORM
+        argument_composer.assessment_mode = Constants.AssessmentModes.AUTOMATIC_BY_PLATFORM
+        runtime = RuntimeCompositor(argument_composer.get_composed_arguments(), True, Constants.APT)
+        runtime.package_manager.get_current_auto_os_patch_state = runtime.backup_get_current_auto_os_patch_state
+        runtime.set_legacy_test_type('HappyPath')
+
+        # mock swap
+        backup_package_manager_get_current_auto_os_patch_state = runtime.package_manager.get_current_auto_os_patch_state
+        runtime.package_manager.get_current_auto_os_patch_state = self.mock_get_current_auto_os_patch_state
+
+        runtime.configure_patching_processor.start_configure_patching()
+
+        # restore sdt.out ouptput
+        sys.stdout = sys.__stdout__
+
+        # assert
+        output = captured_output.getvalue()
+        self.assertIn("Error while processing patch mode configuration", output)
+
+        # check status file
+        with runtime.env_layer.file_system.open(runtime.execution_config.status_file_path, 'r') as file_handle:
+            substatus_file_data = json.load(file_handle)[0]["status"]["substatus"]
+        self.assertEqual(len(substatus_file_data), 1)
+        self.assertTrue(substatus_file_data[0]["name"] == Constants.CONFIGURE_PATCHING_SUMMARY)
+        self.assertTrue(substatus_file_data[0]["status"].lower() == Constants.STATUS_TRANSITIONING.lower())
+
+        # restore
+        runtime.package_manager.get_current_auto_os_patch_state = backup_package_manager_get_current_auto_os_patch_state
+
+        runtime.stop()
+
+    def test_configure_patching_raise_exception_auto_assessment_systemd(self):
+        argument_composer = ArgumentComposer()
+        argument_composer.operation = Constants.CONFIGURE_PATCHING
+        argument_composer.patch_mode = Constants.PatchModes.AUTOMATIC_BY_PLATFORM
+        argument_composer.assessment_mode = Constants.AssessmentModes.AUTOMATIC_BY_PLATFORM
+        runtime = RuntimeCompositor(argument_composer.get_composed_arguments(), True, Constants.APT)
+        runtime.set_legacy_test_type('HappyPath')
+
+        # mock swap
+        back_up_auto_assess_service_manager = runtime.configure_patching_processor.auto_assess_service_manager.systemd_exists
+        runtime.configure_patching_processor.auto_assess_service_manager.systemd_exists = lambda: False
+
+        self.assertRaises(Exception, runtime.configure_patching_processor.start_configure_patching())
+
+        # restore
+        runtime.configure_patching_processor.auto_assess_service_manager.systemd_exists = back_up_auto_assess_service_manager
+
+        runtime.stop()
+
     def __check_telemetry_events(self, runtime):
         all_events = os.listdir(runtime.telemetry_writer.events_folder_path)
         self.assertTrue(len(all_events) > 0)

src/core/tests/Test_MaintenanceWindow.py

@@ -15,6 +15,8 @@
 # Requires Python 2.7+
 
 import datetime
+from io import StringIO
+import sys
 import unittest
 from core.tests.library.ArgumentComposer import ArgumentComposer
 from core.tests.library.RuntimeCompositor import RuntimeCompositor
@@ -62,6 +64,43 @@ def test_RemainingTime_after_duration_complete(self):
         self.assertEqual(int(remaining_time), 0)
         runtime.stop()
 
+    def test_RemainingTime_log_to_stdout_true(self):
+        # Arrange, Capture stdout
+        captured_output = StringIO()
+        sys.stdout = captured_output  # Redirect stdout to the StringIO object
+
+        argument_composer = ArgumentComposer()
+        argument_composer.start_time = "2017-02-15T18:15:12.9828835Z"
+        argument_composer.maximum_duration = "PT1H"
+        runtime = RuntimeCompositor(argument_composer.get_composed_arguments(), True)
+
+        current_time = datetime.datetime.strptime('2017-02-15 18:30:20', "%Y-%m-%d %H:%M:%S")
+        remaining_time = runtime.maintenance_window.get_remaining_time_in_minutes(current_time, log_to_stdout=True)
+
+        # Restore stdout
+        sys.stdout = sys.__stdout__
+
+        # Assert
+        output = captured_output.getvalue()
+        self.assertEqual(int(remaining_time), 44)
+        self.assertIn("Maintenance Window Utilization:", output)  # Verify the log output contains the expected text
+
+        runtime.stop()
+
+    def test_RemainingTime_raise_exception(self):
+        # Arrange
+        argument_composer = ArgumentComposer()
+        argument_composer.start_time = "Invalid datetime format"
+        argument_composer.maximum_duration = "PT1H"
+        runtime = RuntimeCompositor(argument_composer.get_composed_arguments(), True)
+
+        # Assert
+        with self.assertRaises(ValueError) as context:
+            runtime.maintenance_window.get_remaining_time_in_minutes()
+        self.assertIn("Invalid datetime format", str(context.exception))
+
+        runtime.stop()
+
     def test_check_available_time(self):
         argument_composer = ArgumentComposer()
         argument_composer.start_time = (datetime.datetime.utcnow() - datetime.timedelta(minutes=39)).strftime("%Y-%m-%dT%H:%M:%S.9999Z")