From 44755039f1e64b4840d8983672d9137666f2e539 Mon Sep 17 00:00:00 2001 From: Koshy John Date: Mon, 15 Jun 2026 13:38:46 -0700 Subject: [PATCH 1/6] Engg. Hygiene: Introduce Extension Status Asserter and use in one Test class --- .../core_logic/ConfigurePatchingProcessor.py | 43 ++++- src/core/src/core_logic/ServiceManager.py | 1 + .../tests/Test_ConfigurePatchingProcessor.py | 175 +++++++---------- src/core/tests/library/ExtStatusAsserter.py | 179 ++++++++++++++++++ 4 files changed, 283 insertions(+), 115 deletions(-) create mode 100644 src/core/tests/library/ExtStatusAsserter.py diff --git a/src/core/src/core_logic/ConfigurePatchingProcessor.py b/src/core/src/core_logic/ConfigurePatchingProcessor.py index e7ed36f2..496be34c 100644 --- a/src/core/src/core_logic/ConfigurePatchingProcessor.py +++ b/src/core/src/core_logic/ConfigurePatchingProcessor.py @@ -17,9 +17,18 @@ """ Configure Patching """ from core.src.bootstrap.Constants import Constants +from core.src.bootstrap.EnvLayer import EnvLayer +from core.src.core_logic.ExecutionConfig import ExecutionConfig +from core.src.local_loggers.CompositeLogger import CompositeLogger +from core.src.service_interfaces.StatusHandler import StatusHandler +from core.src.package_managers.PackageManager import PackageManager +from core.src.core_logic.ServiceManager import ServiceManager +from core.src.core_logic.TimerManager import TimerManager + class ConfigurePatchingProcessor(object): def __init__(self, env_layer, execution_config, composite_logger, telemetry_writer, status_handler, package_manager, auto_assess_service_manager, auto_assess_timer_manager, lifecycle_manager): + # type: (EnvLayer, ExecutionConfig, CompositeLogger, TelemetryWriter, StatusHandler, PackageManager, ServiceManager, TimerManager) -> None self.env_layer = env_layer self.execution_config = execution_config @@ -40,7 +49,7 @@ def __init__(self, env_layer, execution_config, composite_logger, telemetry_writ def start_configure_patching(self): """ Start configure patching """ try: - self.composite_logger.log("\nStarting configure patching... [MachineId: " + self.env_layer.platform.vm_name() + "][ActivityId: " + self.execution_config.activity_id + "][StartTime: " + self.execution_config.start_time + "]") + self.composite_logger.log("[CPP] Starting configure patching... [MachineId: " + self.env_layer.platform.vm_name() + "][ActivityId: " + self.execution_config.activity_id + "][StartTime: " + self.execution_config.start_time + "]") self.status_handler.set_current_operation(Constants.CONFIGURE_PATCHING) self.__raise_if_telemetry_unsupported() @@ -73,6 +82,7 @@ def set_configure_patching_final_overall_status(self): def __try_set_patch_mode(self): """ Set the patch mode for the VM """ try: + self.composite_logger.log_verbose("[CPP] Processing patch mode configuration...") self.status_handler.set_current_operation(Constants.CONFIGURE_PATCHING) self.current_auto_os_patch_state = self.package_manager.get_current_auto_os_patch_state() @@ -87,9 +97,9 @@ def __try_set_patch_mode(self): if self.execution_config.patch_mode == Constants.PatchModes.AUTOMATIC_BY_PLATFORM and self.current_auto_os_patch_state == Constants.AutomaticOSPatchStates.UNKNOWN: # NOTE: only sending details in error objects for customer visibility on why patch state is unknown, overall configurepatching status will remain successful - self.configure_patching_exception_error = "Could not disable one or more automatic OS update services. Please check if they are configured correctly" + self.configure_patching_exception_error = "Could not disable one or more automatic OS update services. Please check if they are configured correctly." - self.composite_logger.log_debug("Completed processing patch mode configuration.") + self.composite_logger.log_debug("[CPP] Completed processing patch mode configuration.") except Exception as error: self.composite_logger.log_error("Error while processing patch mode configuration. [Error={0}]".format(repr(error))) self.configure_patching_exception_error = error @@ -98,6 +108,7 @@ def __try_set_patch_mode(self): def __try_set_auto_assessment_mode(self): """ Sets the preferred auto-assessment mode for the VM """ try: + self.composite_logger.log_verbose("[CPP] Processing assessment mode configuration...") self.status_handler.set_current_operation(Constants.CONFIGURE_PATCHING_AUTO_ASSESSMENT) self.composite_logger.log_debug("Systemd information: {0}".format(str(self.auto_assess_service_manager.get_version()))) # proactive support telemetry @@ -105,21 +116,25 @@ def __try_set_auto_assessment_mode(self): self.composite_logger.log_debug("No assessment mode config was present. No configuration changes will occur.") elif self.execution_config.assessment_mode == Constants.AssessmentModes.AUTOMATIC_BY_PLATFORM: self.composite_logger.log_debug("Enabling platform-based automatic assessment.") + if not self.auto_assess_service_manager.systemd_exists(): raise Exception("Systemd is not available on this system, and platform-based auto-assessment cannot be configured.") + self.auto_assess_service_manager.create_and_set_service_idem() self.auto_assess_timer_manager.create_and_set_timer_idem() + self.current_auto_assessment_state = Constants.AutoAssessmentStates.ENABLED elif self.execution_config.assessment_mode == Constants.AssessmentModes.IMAGE_DEFAULT: self.composite_logger.log_debug("Disabling platform-based automatic assessment.") self.auto_assess_timer_manager.remove_timer() self.auto_assess_service_manager.remove_service() + # self.__erase_auto_assess_config_if_any(Constants.AUTO_ASSESSMENT_SERVICE_NAME, self.auto_assess_service_manager, self.auto_assess_timer_manager) self.current_auto_assessment_state = Constants.AutoAssessmentStates.DISABLED else: raise Exception("Unknown assessment mode specified. [AssessmentMode={0}]".format(self.execution_config.assessment_mode)) self.__report_consolidated_configure_patch_status() - self.composite_logger.log_debug("Completed processing automatic assessment mode configuration.") + self.composite_logger.log_debug("[CPP] Completed processing automatic assessment mode configuration.") except Exception as error: # deliberately not setting self.configure_patching_exception_error here as it does not feed into the parent object. Not a bug, if you're thinking about it. self.composite_logger.log_error("Error while processing automatic assessment mode configuration. [Error={0}]".format(repr(error))) @@ -130,9 +145,25 @@ def __try_set_auto_assessment_mode(self): self.composite_logger.log_debug("Restoring status handler operation to {0}.".format(Constants.CONFIGURE_PATCHING)) self.status_handler.set_current_operation(Constants.CONFIGURE_PATCHING) + def __erase_auto_assess_config_if_any(self, service_name, service_manager, timer_manager): + # type: (str, ServiceManager, TimerManager) -> None + """ Cleans up the legacy auto-assess service """ + try: + if service_manager is not None: + self.composite_logger.log_debug("[CPP] Cleaning up the {0} service.".format(service_name)) + service_manager.remove_service() + + if timer_manager is not None: + self.composite_logger.log_debug("[CPP] Cleaning up the {0} timer.".format(service_name)) + timer_manager.remove_timer() + except Exception as error: + self.composite_logger.log_warning("[CPP] Retriable error while cleaning up auto-assess config. [Error={0}]".format(repr(error))) + self.configure_patching_successful &= False + def __report_consolidated_configure_patch_status(self, status=Constants.STATUS_TRANSITIONING, error=Constants.DEFAULT_UNSPECIFIED_VALUE): - """ Reports """ - self.composite_logger.log_debug("Reporting consolidated current configure patch status. [OSPatchState={0}][AssessmentState={1}]".format(self.current_auto_os_patch_state, self.current_auto_assessment_state)) + # type: (str, any) -> None + """ Reports the consolidated configure patching status """ + self.composite_logger.log_debug("[CPP] Reporting consolidated current configure patch status. [OSPatchState={0}][AssessmentState={1}]".format(self.current_auto_os_patch_state, self.current_auto_assessment_state)) # report error if specified if error != Constants.DEFAULT_UNSPECIFIED_VALUE: diff --git a/src/core/src/core_logic/ServiceManager.py b/src/core/src/core_logic/ServiceManager.py index 00f87e15..9c8dde78 100644 --- a/src/core/src/core_logic/ServiceManager.py +++ b/src/core/src/core_logic/ServiceManager.py @@ -36,6 +36,7 @@ def __init__(self, env_layer, execution_config, composite_logger, telemetry_writ # region - Service Creation / Removal def remove_service(self): + """ Remove the service if it exists """ service_path = self.__systemd_service_unit_path.format(self.service_name) if os.path.exists(service_path): self.stop_service() diff --git a/src/core/tests/Test_ConfigurePatchingProcessor.py b/src/core/tests/Test_ConfigurePatchingProcessor.py index 04242e0d..2b23cbab 100644 --- a/src/core/tests/Test_ConfigurePatchingProcessor.py +++ b/src/core/tests/Test_ConfigurePatchingProcessor.py @@ -18,6 +18,9 @@ import re import unittest import sys + +from library.ExtStatusAsserter import ExtStatusAsserter + # Conditional import for StringIO try: from StringIO import StringIO # Python 2 @@ -42,16 +45,17 @@ def tearDown(self): # self.runtime.stop() pass - #region Mocks + # region Mocks def mock_package_manager_get_current_auto_os_patch_state_returns_unknown(self): if self.mock_package_manager_get_current_auto_os_patch_state_returns_unknown_call_count == 0: self.mock_package_manager_get_current_auto_os_patch_state_returns_unknown_call_count = 1 return Constants.AutomaticOSPatchStates.DISABLED else: return Constants.AutomaticOSPatchStates.UNKNOWN + def mock_get_current_auto_os_patch_state(self): raise Exception("Mocked Exception") - #endregion Mocks + # endregion Mocks def test_operation_success_for_configure_patching_request_for_apt_with_default_updates_config(self): # create and adjust arguments @@ -76,24 +80,15 @@ def test_operation_success_for_configure_patching_request_for_apt_with_default_u # check telemetry events self.__check_telemetry_events(runtime) - # check status file for configure patching patch mode - 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"] - - # check status file for configure patching patch state (and including for 'Platform' initiated assessment data) + # assertions self.assertTrue(runtime.package_manager.image_default_patch_configuration_backup_exists()) - self.assertEqual(len(substatus_file_data), 2) - self.assertTrue(substatus_file_data[0]["name"] == Constants.PATCH_ASSESSMENT_SUMMARY) - message = json.loads(substatus_file_data[0]["formattedMessage"]["message"]) - self.assertTrue(message["startedBy"], Constants.PatchAssessmentSummaryStartedBy.PLATFORM) - self.assertTrue(substatus_file_data[1]["name"] == Constants.CONFIGURE_PATCHING_SUMMARY) - self.assertTrue(substatus_file_data[1]["status"].lower() == Constants.STATUS_SUCCESS.lower()) - message = json.loads(substatus_file_data[1]["formattedMessage"]["message"]) - self.assertEqual(message["automaticOSPatchState"], Constants.AutomaticOSPatchStates.DISABLED) - - # check status file for configure patching assessment state - message = json.loads(substatus_file_data[1]["formattedMessage"]["message"]) - self.assertEqual(message["autoAssessmentStatus"]["autoAssessmentState"], Constants.AutoAssessmentStates.DISABLED) + ext_status_asserter = ExtStatusAsserter(runtime.execution_config.status_file_path, runtime.env_layer) + ext_status_asserter.assert_status_file_substatuses(substatus_expectations={ + Constants.PATCH_ASSESSMENT_SUMMARY: Constants.STATUS_SUCCESS, + Constants.CONFIGURE_PATCHING_SUMMARY: Constants.STATUS_SUCCESS + }) + ext_status_asserter.assert_configure_patching_patch_mode_state(Constants.AutomaticOSPatchStates.DISABLED) + ext_status_asserter.assert_configure_patching_auto_assessment_state(Constants.AutomaticOSPatchStates.DISABLED) # stop test runtime runtime.stop() @@ -112,13 +107,11 @@ def test_operation_success_for_configure_patching_request_for_apt_without_defaul self.__check_telemetry_events(runtime) # 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), 2) - self.assertTrue(substatus_file_data[0]["name"] == Constants.PATCH_ASSESSMENT_SUMMARY) # assessment is now part of the CP flow - self.assertTrue(substatus_file_data[0]["status"].lower() == Constants.STATUS_SUCCESS.lower()) - self.assertTrue(substatus_file_data[1]["name"] == Constants.CONFIGURE_PATCHING_SUMMARY) - self.assertTrue(substatus_file_data[1]["status"].lower() == Constants.STATUS_SUCCESS.lower()) + ext_status_asserter = ExtStatusAsserter(runtime.execution_config.status_file_path, runtime.env_layer) + ext_status_asserter.assert_status_file_substatuses(substatus_expectations={ + Constants.PATCH_ASSESSMENT_SUMMARY: Constants.STATUS_SUCCESS, + Constants.CONFIGURE_PATCHING_SUMMARY: Constants.STATUS_SUCCESS + }) runtime.stop() def test_operation_success_for_installation_request_with_configure_patching(self): @@ -138,9 +131,7 @@ def test_operation_success_for_installation_request_with_configure_patching(self # check telemetry events self.__check_telemetry_events(runtime) - # 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"] + # assert self.assertTrue(runtime.package_manager.image_default_patch_configuration_backup_exists()) image_default_patch_configuration_backup = json.loads(runtime.env_layer.file_system.read_with_retry(runtime.package_manager.image_default_patch_configuration_backup_path)) self.assertTrue(image_default_patch_configuration_backup is not None) @@ -150,26 +141,14 @@ def test_operation_success_for_installation_request_with_configure_patching(self self.assertTrue(os_patch_configuration_settings is not None) self.assertTrue('APT::Periodic::Update-Package-Lists "0"' in os_patch_configuration_settings) self.assertTrue('APT::Periodic::Unattended-Upgrade "0"' in os_patch_configuration_settings) - self.assertTrue(substatus_file_data[3]["name"] == Constants.CONFIGURE_PATCHING_SUMMARY) - self.assertTrue(substatus_file_data[3]["status"].lower() == Constants.STATUS_SUCCESS.lower()) - self.assertEqual(len(substatus_file_data), 4) - self.assertTrue(substatus_file_data[0]["name"] == Constants.PATCH_ASSESSMENT_SUMMARY) - self.assertTrue(substatus_file_data[0]["status"].lower() == Constants.STATUS_SUCCESS.lower()) - self.assertTrue(substatus_file_data[1]["name"] == Constants.PATCH_INSTALLATION_SUMMARY) - self.assertTrue(substatus_file_data[1]["status"].lower() == Constants.STATUS_SUCCESS.lower()) - self.assertEqual(json.loads(substatus_file_data[1]["formattedMessage"]["message"])["patches"][0]["name"], "python-samba") - self.assertTrue("Security" in str(json.loads(substatus_file_data[1]["formattedMessage"]["message"])["patches"][0]["classifications"])) - self.assertEqual(json.loads(substatus_file_data[1]["formattedMessage"]["message"])["patches"][1]["name"], "samba-common-bin") - self.assertTrue("Security" in str(json.loads(substatus_file_data[1]["formattedMessage"]["message"])["patches"][1]["classifications"])) - self.assertEqual(json.loads(substatus_file_data[1]["formattedMessage"]["message"])["patches"][2]["name"], "samba-libs") - self.assertTrue("python-samba_2:4.4.5+dfsg-2ubuntu5.4" in str(json.loads(substatus_file_data[1]["formattedMessage"]["message"])["patches"][0]["patchId"])) - self.assertTrue("Security" in str(json.loads(substatus_file_data[1]["formattedMessage"]["message"])["patches"][2]["classifications"])) - self.assertTrue(substatus_file_data[2]["name"] == Constants.PATCH_METADATA_FOR_HEALTHSTORE) - self.assertTrue(substatus_file_data[2]["status"].lower() == Constants.STATUS_SUCCESS.lower()) - substatus_file_data_patch_metadata_summary = json.loads(substatus_file_data[2]["formattedMessage"]["message"]) - self.assertEqual(substatus_file_data_patch_metadata_summary["patchVersion"], "pub_off_sku_2020.09.23") - self.assertTrue(substatus_file_data_patch_metadata_summary["shouldReportToHealthStore"]) + # check status file + ext_status_asserter = ExtStatusAsserter(runtime.execution_config.status_file_path, runtime.env_layer) + ext_status_asserter.assert_status_file_substatuses() + ext_status_asserter.assert_operation_summary_has_patch(Constants.PATCH_INSTALLATION_SUMMARY, "python-samba", Constants.PackageClassification.SECURITY, "python-samba_2:4.4.5+dfsg-2ubuntu5.4") + ext_status_asserter.assert_operation_summary_has_patch(Constants.PATCH_INSTALLATION_SUMMARY,"samba-common-bin", Constants.PackageClassification.SECURITY) + ext_status_asserter.assert_operation_summary_has_patch(Constants.PATCH_INSTALLATION_SUMMARY,"samba-libs", Constants.PackageClassification.SECURITY) + ext_status_asserter.assert_healthstore_status_info(patch_version="pub_off_sku_2020.09.23", should_report=True) runtime.stop() def test_operation_fail_for_configure_patching_telemetry_not_supported(self): @@ -182,18 +161,19 @@ def test_operation_fail_for_configure_patching_telemetry_not_supported(self): runtime.configure_patching_processor.start_configure_patching() # 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) + ext_status_asserter = ExtStatusAsserter(runtime.execution_config.status_file_path, runtime.env_layer) + ext_status_asserter.assert_status_file_substatuses(substatus_expectations={ + Constants.CONFIGURE_PATCHING_SUMMARY: Constants.STATUS_ERROR + }) + if runtime.vm_cloud_type == Constants.VMCloudType.AZURE: - self.assertTrue(substatus_file_data[0]["status"].lower() == Constants.STATUS_ERROR.lower()) - self.assertTrue(len(json.loads(substatus_file_data[0]["formattedMessage"]["message"])["errors"]["details"]), 1) - self.assertTrue(Constants.TELEMETRY_NOT_COMPATIBLE_ERROR_MSG in json.loads(substatus_file_data[0]["formattedMessage"]["message"])["errors"]["details"][0]["message"]) - self.assertTrue(Constants.TELEMETRY_NOT_COMPATIBLE_ERROR_MSG in json.loads(substatus_file_data[0]["formattedMessage"]["message"])["autoAssessmentStatus"]["errors"]["details"][0]["message"]) - self.assertTrue(Constants.STATUS_ERROR in json.loads(substatus_file_data[0]["formattedMessage"]["message"])["autoAssessmentStatus"]["autoAssessmentState"]) + ext_status_asserter.assert_status_file_substatus(Constants.CONFIGURE_PATCHING_SUMMARY, Constants.STATUS_ERROR) + ext_status_asserter.assert_operation_summary_has_error(Constants.CONFIGURE_PATCHING_SUMMARY, Constants.TELEMETRY_NOT_COMPATIBLE_ERROR_MSG) + ext_status_asserter.assert_operation_summary_has_error(Constants.CONFIGURE_PATCHING_SUMMARY, Constants.TELEMETRY_NOT_COMPATIBLE_ERROR_MSG, 'autoAssessmentStatus') + ext_status_asserter.assert_configure_patching_auto_assessment_state(Constants.STATUS_ERROR) else: - self.assertTrue(substatus_file_data[0]["status"].lower() == Constants.STATUS_SUCCESS.lower()) + # this code path never executed in the test as it sat (discovered in refactoring). Marking as TODO for Arc. + ext_status_asserter.assert_status_file_substatus(Constants.CONFIGURE_PATCHING_SUMMARY, Constants.STATUS_SUCCESS) runtime.stop() def test_patch_mode_set_failure_for_configure_patching(self): @@ -216,21 +196,18 @@ def test_patch_mode_set_failure_for_configure_patching(self): self.__check_telemetry_events(runtime) # 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), 2) - self.assertTrue(substatus_file_data[0]["name"] == Constants.PATCH_ASSESSMENT_SUMMARY) # assessment is now part of the CP flow - self.assertTrue(substatus_file_data[0]["status"].lower() == Constants.STATUS_SUCCESS.lower()) - self.assertTrue(substatus_file_data[1]["name"] == Constants.CONFIGURE_PATCHING_SUMMARY) - self.assertTrue(substatus_file_data[1]["status"].lower() == Constants.STATUS_ERROR.lower()) + ext_status_asserter = ExtStatusAsserter(runtime.execution_config.status_file_path, runtime.env_layer) + ext_status_asserter.assert_status_file_substatuses(substatus_expectations={ + Constants.PATCH_ASSESSMENT_SUMMARY: Constants.STATUS_SUCCESS, + Constants.CONFIGURE_PATCHING_SUMMARY: Constants.STATUS_ERROR + }) - #restore + # 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_with_assessment_mode_by_platform(self): - # create and adjust arguments argument_composer = ArgumentComposer() argument_composer.operation = Constants.CONFIGURE_PATCHING @@ -253,22 +230,14 @@ def test_configure_patching_with_assessment_mode_by_platform(self): # check telemetry events self.__check_telemetry_events(runtime) - # check status file for configure patching patch mode - 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"] - - # check status file for configure patching patch state - self.assertEqual(len(substatus_file_data), 2) - self.assertTrue(substatus_file_data[1]["name"] == Constants.CONFIGURE_PATCHING_SUMMARY) - self.assertTrue(substatus_file_data[1]["status"].lower() == Constants.STATUS_SUCCESS.lower()) - message = json.loads(substatus_file_data[1]["formattedMessage"]["message"]) - self.assertEqual(message["automaticOSPatchState"], Constants.AutomaticOSPatchStates.ENABLED) # no change is made on Auto OS updates for patch mode 'ImageDefault' - self.assertTrue(substatus_file_data[0]["name"] == Constants.PATCH_ASSESSMENT_SUMMARY) - self.assertTrue(substatus_file_data[0]["status"].lower() == Constants.STATUS_SUCCESS.lower()) - - # check status file for configure patching assessment state - message = json.loads(substatus_file_data[1]["formattedMessage"]["message"]) - self.assertEqual(message["autoAssessmentStatus"]["autoAssessmentState"], Constants.AutoAssessmentStates.ENABLED) # auto assessment is enabled + # assertions + ext_status_asserter = ExtStatusAsserter(runtime.execution_config.status_file_path, runtime.env_layer) + ext_status_asserter.assert_status_file_substatuses(substatus_expectations={ + Constants.PATCH_ASSESSMENT_SUMMARY: Constants.STATUS_SUCCESS, + Constants.CONFIGURE_PATCHING_SUMMARY: Constants.STATUS_SUCCESS + }) + ext_status_asserter.assert_configure_patching_patch_mode_state(Constants.AutomaticOSPatchStates.ENABLED) # no change is made on Auto OS updates for patch mode 'ImageDefault' + ext_status_asserter.assert_configure_patching_auto_assessment_state(Constants.AutomaticOSPatchStates.ENABLED) # auto assessment is enabled # stop test runtime runtime.stop() @@ -297,23 +266,15 @@ def test_configure_patching_with_patch_mode_and_assessment_mode_by_platform(self # check telemetry events self.__check_telemetry_events(runtime) - # check status file for configure patching patch mode - 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"] - - # check status file for configure patching patch state + # assertions self.assertTrue(runtime.package_manager.image_default_patch_configuration_backup_exists()) - self.assertEqual(len(substatus_file_data), 2) - self.assertTrue(substatus_file_data[1]["name"] == Constants.CONFIGURE_PATCHING_SUMMARY) - self.assertTrue(substatus_file_data[1]["status"].lower() == Constants.STATUS_SUCCESS.lower()) - message = json.loads(substatus_file_data[1]["formattedMessage"]["message"]) - self.assertEqual(message["automaticOSPatchState"], Constants.AutomaticOSPatchStates.DISABLED) # auto OS updates are disabled on patch mode 'AutomaticByPlatform' - self.assertTrue(substatus_file_data[0]["name"] == Constants.PATCH_ASSESSMENT_SUMMARY) - self.assertTrue(substatus_file_data[0]["status"].lower() == Constants.STATUS_SUCCESS.lower()) - - # check status file for configure patching assessment state - message = json.loads(substatus_file_data[1]["formattedMessage"]["message"]) - self.assertEqual(message["autoAssessmentStatus"]["autoAssessmentState"], Constants.AutoAssessmentStates.ENABLED) # auto assessment is enabled + ext_status_asserter = ExtStatusAsserter(runtime.execution_config.status_file_path, runtime.env_layer) + ext_status_asserter.assert_status_file_substatuses(substatus_expectations={ + Constants.PATCH_ASSESSMENT_SUMMARY: Constants.STATUS_SUCCESS, + Constants.CONFIGURE_PATCHING_SUMMARY: Constants.STATUS_SUCCESS + }) + ext_status_asserter.assert_configure_patching_patch_mode_state(Constants.AutomaticOSPatchStates.DISABLED) # auto OS updates are disabled on patch mode 'AutomaticByPlatform' + ext_status_asserter.assert_configure_patching_auto_assessment_state(Constants.AutomaticOSPatchStates.ENABLED) # auto assessment is enabled # stop test runtime runtime.stop() @@ -338,7 +299,7 @@ def test_configure_patching_raise_exception_auto_os_patch_state(self): runtime.configure_patching_processor.start_configure_patching() - # restore sdt.out ouptput + # restore sdt.out output sys.stdout = original_stdout # assert @@ -346,11 +307,10 @@ def test_configure_patching_raise_exception_auto_os_patch_state(self): 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()) + ext_status_asserter = ExtStatusAsserter(runtime.execution_config.status_file_path, runtime.env_layer) + ext_status_asserter.assert_status_file_substatuses(substatus_expectations={ + Constants.CONFIGURE_PATCHING_SUMMARY: Constants.STATUS_TRANSITIONING + }) # restore runtime.package_manager.get_current_auto_os_patch_state = backup_package_manager_get_current_auto_os_patch_state @@ -365,13 +325,10 @@ def test_configure_patching_raise_exception_auto_assessment_systemd(self): runtime = RuntimeCompositor(argument_composer.get_composed_arguments(), True, Constants.APT) runtime.set_legacy_test_type('HappyPath') - # mock swap + # mock swap service manager 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() diff --git a/src/core/tests/library/ExtStatusAsserter.py b/src/core/tests/library/ExtStatusAsserter.py new file mode 100644 index 00000000..ab60c670 --- /dev/null +++ b/src/core/tests/library/ExtStatusAsserter.py @@ -0,0 +1,179 @@ +# Copyright 2025 Microsoft Corporation +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# +# Requires Python 2.7+ +import json + +from core.src.bootstrap.Constants import Constants + + +class ExtStatusAsserter(object): + def __init__(self, status_file_path, env_layer): + self.__status_file_path = status_file_path + self.__env_layer = env_layer + + self.__substatus_file_data = self.__read_status_file(self.__status_file_path) + self.__substatus_high_level_summary = None + self.__load_substatus_high_level_summary(self.__substatus_file_data) + + # region Data structure helpers + @staticmethod + def __get_high_level_summary_template(): + # type: () -> dict + """ Internal template for in-memory representation of substatus elements """ + return { + Constants.CONFIGURE_PATCHING_SUMMARY: {"index": -1, "status": None}, + Constants.PATCH_ASSESSMENT_SUMMARY: {"index": -1, "status": None}, + Constants.PATCH_INSTALLATION_SUMMARY: {"index": -1, "status": None}, + Constants.PATCH_METADATA_FOR_HEALTHSTORE: {"index": -1, "status": None}, + } + + def __get_substatus_index_with_assert(self, operation): + # type: (str) -> int + """ Get the index of the substatus """ + if operation not in self.__substatus_high_level_summary: + raise KeyError("Unknown operation: {0}".format(operation)) + + substatus_index = self.__substatus_high_level_summary[operation]["index"] + if substatus_index == -1: + raise AssertionError("Substatus index not found for operation: {0}".format(operation)) + + return substatus_index + + @staticmethod + def get_default_substatus_expectations(): + # type: () -> dict + """ Get the default substatus expectations """ + return { + Constants.CONFIGURE_PATCHING_SUMMARY: Constants.STATUS_SUCCESS, + Constants.PATCH_ASSESSMENT_SUMMARY: Constants.STATUS_SUCCESS, + Constants.PATCH_INSTALLATION_SUMMARY: Constants.STATUS_SUCCESS, + Constants.PATCH_METADATA_FOR_HEALTHSTORE: Constants.STATUS_SUCCESS, + } + # endregion + + # region Data loaders + def __read_status_file(self, status_file_path): + # type: (str) -> dict + with self.__env_layer.file_system.open(status_file_path, 'r') as file_handle: + return json.load(file_handle)[0]["status"]["substatus"] + + def __load_substatus_high_level_summary(self, substatus_file_data): + # type: (dict) -> None + """ Makes one-time inferences about the structure of the status file """ + self.__substatus_high_level_summary = self.__get_high_level_summary_template() + + for index, substatus in enumerate(substatus_file_data): + summary_name = substatus["name"] + if summary_name in self.__substatus_high_level_summary: + self.configure_patching_substatus_index = index + self.__substatus_high_level_summary[summary_name]["index"] = index + self.__substatus_high_level_summary[summary_name]["status"] = substatus["status"] + else: + raise KeyError("Unknown substatus: {0}".format(substatus["name"])) + # endregion + + # region Data Navigators + def __get_substatus_message_as_dict(self, operation): + # type: (str) -> dict + """ Get the substatus message as a dictionary """ + substatus_index = self.__get_substatus_index_with_assert(operation) + return json.loads(self.__substatus_file_data[substatus_index]["formattedMessage"]["message"]) + # endregion + + # region Public Assertion methods + def assert_status_file_substatus(self, operation, expected_status): + # type: (str, str) -> None + """ Check if the status file has a specific substatus """ + substatus_index = self.__get_substatus_index_with_assert(operation) + + actual_status = self.__substatus_file_data[substatus_index]["status"].lower() + if actual_status != expected_status.lower(): + raise AssertionError("Substatus expectations do not match for {0}. Expected: {1}, Actual: {2}".format(operation, expected_status, actual_status)) + + def assert_status_file_substatuses(self, substatus_expectations=None): + # type: (dict) -> None + """ Batch check the status file for substatus expectations """ + if substatus_expectations is None: + substatus_expectations = self.get_default_substatus_expectations() + + for key, value in substatus_expectations.items(): + self.assert_status_file_substatus(key, value) + + def assert_operation_summary_has_patch(self, operation, patch_name, classification=None, patch_id=None): + # type: (str, str, str, str) -> bool + """ Check if the defined operation summary has a specific patch """ + substatus_message = self.__get_substatus_message_as_dict(operation) + summary_patches = substatus_message["patches"] + + for patch in summary_patches: + if patch["name"] == patch_name: + if classification and classification not in patch["classifications"]: + raise AssertionError("Classification '{0}' does not match expected value '{1}' for patch '{2}'.".format(classification, str(patch["classifications"]), patch_name)) + if patch_id and patch_id not in str(patch["patchId"]): + raise AssertionError("Patch ID '{0}' does not match expected value '{1}' for patch '{2}'.".format(patch_id, str(patch["patch_id"]), patch_name)) + return True + + raise AssertionError("Patch '{0}' not found in '{1}' summary.".format(patch_name, operation)) + + def assert_operation_summary_has_error(self, operation, error_message, sub_level_for_configure_patching_only=None): + # type: (str, str, str) -> bool + """ Check if the defined operation summary has a specific error """ + substatus_message = self.__get_substatus_message_as_dict(operation) + + if sub_level_for_configure_patching_only not in [None, "autoAssessmentStatus", "patchModeStatus"]: + raise ValueError("sub_level_for_configure_patching_only must be None, 'autoAssessmentStatus', or 'patchModeStatus'.") + + if operation == Constants.CONFIGURE_PATCHING_SUMMARY and sub_level_for_configure_patching_only: + error_detail_list = substatus_message[sub_level_for_configure_patching_only]["errors"]["details"] + else: + error_detail_list = substatus_message["errors"]["details"] + + for error in error_detail_list: + if error_message in error["message"]: + return True + raise AssertionError("Error '{0}' not found in '{1}' summary.".format(error_message, operation)) + + def assert_operation_summary_has_started_by(self, operation, started_by): + # type: (str, str) -> None + """ Check if the defined operation summary has a specific started by """ + substatus_message = self.__get_substatus_message_as_dict(operation) + if substatus_message["startedBy"] != started_by: + raise AssertionError("Started by '{0}' does not match expected value '{1}' for operation '{2}.".format(substatus_message["startedBy"], started_by, operation)) + + def assert_configure_patching_patch_mode_state(self, expected_state): + # type: (str) -> None + """ Check if the patch mode state is as expected """ + substatus_message = self.__get_substatus_message_as_dict(Constants.CONFIGURE_PATCHING_SUMMARY) + if substatus_message["automaticOSPatchState"] != expected_state: + raise AssertionError("Patch mode state '{0}' does not match expected value '{1}'.".format(substatus_message["automaticOSPatchState"], expected_state)) + + def assert_configure_patching_auto_assessment_state(self, expected_state): + # type: (str) -> None + """ Check if the auto-assessment state is as expected """ + substatus_message = self.__get_substatus_message_as_dict(Constants.CONFIGURE_PATCHING_SUMMARY) + if substatus_message["autoAssessmentStatus"]["autoAssessmentState"] != expected_state: + raise AssertionError("Auto-assessment state '{0}' does not match expected value '{1}'.".format(substatus_message["autoAssessmentStatus"]["state"], expected_state)) + + def assert_healthstore_status_info(self, patch_version, should_report=True): + # type: (str, bool) -> None + """Check if the healthstore patch version is as expected""" + healthstore_summary = self.__get_substatus_message_as_dict(Constants.PATCH_METADATA_FOR_HEALTHSTORE) + + if should_report and healthstore_summary["shouldReportToHealthStore"] != True: + raise AssertionError("Healthstore summary should report to healthstore.") + + if patch_version != healthstore_summary["patchVersion"]: + raise AssertionError("Healthstore summary patch version '{0}' does not match expected value {1}.".format(str(healthstore_summary["patchVersion"]), patch_version)) + # endregion From 8a2fa7f324676b92f0a9a6372ab7c902edebf64f Mon Sep 17 00:00:00 2001 From: Koshy John Date: Mon, 15 Jun 2026 13:53:15 -0700 Subject: [PATCH 2/6] Typo fix --- src/core/tests/library/ExtStatusAsserter.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/core/tests/library/ExtStatusAsserter.py b/src/core/tests/library/ExtStatusAsserter.py index ab60c670..614516b3 100644 --- a/src/core/tests/library/ExtStatusAsserter.py +++ b/src/core/tests/library/ExtStatusAsserter.py @@ -164,7 +164,7 @@ def assert_configure_patching_auto_assessment_state(self, expected_state): """ Check if the auto-assessment state is as expected """ substatus_message = self.__get_substatus_message_as_dict(Constants.CONFIGURE_PATCHING_SUMMARY) if substatus_message["autoAssessmentStatus"]["autoAssessmentState"] != expected_state: - raise AssertionError("Auto-assessment state '{0}' does not match expected value '{1}'.".format(substatus_message["autoAssessmentStatus"]["state"], expected_state)) + raise AssertionError("Auto-assessment state '{0}' does not match expected value '{1}'.".format(substatus_message["autoAssessmentStatus"]["autoAssessmentState"], expected_state)) def assert_healthstore_status_info(self, patch_version, should_report=True): # type: (str, bool) -> None From a7458f2569244c26fe45b7486910555cf7419a3f Mon Sep 17 00:00:00 2001 From: Koshy John Date: Mon, 15 Jun 2026 14:02:53 -0700 Subject: [PATCH 3/6] Applying copilot review suggestions --- .../src/core_logic/ConfigurePatchingProcessor.py | 16 ---------------- .../tests/Test_ConfigurePatchingProcessor.py | 2 +- src/core/tests/library/ExtStatusAsserter.py | 2 +- 3 files changed, 2 insertions(+), 18 deletions(-) diff --git a/src/core/src/core_logic/ConfigurePatchingProcessor.py b/src/core/src/core_logic/ConfigurePatchingProcessor.py index 496be34c..3a737b13 100644 --- a/src/core/src/core_logic/ConfigurePatchingProcessor.py +++ b/src/core/src/core_logic/ConfigurePatchingProcessor.py @@ -128,7 +128,6 @@ def __try_set_auto_assessment_mode(self): self.composite_logger.log_debug("Disabling platform-based automatic assessment.") self.auto_assess_timer_manager.remove_timer() self.auto_assess_service_manager.remove_service() - # self.__erase_auto_assess_config_if_any(Constants.AUTO_ASSESSMENT_SERVICE_NAME, self.auto_assess_service_manager, self.auto_assess_timer_manager) self.current_auto_assessment_state = Constants.AutoAssessmentStates.DISABLED else: raise Exception("Unknown assessment mode specified. [AssessmentMode={0}]".format(self.execution_config.assessment_mode)) @@ -145,21 +144,6 @@ def __try_set_auto_assessment_mode(self): self.composite_logger.log_debug("Restoring status handler operation to {0}.".format(Constants.CONFIGURE_PATCHING)) self.status_handler.set_current_operation(Constants.CONFIGURE_PATCHING) - def __erase_auto_assess_config_if_any(self, service_name, service_manager, timer_manager): - # type: (str, ServiceManager, TimerManager) -> None - """ Cleans up the legacy auto-assess service """ - try: - if service_manager is not None: - self.composite_logger.log_debug("[CPP] Cleaning up the {0} service.".format(service_name)) - service_manager.remove_service() - - if timer_manager is not None: - self.composite_logger.log_debug("[CPP] Cleaning up the {0} timer.".format(service_name)) - timer_manager.remove_timer() - except Exception as error: - self.composite_logger.log_warning("[CPP] Retriable error while cleaning up auto-assess config. [Error={0}]".format(repr(error))) - self.configure_patching_successful &= False - def __report_consolidated_configure_patch_status(self, status=Constants.STATUS_TRANSITIONING, error=Constants.DEFAULT_UNSPECIFIED_VALUE): # type: (str, any) -> None """ Reports the consolidated configure patching status """ diff --git a/src/core/tests/Test_ConfigurePatchingProcessor.py b/src/core/tests/Test_ConfigurePatchingProcessor.py index 2b23cbab..a089e5f6 100644 --- a/src/core/tests/Test_ConfigurePatchingProcessor.py +++ b/src/core/tests/Test_ConfigurePatchingProcessor.py @@ -19,7 +19,7 @@ import unittest import sys -from library.ExtStatusAsserter import ExtStatusAsserter +from core.tests.library.ExtStatusAsserter import ExtStatusAsserter # Conditional import for StringIO try: diff --git a/src/core/tests/library/ExtStatusAsserter.py b/src/core/tests/library/ExtStatusAsserter.py index 614516b3..43a70b87 100644 --- a/src/core/tests/library/ExtStatusAsserter.py +++ b/src/core/tests/library/ExtStatusAsserter.py @@ -122,7 +122,7 @@ def assert_operation_summary_has_patch(self, operation, patch_name, classificati if classification and classification not in patch["classifications"]: raise AssertionError("Classification '{0}' does not match expected value '{1}' for patch '{2}'.".format(classification, str(patch["classifications"]), patch_name)) if patch_id and patch_id not in str(patch["patchId"]): - raise AssertionError("Patch ID '{0}' does not match expected value '{1}' for patch '{2}'.".format(patch_id, str(patch["patch_id"]), patch_name)) + raise AssertionError("Patch ID '{0}' does not match expected value '{1}' for patch '{2}'.".format(patch_id, str(patch["patchId"]), patch_name)) return True raise AssertionError("Patch '{0}' not found in '{1}' summary.".format(patch_name, operation)) From ffc07817ccd00f83520600a4cf8303c5c67cfe02 Mon Sep 17 00:00:00 2001 From: Koshy John Date: Mon, 15 Jun 2026 14:04:55 -0700 Subject: [PATCH 4/6] Addressing coverage --- src/core/tests/Test_ConfigurePatchingProcessor.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/core/tests/Test_ConfigurePatchingProcessor.py b/src/core/tests/Test_ConfigurePatchingProcessor.py index a089e5f6..99077f56 100644 --- a/src/core/tests/Test_ConfigurePatchingProcessor.py +++ b/src/core/tests/Test_ConfigurePatchingProcessor.py @@ -171,9 +171,9 @@ def test_operation_fail_for_configure_patching_telemetry_not_supported(self): ext_status_asserter.assert_operation_summary_has_error(Constants.CONFIGURE_PATCHING_SUMMARY, Constants.TELEMETRY_NOT_COMPATIBLE_ERROR_MSG) ext_status_asserter.assert_operation_summary_has_error(Constants.CONFIGURE_PATCHING_SUMMARY, Constants.TELEMETRY_NOT_COMPATIBLE_ERROR_MSG, 'autoAssessmentStatus') ext_status_asserter.assert_configure_patching_auto_assessment_state(Constants.STATUS_ERROR) - else: + #else: # this code path never executed in the test as it sat (discovered in refactoring). Marking as TODO for Arc. - ext_status_asserter.assert_status_file_substatus(Constants.CONFIGURE_PATCHING_SUMMARY, Constants.STATUS_SUCCESS) + # ext_status_asserter.assert_status_file_substatus(Constants.CONFIGURE_PATCHING_SUMMARY, Constants.STATUS_SUCCESS) runtime.stop() def test_patch_mode_set_failure_for_configure_patching(self): From 5005591e176b5b0b273a90c8eb66ad87c36d2121 Mon Sep 17 00:00:00 2001 From: Koshy John Date: Tue, 16 Jun 2026 14:42:46 -0700 Subject: [PATCH 5/6] Helper class test coverage --- src/core/tests/Test_ExtStatusAsserter.py | 216 +++++++++++++++++++++++ 1 file changed, 216 insertions(+) create mode 100644 src/core/tests/Test_ExtStatusAsserter.py diff --git a/src/core/tests/Test_ExtStatusAsserter.py b/src/core/tests/Test_ExtStatusAsserter.py new file mode 100644 index 00000000..92dc2f40 --- /dev/null +++ b/src/core/tests/Test_ExtStatusAsserter.py @@ -0,0 +1,216 @@ +# Copyright 2026 Microsoft Corporation +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# +# Requires Python 2.7+ + +import json +import os +import shutil +import tempfile +import unittest + +from core.src.bootstrap.Constants import Constants +from core.tests.library.ExtStatusAsserter import ExtStatusAsserter + + +class _TestFileSystem(object): + @staticmethod + def open(path, mode): + return open(path, mode) + + +class _TestEnvLayer(object): + def __init__(self): + self.file_system = _TestFileSystem() + +# IMPORTANT: THIS CLASS ONLY VALIDATES TEST CODE THAT NEEDS TO RELIABLY RAISE PRODUCT FAILURES IF THEY HAPPEN. +# NONE OF THE TESTS HERE ARE INTENDED TO BE DIRECTLY PRODUCT-FACING. +class TestExtStatusAsserter(unittest.TestCase): + def setUp(self): + self._temp_dir = tempfile.mkdtemp() + self._env_layer = _TestEnvLayer() + + def tearDown(self): + shutil.rmtree(self._temp_dir, ignore_errors=True) + + def _write_status_file(self, substatuses): + status_file_path = os.path.join(self._temp_dir, "status.json") + payload = [{"status": {"substatus": substatuses}}] + with open(status_file_path, "w") as file_handle: + file_handle.write(json.dumps(payload)) + return status_file_path + + @staticmethod + def _build_substatus(name, status, message_dict): + return { + "name": name, + "status": status, + "formattedMessage": {"message": json.dumps(message_dict)} + } + + def _build_default_substatuses(self): + configure_message = { + "patches": [{"name": "pkg1", "classifications": ["Security"], "patchId": "id-123"}], + "errors": {"details": [{"message": "configure root error"}]}, + "patchModeStatus": {"errors": {"details": [{"message": "patch mode problem"}]}}, + "autoAssessmentStatus": { + "autoAssessmentState": Constants.AutoAssessmentStates.ENABLED, + "errors": {"details": [{"message": "auto assessment problem"}]} + }, + "startedBy": "Platform", + "automaticOSPatchState": Constants.AutomaticOSPatchStates.DISABLED + } + + assessment_message = { + "patches": [{"name": "pkgA", "classifications": ["Critical"], "patchId": "id-A"}], + "errors": {"details": [{"message": "assessment error"}]}, + "startedBy": "User" + } + + installation_message = { + "patches": [{"name": "pkgI", "classifications": ["Security"], "patchId": "id-I"}], + "errors": {"details": [{"message": "installation error"}]}, + "startedBy": "User" + } + + healthstore_message = { + "patchVersion": "v1", + "shouldReportToHealthStore": True + } + + return [ + self._build_substatus(Constants.PATCH_ASSESSMENT_SUMMARY, Constants.STATUS_SUCCESS, assessment_message), + self._build_substatus(Constants.PATCH_INSTALLATION_SUMMARY, Constants.STATUS_SUCCESS, installation_message), + self._build_substatus(Constants.CONFIGURE_PATCHING_SUMMARY, Constants.STATUS_SUCCESS, configure_message), + self._build_substatus(Constants.PATCH_METADATA_FOR_HEALTHSTORE, Constants.STATUS_SUCCESS, healthstore_message) + ] + + def _create_asserter(self, substatuses): + status_file_path = self._write_status_file(substatuses) + return ExtStatusAsserter(status_file_path, self._env_layer) + + def test_constructor_raises_for_unknown_substatus(self): + substatuses = [ + self._build_substatus("UnknownSummary", Constants.STATUS_SUCCESS, {"errors": {"details": []}}) + ] + status_file_path = self._write_status_file(substatuses) + + with self.assertRaises(KeyError): + ExtStatusAsserter(status_file_path, self._env_layer) + + def test_get_default_substatus_expectations_contains_expected_defaults(self): + defaults = ExtStatusAsserter.get_default_substatus_expectations() + + self.assertEqual(defaults[Constants.CONFIGURE_PATCHING_SUMMARY], Constants.STATUS_SUCCESS) + self.assertEqual(defaults[Constants.PATCH_ASSESSMENT_SUMMARY], Constants.STATUS_SUCCESS) + self.assertEqual(defaults[Constants.PATCH_INSTALLATION_SUMMARY], Constants.STATUS_SUCCESS) + self.assertEqual(defaults[Constants.PATCH_METADATA_FOR_HEALTHSTORE], Constants.STATUS_SUCCESS) + + def test_assert_status_file_substatuses_uses_defaults(self): + asserter = self._create_asserter(self._build_default_substatuses()) + asserter.assert_status_file_substatuses() + + def test_assert_status_file_substatus_raises_for_unknown_operation(self): + asserter = self._create_asserter(self._build_default_substatuses()) + + with self.assertRaises(KeyError): + asserter.assert_status_file_substatus("UnknownOperation", Constants.STATUS_SUCCESS) + + def test_assert_status_file_substatus_raises_when_summary_not_present(self): + substatuses = [ + self._build_substatus(Constants.PATCH_ASSESSMENT_SUMMARY, Constants.STATUS_SUCCESS, {"errors": {"details": []}}) + ] + asserter = self._create_asserter(substatuses) + + with self.assertRaises(AssertionError): + asserter.assert_status_file_substatus(Constants.PATCH_INSTALLATION_SUMMARY, Constants.STATUS_SUCCESS) + + def test_assert_status_file_substatus_raises_for_status_mismatch(self): + asserter = self._create_asserter(self._build_default_substatuses()) + + with self.assertRaises(AssertionError): + asserter.assert_status_file_substatus(Constants.CONFIGURE_PATCHING_SUMMARY, Constants.STATUS_ERROR) + + def test_assert_operation_summary_has_patch_validates_classification_and_patch_id(self): + asserter = self._create_asserter(self._build_default_substatuses()) + + self.assertTrue(asserter.assert_operation_summary_has_patch(Constants.CONFIGURE_PATCHING_SUMMARY, "pkg1", "Security", "id-123")) + + def test_assert_operation_summary_has_patch_raises_when_patch_or_fields_mismatch(self): + asserter = self._create_asserter(self._build_default_substatuses()) + + with self.assertRaises(AssertionError): + asserter.assert_operation_summary_has_patch(Constants.CONFIGURE_PATCHING_SUMMARY, "pkg1", "Critical") + + with self.assertRaises(AssertionError): + asserter.assert_operation_summary_has_patch(Constants.CONFIGURE_PATCHING_SUMMARY, "pkg1", patch_id="missing") + + with self.assertRaises(AssertionError): + asserter.assert_operation_summary_has_patch(Constants.CONFIGURE_PATCHING_SUMMARY, "not-found") + + def test_assert_operation_summary_has_error_validates_sub_level(self): + asserter = self._create_asserter(self._build_default_substatuses()) + + with self.assertRaises(ValueError): + asserter.assert_operation_summary_has_error(Constants.CONFIGURE_PATCHING_SUMMARY, "error", "invalid-level") + + def test_assert_operation_summary_has_error_for_configure_sub_levels(self): + asserter = self._create_asserter(self._build_default_substatuses()) + + self.assertTrue(asserter.assert_operation_summary_has_error(Constants.CONFIGURE_PATCHING_SUMMARY, "auto assessment", "autoAssessmentStatus")) + self.assertTrue(asserter.assert_operation_summary_has_error(Constants.CONFIGURE_PATCHING_SUMMARY, "patch mode", "patchModeStatus")) + + def test_assert_operation_summary_has_error_raises_when_error_missing(self): + asserter = self._create_asserter(self._build_default_substatuses()) + + with self.assertRaises(AssertionError): + asserter.assert_operation_summary_has_error(Constants.PATCH_ASSESSMENT_SUMMARY, "missing error text") + + def test_assert_operation_summary_has_started_by_raises_for_mismatch(self): + asserter = self._create_asserter(self._build_default_substatuses()) + + with self.assertRaises(AssertionError): + asserter.assert_operation_summary_has_started_by(Constants.CONFIGURE_PATCHING_SUMMARY, "User") + + def test_assert_configure_patching_states_raise_for_mismatch(self): + asserter = self._create_asserter(self._build_default_substatuses()) + + with self.assertRaises(AssertionError): + asserter.assert_configure_patching_patch_mode_state(Constants.AutomaticOSPatchStates.ENABLED) + + with self.assertRaises(AssertionError): + asserter.assert_configure_patching_auto_assessment_state(Constants.AutoAssessmentStates.DISABLED) + + def test_assert_healthstore_status_info_validates_reporting_and_patch_version(self): + asserter = self._create_asserter(self._build_default_substatuses()) + asserter.assert_healthstore_status_info("v1", should_report=True) + + with self.assertRaises(AssertionError): + asserter.assert_healthstore_status_info("v2", should_report=True) + + substatuses = self._build_default_substatuses() + substatuses[3] = self._build_substatus( + Constants.PATCH_METADATA_FOR_HEALTHSTORE, + Constants.STATUS_SUCCESS, + {"patchVersion": "v1", "shouldReportToHealthStore": False} + ) + asserter = self._create_asserter(substatuses) + + asserter.assert_healthstore_status_info("v1", should_report=False) + with self.assertRaises(AssertionError): + asserter.assert_healthstore_status_info("v1", should_report=True) + + +if __name__ == '__main__': + unittest.main() \ No newline at end of file From 96b17ca48242d32e72bfa517e57db4e083a85835 Mon Sep 17 00:00:00 2001 From: Koshy John Date: Wed, 17 Jun 2026 16:10:18 -0700 Subject: [PATCH 6/6] Addressing copilot comments --- src/core/tests/Test_ConfigurePatchingProcessor.py | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) diff --git a/src/core/tests/Test_ConfigurePatchingProcessor.py b/src/core/tests/Test_ConfigurePatchingProcessor.py index 99077f56..4bb98eea 100644 --- a/src/core/tests/Test_ConfigurePatchingProcessor.py +++ b/src/core/tests/Test_ConfigurePatchingProcessor.py @@ -88,7 +88,7 @@ def test_operation_success_for_configure_patching_request_for_apt_with_default_u Constants.CONFIGURE_PATCHING_SUMMARY: Constants.STATUS_SUCCESS }) ext_status_asserter.assert_configure_patching_patch_mode_state(Constants.AutomaticOSPatchStates.DISABLED) - ext_status_asserter.assert_configure_patching_auto_assessment_state(Constants.AutomaticOSPatchStates.DISABLED) + ext_status_asserter.assert_configure_patching_auto_assessment_state(Constants.AutoAssessmentStates.DISABLED) # stop test runtime runtime.stop() @@ -170,7 +170,7 @@ def test_operation_fail_for_configure_patching_telemetry_not_supported(self): ext_status_asserter.assert_status_file_substatus(Constants.CONFIGURE_PATCHING_SUMMARY, Constants.STATUS_ERROR) ext_status_asserter.assert_operation_summary_has_error(Constants.CONFIGURE_PATCHING_SUMMARY, Constants.TELEMETRY_NOT_COMPATIBLE_ERROR_MSG) ext_status_asserter.assert_operation_summary_has_error(Constants.CONFIGURE_PATCHING_SUMMARY, Constants.TELEMETRY_NOT_COMPATIBLE_ERROR_MSG, 'autoAssessmentStatus') - ext_status_asserter.assert_configure_patching_auto_assessment_state(Constants.STATUS_ERROR) + ext_status_asserter.assert_configure_patching_auto_assessment_state(Constants.AutoAssessmentStates.ERROR) #else: # this code path never executed in the test as it sat (discovered in refactoring). Marking as TODO for Arc. # ext_status_asserter.assert_status_file_substatus(Constants.CONFIGURE_PATCHING_SUMMARY, Constants.STATUS_SUCCESS) @@ -328,7 +328,13 @@ def test_configure_patching_raise_exception_auto_assessment_systemd(self): # mock swap service manager 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()) + runtime.configure_patching_processor.start_configure_patching() + runtime.configure_patching_processor.set_configure_patching_final_overall_status() + ext_status_asserter = ExtStatusAsserter(runtime.execution_config.status_file_path, runtime.env_layer) + ext_status_asserter.assert_operation_summary_has_error(operation=Constants.CONFIGURE_PATCHING_SUMMARY, + error_message="Systemd is not available on this system, and platform-based auto-assessment cannot be configured", + sub_level_for_configure_patching_only="autoAssessmentStatus") + runtime.configure_patching_processor.auto_assess_service_manager.systemd_exists = back_up_auto_assess_service_manager runtime.stop()