diff --git a/src/core/tests/Test_ConfigurePatchingProcessor.py b/src/core/tests/Test_ConfigurePatchingProcessor.py index e953a839..76db40e3 100644 --- a/src/core/tests/Test_ConfigurePatchingProcessor.py +++ b/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) diff --git a/src/core/tests/Test_MaintenanceWindow.py b/src/core/tests/Test_MaintenanceWindow.py index ae7c6c84..e9e0fb29 100644 --- a/src/core/tests/Test_MaintenanceWindow.py +++ b/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")