Minor comments

src/core/src/core_logic/ExecutionConfig.py

@@ -107,7 +107,7 @@ def __get_max_patch_publish_date(self, health_store_id):
         if health_store_id is not None and health_store_id != "":
             split = health_store_id.split("_")
             if len(split) == 4 and len(split[3]) == 10:
-                max_patch_publish_date = "{0}T000000Z".format(str(split[3].replace(".", "")))
+                max_patch_publish_date = "{0}T000000Z".format(split[3].replace(".", ""))
 
         self.composite_logger.log_debug("[EC] Getting max patch publish date. [MaxPatchPublishDate={0}][HealthStoreId={1}]".format(str(max_patch_publish_date), str(health_store_id)))
         return max_patch_publish_date

src/core/src/package_managers/AptitudePackageManager.py

@@ -54,7 +54,7 @@ def __init__(self, env_layer, execution_config, composite_logger, telemetry_writ
         # Install update
         # --only-upgrade: upgrade only single package (only if it is installed)
         self.single_package_upgrade_cmd = '''sudo DEBIAN_FRONTEND=noninteractive LANG=en_US.UTF8 ''' + optional_accept_eula_in_cmd + ''' apt-get -y --only-upgrade true install '''
-        self.install_security_updates_azgps_coordinated_cmd = '''sudo DEBIAN_FRONTEND=noninteractive LANG=en_US.UTF8 ''' + optional_accept_eula_in_cmd + ''' apt-get -y --only-upgrade true upgrade '''
+        self.install_security_updates_azgps_coordinated_cmd = '''sudo DEBIAN_FRONTEND=noninteractive LANG=en_US.UTF8 ''' + optional_accept_eula_in_cmd + ''' apt-get -y --only-upgrade true upgrade <SOURCES> '''
 
         # Package manager exit code(s)
         self.apt_exitcode_ok = 0
@@ -85,6 +85,9 @@ def __get_custom_sources_to_spec(self, max_patch_published_date=str(), base_clas
         # type: (str, str) -> str
         """ Prepares the custom sources list for use in a command. Idempotent. """
         try:
+            if max_patch_published_date != str() and len(max_patch_published_date) != 16:
+                raise Exception("[APM] Invalid max patch published date received. [Value={0}]".format(str(max_patch_published_date)))
+
             formula = "F-[{0}]-[{1}]".format(max_patch_published_date, base_classification)
             if self.cached_customer_source_list_formula == formula:
                 return self.custom_sources_list
@@ -120,7 +123,7 @@ def __get_custom_sources_to_spec(self, max_patch_published_date=str(), base_clas
         return self.custom_sources_list
 
     def refresh_repo(self, sources=str()):
-        self.composite_logger.log("[APM] Refreshing local repo... [Sources={0}]".format(sources if sources != "" else "Default"))
+        self.composite_logger.log("[APM] Refreshing local repo... [Sources={0}]".format(sources if sources != str() else "Default"))
         self.invoke_package_manager(self.__generate_command(self.cmd_repo_refresh_template, sources))
 
     @staticmethod
@@ -341,7 +344,7 @@ def install_updates_fail_safe(self, excluded_packages):
         return
 
     def install_security_updates_azgps_coordinated(self):
-        command = self.__generate_command(self.install_security_updates_azgps_coordinated_cmd, self.__get_custom_sources_to_spec(self.__get_custom_sources_to_spec(self.max_patch_publish_date, base_classification="security")))
+        command = self.__generate_command(self.install_security_updates_azgps_coordinated_cmd, self.__get_custom_sources_to_spec(self.max_patch_publish_date, base_classification="security"))
         out, code = self.invoke_package_manager_advanced(command, raise_on_exception=False)
         return code, out
     # endregion

src/core/tests/Test_CoreMain.py

@@ -197,7 +197,7 @@ def test_operation_success_for_autopatching_request_with_security_classification
         runtime.stop()
 
     def test_health_store_id_reporting(self):
-        # test with empty string for maintenence run id
+        # test with empty string for healthstoreid
         argument_composer = ArgumentComposer()
         health_store_id = "pub_offer_sku_wrong_123"
         argument_composer.health_store_id = health_store_id
@@ -226,7 +226,7 @@ def test_health_store_id_reporting(self):
         self.assertTrue(substatus_file_data[3]["status"].lower() == Constants.STATUS_SUCCESS.lower())
         runtime.stop()
 
-        # test with a random string for maintenance run id
+        # test with healthstoreid
         argument_composer = ArgumentComposer()
         health_store_id = "publ_off_sku_2024.04.01"
         argument_composer.health_store_id = health_store_id

src/core/tests/Test_PatchInstaller.py

@@ -249,6 +249,9 @@ def test_patch_installer_for_azgps_coordinated(self):
         self.assertFalse(runtime.patch_installer.start_installation())                 # failure is in unrelated patch installation batch processing
         self.assertEqual(runtime.execution_config.max_patch_publish_date, "20240401T000000Z")
         self.assertEqual(runtime.package_manager.max_patch_publish_date, "")    # reason: not enough time to use
+
+        runtime.package_manager.max_patch_publish_date = "Wrong"
+        runtime.package_manager.get_security_updates()      # exercises an exception path on bad data without throwing an exception (graceful degradation to security)
         runtime.stop()
 
         argument_composer.maximum_duration = "PT235M"
@@ -257,7 +260,7 @@ def test_patch_installer_for_azgps_coordinated(self):
         runtime.package_manager.install_security_updates_azgps_coordinated = lambda: (1, "Failed")
         self.assertFalse(runtime.patch_installer.start_installation())
         self.assertEqual(runtime.execution_config.max_patch_publish_date, "20240401T000000Z")
-        self.assertEqual(runtime.package_manager.max_patch_publish_date, "")    # reason: not enough time to use
+        self.assertEqual(runtime.package_manager.max_patch_publish_date, "")    # reason: the strict SDP is forced to fail with the lambda above
         runtime.stop()
 
         runtime = RuntimeCompositor(argument_composer.get_composed_arguments(), True, Constants.YUM)