[AzL3Tdnf] Refactor: Addressing PR feedback #3.1

Author: rane-rajasi

src/core/src/package_managers/AzL3TdnfPackageManager.py

@@ -29,11 +29,8 @@ class AzL3TdnfPackageManager(TdnfPackageManager):
     def __init__(self, env_layer, execution_config, composite_logger, telemetry_writer, status_handler):
         super(AzL3TdnfPackageManager, self).__init__(env_layer, execution_config, composite_logger, telemetry_writer, status_handler)
 
-        # Support to get updates and their dependencies
-        self.tdnf_check = 'sudo tdnf -q list updates <SNAPSHOTTIME>'
-
         # Install update
-        self.install_security_updates_azgps_coordinated_cmd = 'sudo tdnf -y upgrade --skip-broken <SNAPSHOTTIME>'
+        self.install_security_updates_azgps_coordinated_cmd = 'sudo tdnf -y upgrade --skip-broken'
 
         # Strict SDP specializations
         self.TDNF_MINIMUM_VERSION_FOR_STRICT_SDP = "3.5.8-3.azl3"  # minimum version of tdnf required to support Strict SDP in Azure Linux
@@ -51,14 +48,11 @@ def __init__(self, env_layer, execution_config, composite_logger, telemetry_writ
             execution_config.included_classifications_list = [Constants.PackageClassification.CRITICAL, Constants.PackageClassification.SECURITY, Constants.PackageClassification.OTHER]
 
     # region Strict SDP using SnapshotTime
-    @staticmethod
-    def __generate_command_with_snapshotposixtime_if_specified(command_template, snapshot_posix_time=str()):
-        # type: (str, str) -> str
-        if snapshot_posix_time == str():
-            return command_template.replace('<SNAPSHOTTIME>', str())
+    def _Base__add_additional_parameters_as_required_to_cmd(self, cmd):
+        if self.max_patch_publish_date == str():
+            return cmd
         else:
-            return command_template.replace('<SNAPSHOTTIME>', ('--snapshottime={0}'.format(str(snapshot_posix_time))))
-    # endregion
+            return cmd + ' --snapshottime={0}'.format(str(self.max_patch_publish_date))
 
     # region Get Available Updates
     # region Classification-based (incl. All) update check
@@ -69,28 +63,15 @@ def get_all_updates(self, cached=False):
             self.composite_logger.log_debug("[AzL3TDNF] Get all updates : [Cached={0}][PackagesCount={1}]]".format(str(cached), len(self.all_updates_cached)))
             return self.all_updates_cached, self.all_update_versions_cached  # allows for high performance reuse in areas of the code explicitly aware of the cache
 
-        out = self.invoke_package_manager(self.__generate_command_with_snapshotposixtime_if_specified(self.tdnf_check, self.max_patch_publish_date))
+        out = self.invoke_package_manager(self._Base__add_additional_parameters_as_required_to_cmd(self.tdnf_check))
         self.all_updates_cached, self.all_update_versions_cached = self.extract_packages_and_versions(out)
         self.composite_logger.log_debug("[AzL3TDNF] Get all updates : [Cached={0}][PackagesCount={1}]]".format(str(False), len(self.all_updates_cached)))
         return self.all_updates_cached, self.all_update_versions_cached
 
-    def get_security_updates(self):
-        """Get missing security updates. NOTE: Classification based categorization of patches is not available in Azure Linux as of now"""
-        self.composite_logger.log_verbose("[AzL3TDNF] Discovering all packages as 'security' packages, since TDNF does not support package classification...")
-        security_packages, security_package_versions = self.get_all_updates(cached=False)
-        self.composite_logger.log_debug("[AzL3TDNF] Discovered 'security' packages. [Count={0}]".format(len(security_packages)))
-        return security_packages, security_package_versions
-
-    def get_other_updates(self):
-        """Get missing other updates."""
-        self.composite_logger.log_verbose("[AzL3TDNF] Discovering 'other' packages...")
-        return [], []
-
     def set_max_patch_publish_date(self, max_patch_publish_date=str()):
         """Set the max patch publish date in POSIX time for strict SDP"""
-        self.composite_logger.log_debug("[AzL3TDNF] Setting max patch publish date. [MaxPatchPublishDate={0}]".format(str(max_patch_publish_date)))
         self.max_patch_publish_date = str(self.env_layer.datetime.datetime_string_to_posix_time(max_patch_publish_date, '%Y%m%dT%H%M%SZ')) if max_patch_publish_date != str() else max_patch_publish_date
-        self.composite_logger.log_debug("[AzL3TDNF] Set max patch publish date. [MaxPatchPublishDatePosixTime={0}]".format(str(self.max_patch_publish_date)))
+        self.composite_logger.log_debug("[AzL3TDNF] Set max patch publish date in posix time for Strict SDP. [MaxPatchPublishDate={0}][MaxPatchPublishDatePosixTime={1}]".format(str(max_patch_publish_date), str(self.max_patch_publish_date)))
     # endregion
 
     # endregion
@@ -101,7 +82,7 @@ def install_updates_fail_safe(self, excluded_packages):
 
     def install_security_updates_azgps_coordinated(self):
         """Install security updates in Azure Linux following strict SDP"""
-        command = self.__generate_command_with_snapshotposixtime_if_specified(self.install_security_updates_azgps_coordinated_cmd, self.max_patch_publish_date)
+        command = self._Base__add_additional_parameters_as_required_to_cmd(self.install_security_updates_azgps_coordinated_cmd)
         out, code = self.invoke_package_manager_advanced(command, raise_on_exception=False)
         return code, out
 

src/core/src/package_managers/PackageManager.py

@@ -349,9 +349,8 @@ def install_security_updates_azgps_coordinated(self):
 
     @abstractmethod
     def try_meet_azgps_coordinated_requirements(self):
-        # type: () -> bool
         """ Returns true if the package manager meets the requirements for azgps coordinated security updates """
-        return False
+        pass
     # endregion
 
     # region Package Information

src/core/src/package_managers/TdnfPackageManager.py

@@ -19,7 +19,7 @@
 import os
 import re
 
-from abc import abstractmethod
+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
@@ -35,6 +35,7 @@ def __init__(self, env_layer, execution_config, composite_logger, telemetry_writ
         self.cmd_repo_refresh = "sudo tdnf -q list updates"
 
         # Support to get updates and their dependencies
+        self.tdnf_check = 'sudo tdnf -q list updates'
         self.single_package_check_versions = 'sudo tdnf list available <PACKAGE-NAME> '
         self.single_package_check_installed = 'sudo tdnf list installed <PACKAGE-NAME> '
         self.single_package_upgrade_simulation_cmd = 'sudo tdnf install --assumeno --skip-broken '
@@ -77,6 +78,8 @@ def __init__(self, env_layer, execution_config, composite_logger, telemetry_writ
 
         self.package_install_expected_avg_time_in_seconds = 90  # Setting a default value of 90 seconds as the avg time to install a package using tdnf, might be changed later if needed.
 
+    __metaclass__ = ABCMeta  # For Python 3.0+, it changes to class Abstract(metaclass=ABCMeta)
+
     def refresh_repo(self):
         self.composite_logger.log("[TDNF] Refreshing local repo...")
         self.invoke_package_manager(self.cmd_clean_cache)
@@ -102,23 +105,36 @@ def invoke_package_manager_advanced(self, command, raise_on_exception=True):
         return out, code
 
     # region Classification-based (incl. All) update check
-    @abstractmethod
     def get_all_updates(self, cached=False):
-        """Same behavior as get_available_updates, but higher performance with no filters"""
-        pass
-        return [], []  # only here to suppress a static syntax validation problem
+        """Get all missing updates"""
+        self.composite_logger.log_verbose("[TDNF] Discovering all packages...")
+        if cached and not len(self.all_updates_cached) == 0:
+            self.composite_logger.log_debug("[TDNF] Get all updates : [Cached={0}][PackagesCount={1}]]".format(str(cached), len(self.all_updates_cached)))
+            return self.all_updates_cached, self.all_update_versions_cached  # allows for high performance reuse in areas of the code explicitly aware of the cache
+
+        out = self.invoke_package_manager(self.__add_additional_parameters_as_required_to_cmd(self.tdnf_check))
+        self.all_updates_cached, self.all_update_versions_cached = self.extract_packages_and_versions(out)
+        self.composite_logger.log_debug("[TDNF] Get all updates : [Cached={0}][PackagesCount={1}]]".format(str(False), len(self.all_updates_cached)))
+        return self.all_updates_cached, self.all_update_versions_cached
 
-    @abstractmethod
     def get_security_updates(self):
-        pass
+        """Get missing security updates. NOTE: Classification based categorization of patches is not available in Azure Linux as of now"""
+        self.composite_logger.log_verbose("[TDNF] Discovering all packages as 'security' packages, since TDNF does not support package classification...")
+        security_packages, security_package_versions = self.get_all_updates(cached=False)
+        self.composite_logger.log_debug("[TDNF] Discovered 'security' packages. [Count={0}]".format(len(security_packages)))
+        return security_packages, security_package_versions
 
-    @abstractmethod
     def get_other_updates(self):
-        pass
+        """Get missing other updates."""
+        self.composite_logger.log_verbose("[TDNF] Discovering 'other' packages...")
+        return [], []
 
-    @abstractmethod
     def set_max_patch_publish_date(self, max_patch_publish_date=str()):
-        pass
+        self.composite_logger.log_debug("[TDNF] Setting max patch publish date. [Date={0}]".format(str()))
+        self.max_patch_publish_date = str()
+
+    def __add_additional_parameters_as_required_to_cmd(self, cmd):
+        return cmd
     # endregion
 
     # region Output Parser(s)
@@ -195,7 +211,6 @@ def install_updates_fail_safe(self, excluded_packages):
     def install_security_updates_azgps_coordinated(self):
         pass
 
-    @abstractmethod
     def try_meet_azgps_coordinated_requirements(self):
         # type: () -> bool
         """ Returns true if the package manager meets the requirements for azgps coordinated security updates """

src/core/tests/Test_AzL3TdnfPackageManager.py

@@ -44,10 +44,6 @@ def mock_linux_distribution_to_return_azure_linux(self):
     def mock_linux_distribution_to_return_azure_linux_2(self):
         return ['Common Base Linux Mariner', '2.0', '']
 
-    def mock_run_command_output_return_tdnf_3(self, cmd, no_output=False, chk_err=True):
-        """ Mock for run_command_output to return tdnf 3 """
-        return 0, "3.5.8-3\n"
-
     def mock_run_command_output_return_1(self, cmd, no_output=False, chk_err=True):
         """ Mock for run_command_output to return None """
         return 1, "No output available\n"
@@ -119,24 +115,6 @@ def test_set_max_patch_publish_date(self):
         # posix time computation throws an exception if the date is not in the correct format
         self.assertRaises(ValueError, package_manager.set_max_patch_publish_date, "2024-07-02T00:00:00Z")
 
-    def test_get_tdnf_version(self):
-        """Unit test for tdnf package manager get_tdnf_version method"""
-        package_manager = self.container.get('package_manager')
-        self.assertTrue(package_manager is not None)
-        self.backup_run_command_output = self.runtime.env_layer.run_command_output
-
-        test_input_output_table = [
-            [self.mock_run_command_output_return_tdnf_3, "3.5.8-3"],
-            [self.mock_run_command_output_return_1, None],
-        ]
-
-        for row in test_input_output_table:
-            self.runtime.env_layer.run_command_output = row[0]
-            version = package_manager.get_tdnf_version()
-            self.assertEqual(version, row[1])
-
-        self.runtime.env_layer.run_command_output = self.backup_run_command_output
-
     def test_is_mininum_tdnf_version_for_strict_sdp_installed(self):
         """Unit test for tdnf package manager is_minimum_tdnf_version method"""
         package_manager = self.container.get('package_manager')

src/core/tests/Test_TdnfPackageManager.py

@@ -28,7 +28,7 @@
 from core.tests.library.RuntimeCompositor import RuntimeCompositor
 
 
-class TestAzL3TdnfPackageManager(unittest.TestCase):
+class TestTdnfPackageManager(unittest.TestCase):
     def setUp(self):
         self.runtime = RuntimeCompositor(ArgumentComposer().get_composed_arguments(), True, Constants.TDNF)
         self.container = self.runtime.container
@@ -42,6 +42,14 @@ def mock_do_processes_require_restart_raise_exception(self):
 
     def mock_write_with_retry_raise_exception(self, file_path_or_handle, data, mode='a+'):
         raise Exception
+
+    def mock_run_command_output_return_tdnf_3(self, cmd, no_output=False, chk_err=True):
+        """ Mock for run_command_output to return tdnf 3 """
+        return 0, "3.5.8-3\n"
+
+    def mock_run_command_output_return_1(self, cmd, no_output=False, chk_err=True):
+        """ Mock for run_command_output to return None """
+        return 1, "No output available\n"
     # endregion
 
     # region Utility Functions
@@ -449,6 +457,23 @@ def test_revert_auto_os_update_to_system_default(self):
                 self.__assert_std_io(captured_output=captured_output, expected_output=testcase["stdio"]["expected_output"])
             self.__assert_reverted_automatic_patch_configuration_settings(package_manager, config_exists=bool(testcase["assertions"]["config_exists"]), config_value_expected=testcase["assertions"]["config_value_expected"])
 
+    def test_get_tdnf_version(self):
+        """Unit test for tdnf package manager get_tdnf_version method"""
+        package_manager = self.container.get('package_manager')
+        self.assertTrue(package_manager is not None)
+        self.backup_run_command_output = self.runtime.env_layer.run_command_output
+
+        test_input_output_table = [
+            [self.mock_run_command_output_return_tdnf_3, "3.5.8-3"],
+            [self.mock_run_command_output_return_1, None],
+        ]
+
+        for row in test_input_output_table:
+            self.runtime.env_layer.run_command_output = row[0]
+            version = package_manager.get_tdnf_version()
+            self.assertEqual(version, row[1])
+
+        self.runtime.env_layer.run_command_output = self.backup_run_command_output
 
     def test_package_manager_no_updates(self):
         """Unit test for tdnf package manager with no updates"""