Modifed code to just log Execstart and fix import

Author: yashnap

src/core/src/package_managers/AptitudePackageManager.py

@@ -183,7 +183,7 @@ def __read_one_line_style_list_format(self, file_path, max_patch_published_date,
 
         for line in sources_content_lines:
             if len(line.strip()) != 0 and not line.strip().startswith("#"):
-                if base_classification == Constants.PackageClassification.SECURITY and "security" not in line:
+                if base_classification == Constants.PackageClassification.SECURITY and "security" not in line and "fips-updates" not in line:
                     continue
                 std_source_list_content += "\n" + self.__apply_max_patch_publish_date(sources_content=line, max_patch_publish_date=max_patch_published_date)
 
@@ -207,7 +207,7 @@ def __read_deb882_style_format(self, file_path, max_patch_published_date, base_c
 
                 if len(line.strip()) == 0:  # stanza separating line
                     if stanza != str():
-                        if base_classification == str() or (base_classification == Constants.PackageClassification.SECURITY and "security" in stanza):
+                        if base_classification == str() or (base_classification == Constants.PackageClassification.SECURITY and ("security" in stanza or "fips-updates" in stanza)):
                             std_source_parts_content += self.__apply_max_patch_publish_date(sources_content=stanza, max_patch_publish_date=max_patch_published_date) + '\n'
                         stanza = str()
                         continue

src/core/src/package_managers/Dnf5PackageManager.py

@@ -17,10 +17,10 @@
 """Dnf5PackageManager for Azure Linux 4 or above and RHEL"""
 import json
 import re
-from abc import ABCMeta, abstractmethod
 
-from core.src.bootstrap.Constants import Constants
+from abc import ABCMeta, abstractmethod
 from core.src.core_logic.VersionComparator import VersionComparator
+from core.src.bootstrap.Constants import Constants
 from core.src.package_managers.PackageManager import PackageManager
 
 
@@ -41,14 +41,12 @@ def __init__(self, env_layer, execution_config, composite_logger, telemetry_writ
 
         # Install update
         self.single_package_upgrade_cmd = 'sudo dnf5 -y upgrade'
-
         # Support to check if reboot is required
         # dnf-utils not required (needs-restarting is built into dnf5)
         self.needs_restarting_with_flag = 'sudo LANG=en_US.UTF8 dnf5 needs-restarting -r'
 
         # DNF5 exit codes
         self.dnf_exitcode_ok = [0, 100]
-
         # DNF5 valid exit codes for simulation commands
         self.dnf5_simulation_valid_exit_codes = [0, 1]
         self.dnf5_dependency_failure_text = ["Skipping packages with broken dependencies", "Nothing to do."]
@@ -469,10 +467,6 @@ def __init_constants_for_dnf5_automatic(self):
         self.dnf5_automatic_config_pattern_match_text = ' = (no|yes)'
         self.dnf5_automatic_download_updates_identifier_text = "download_updates"
         self.dnf5_automatic_apply_updates_identifier_text = "apply_updates"
-        # ExecStart flag identifiers
-        self.dnf5_automatic_download_updates_flag = '--downloadupdates'
-        self.dnf5_automatic_apply_updates_flag = '--installupdates'
-
         self.dnf5_automatic_enable_on_reboot_identifier_text = "enable_on_reboot"
         self.dnf5_automatic_installation_state_identifier_text = "installation_state"
         self.dnf5_auto_os_update_service = "dnf5-automatic"
@@ -534,7 +528,13 @@ def __get_current_auto_os_updates_setting_on_machine(self):
                 return is_service_installed, enable_on_reboot_value, download_updates_value, apply_updates_value
 
             code, service_output = self.env_layer.run_command_output(self.dnf5_automatic_configuration_service, False, False)
-            self.composite_logger.log_debug("[DNF5] dnf5-automatic ExecStart details. [Command={0}][Code={1}][Output={2}]".format(self.dnf5_automatic_configuration_service, str(code), service_output))
+            exec_start_line = ""
+            #Only print ExecStart details
+            for line in service_output.splitlines():
+                if line.strip().startswith("ExecStart"):
+                    exec_start_line = line.strip()
+                    break
+            self.composite_logger.log_debug("[DNF5] dnf5-automatic ExecStart details. [Command={0}][Code={1}][Output={2}]".format(self.dnf5_automatic_configuration_service, str(code), exec_start_line))
 
             is_service_installed = True
             enable_on_reboot_value = self.is_service_set_to_enable_on_reboot(self.enable_on_reboot_check_cmd)
@@ -670,7 +670,6 @@ def backup_image_default_patch_configuration_if_not_exists(self):
                         self.installation_state_identifier_text: is_service_installed
                     }
                 }
-
                 image_default_patch_configuration_backup.update(backup_image_default_patch_configuration_json_to_add)
 
                 self.composite_logger.log_debug("[DNF5] Logging default system configuration settings for auto OS updates. ""[Settings={0}] [Log file path={1}]".format(str(image_default_patch_configuration_backup),self.image_default_patch_configuration_backup_path))
@@ -701,8 +700,7 @@ def update_os_patch_configuration_sub_setting(self, patch_configuration_sub_sett
             # note: adding space between the patch_configuration_sub_setting and value since, we will have to do that if we have to add a patch_configuration_sub_setting that did not exist before
             self.composite_logger.log_debug("[DNF5] Updating system configuration settings for auto OS updates. [Patch Configuration Sub Setting={0}] [Value={1}]".format(
                     str(patch_configuration_sub_setting), value))
-            os_patch_configuration_settings = self.env_layer.file_system.read_with_retry(
-                self.os_patch_configuration_settings_file_path)
+            os_patch_configuration_settings = self.env_layer.file_system.read_with_retry(self.os_patch_configuration_settings_file_path)
             patch_configuration_sub_setting_to_update = patch_configuration_sub_setting + ' = ' + value
             patch_configuration_sub_setting_found_in_file = False
             updated_patch_configuration_sub_setting = ""
@@ -720,9 +718,7 @@ def update_os_patch_configuration_sub_setting(self, patch_configuration_sub_sett
             if not patch_configuration_sub_setting_found_in_file:
                 updated_patch_configuration_sub_setting += patch_configuration_sub_setting_to_update + "\n"
 
-            self.env_layer.file_system.write_with_retry(self.os_patch_configuration_settings_file_path,
-                                                        '{0}'.format(updated_patch_configuration_sub_setting.lstrip()),
-                                                        mode='w+')
+            self.env_layer.file_system.write_with_retry(self.os_patch_configuration_settings_file_path,'{0}'.format(updated_patch_configuration_sub_setting.lstrip()),mode='w+')
         except Exception as error:
             error_msg = "[DNF5] Error occurred while updating system configuration settings for auto OS updates. [Patch Configuration={0}] [Error={1}]".format(
                 str(patch_configuration_sub_setting), repr(error))
@@ -789,8 +785,7 @@ def __get_image_default_patch_configuration_backup(self):
         # read existing backup since it also contains backup from other update services. We need to preserve any existing data within the backup file
         if self.image_default_patch_configuration_backup_exists():
             try:
-                image_default_patch_configuration_backup = json.loads(
-                    self.env_layer.file_system.read_with_retry(self.image_default_patch_configuration_backup_path))
+                image_default_patch_configuration_backup = json.loads(self.env_layer.file_system.read_with_retry(self.image_default_patch_configuration_backup_path))
             except Exception as error:
                 self.composite_logger.log_error("[DNF5] Unable to read backup for default patch state. Will attempt to re-write. [Exception={0}]".format(repr(error)))
         return image_default_patch_configuration_backup

src/core/tests/Test_AptitudePackageManagerCustomSources.py

@@ -99,8 +99,8 @@ def __lib_test_custom_sources_with(self, include_sources_list=False, include_sou
         # Checks swapping and caching: All -> Security -> All -> All
         for i in range(3):
             # All
-            expected_debstyle_entry_count = 2 if include_source_parts_debstyle else 0   # 2 entries in the debstyle mock
-            expected_sources_list_entry_count = (4 if include_sources_list else 0) + (5 if include_source_parts_list else 0)    # 4 in regular file, 3 in mock folder
+            expected_debstyle_entry_count = 3 if include_source_parts_debstyle else 0   # 3 entries in the debstyle mock(incl. fips-updates)
+            expected_sources_list_entry_count = (4 if include_sources_list else 0) + (6 if include_source_parts_list else 0)    # 4 in regular file, 6 in mock folder (incl. fips-updates)
             sources_dir, sources_list = package_manager._AptitudePackageManager__get_custom_sources_to_spec(include_max_patch_publish_date)
             self.__check_custom_sources(sources_dir, sources_list,
                                         sources_debstyle_expected=include_source_parts_debstyle,
@@ -115,8 +115,8 @@ def __lib_test_custom_sources_with(self, include_sources_list=False, include_sou
                 continue
 
             # Security
-            expected_debstyle_entry_count = 1 if include_source_parts_debstyle else 0   # 1 security entry in the debstyle mock
-            expected_sources_list_entry_count = (1 if include_sources_list else 0) + (1 if include_source_parts_list else 0)    # 1 security entry in regular file, 1 security entry in mock folder
+            expected_debstyle_entry_count = 2 if include_source_parts_debstyle else 0   # 2 security entries in the debstyle mock (security + fips-updates)
+            expected_sources_list_entry_count = (1 if include_sources_list else 0) + (2 if include_source_parts_list else 0)    # 1 security entry in regular file, 2 security entries in mock folder (security + fips-updates)
             sources_dir, sources_list = package_manager._AptitudePackageManager__get_custom_sources_to_spec(include_max_patch_publish_date, "Security")
             self.__check_custom_sources(sources_dir, sources_list,
                                         sources_debstyle_expected=include_source_parts_debstyle,
@@ -146,8 +146,8 @@ def __check_custom_sources(self, sources_dir, sources_list, sources_debstyle_exp
                 self.assertEqual(len(data), sources_debstyle_entry_count)
                 for entry in data:
                     if security_only:
-                        self.assertTrue("security" in entry)
-                    if max_patch_publish_date != str():
+                        self.assertTrue("security" in entry or "fips-updates" in entry)
+                    if max_patch_publish_date != str() and "fips-updates" not in entry:  # exception for unsupported repo:
                         self.assertTrue(max_patch_publish_date in entry)
         else:
             self.assertFalse(os.path.isdir(sources_dir))
@@ -159,8 +159,8 @@ def __check_custom_sources(self, sources_dir, sources_list, sources_debstyle_exp
                 self.assertEqual(len(data), sources_list_entry_count)
                 for entry in data:
                     if security_only:
-                        self.assertTrue("security" in entry)
-                    if max_patch_publish_date != str() and "ppa" not in entry:  # exception for unsupported repo
+                        self.assertTrue("security" in entry or "fips-updates" in entry)
+                    if max_patch_publish_date != str() and "ppa" not in entry and "fips-updates" not in entry:  # exception for unsupported repo
                         self.assertTrue(max_patch_publish_date in entry)
 
     # region - Mock sources preparation and clean up
@@ -212,7 +212,8 @@ def __get_sources_data_one_line_style_ext():
                 "deb http://ppa.launchpad.net/upubuntu-com/web/ubuntu focal main\n" + \
                 "deb http://azure.archive.ubuntu.com/ubuntu/ focal-security universe\n" + \
                 "deb http://in.archive.ubuntu.com/ubuntu/ focal multiverse\n" + \
-                "deb http://cn.archive.ubuntu.com/ubuntu/ focal main\n"
+                "deb http://cn.archive.ubuntu.com/ubuntu/ focal main\n" + \
+                "deb https://esm.ubuntu.com/fips-updates/ubuntu jammy-updates main\n"
 
     @staticmethod
     def __get_sources_data_debstyle():
@@ -229,7 +230,13 @@ def __get_sources_data_debstyle():
                 "URIs: http://azure.archive.ubuntu.com/ubuntu/ \n" + \
                 "Suites: noble-security \n" + \
                 "Components: main universe restricted multiverse \n" + \
-                "Signed-By: /usr/share/keyrings/ubuntu-archive-keyring.gpg \n"
+                "Signed-By: /usr/share/keyrings/ubuntu-archive-keyring.gpg \n" + \
+                "\n" + \
+                "Types: deb \n" + \
+                "URIs: https://esm.ubuntu.com/fips-updates/ubuntu \n" + \
+                "Suites: jammy-updates \n" + \
+                "Components: main \n" + \
+                "Signed-By: /usr/share/keyrings/ubuntu-pro-fips-updates.gpg \n"
     # endregion
 
 if __name__ == '__main__':

src/core/tests/Test_EnvLayer.py

@@ -17,12 +17,6 @@
 import platform
 import sys
 import unittest
-# Conditional import for StringIO
-try:
-    from StringIO import StringIO  # Python 2
-except ImportError:
-    from io import StringIO  # Python 3
-
 from core.src.bootstrap.EnvLayer import EnvLayer
 from core.src.bootstrap.Constants import Constants
 from core.src.external_dependencies import distro
@@ -171,16 +165,13 @@ def test_get_package_manager_azure_linux_4_and_rhel10_not_supported(self):
         ]
 
         for row in test_input_output_table:
-            captured_output = StringIO()
-            original_output = sys.stdout
-            sys.stdout = captured_output  # Redirect stdout to the StringIO object
-
             self.envlayer.platform.linux_distribution = row[0]
             distro.os_release_attr = row[1]
 
+            captured_output = io.StringIO()
             sys.stdout = captured_output
             result = self.envlayer.get_package_manager()
-            sys.stdout = original_output
+            sys.stdout = sys.__stdout__
             self.assertEqual(row[2], captured_output.getvalue())
             self.assertEqual(result, "")