Feature: Azure Linux 3.0: Enhanced Support

src/core/src/bootstrap/EnvLayer.py

@@ -534,6 +534,14 @@ def utc_to_standard_datetime(utc_datetime):
         def standard_datetime_to_utc(std_datetime):
             """ Converts datetime object to string of format '"%Y-%m-%dT%H:%M:%SZ"' """
             return std_datetime.strftime("%Y-%m-%dT%H:%M:%SZ")
+
+        @staticmethod
+        def datetime_string_to_posix_time(datetime_string, format_string):
+            """ Converts string of given format to posix datetime string. """
+            # eg: Input: datetime_string: 20241220T000000Z (str), format_string: '%Y%m%dT%H%M%SZ' -> Output: 1734681600 (str)
+            datetime_object = datetime.datetime.strptime(datetime_string, format_string)
+            posix_timestamp = str(int(time.mktime(datetime_object.timetuple())))
+            return posix_timestamp
 # endregion - DateTime emulator and extensions
 
 # region - Core Emulator support functions

src/core/src/package_managers/TdnfPackageManager.py

@@ -15,9 +15,11 @@
 # Requires Python 2.7+
 
 """TdnfPackageManager for Azure Linux"""
+import datetime
 import json
 import os
 import re
+import time
 
 from core.src.core_logic.VersionComparator import VersionComparator
 from core.src.package_managers.PackageManager import PackageManager
@@ -33,14 +35,17 @@ def __init__(self, env_layer, execution_config, composite_logger, telemetry_writ
         self.cmd_clean_cache = "sudo tdnf clean expire-cache"
         self.cmd_repo_refresh = "sudo tdnf -q list updates"
 
+        # fetch snapshottime from health_store_id
+        self.snapshot_posix_time = self.__get_posix_time(execution_config.max_patch_publish_date, env_layer)
+
         # 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 '
+        self.tdnf_check = self.__generate_command_with_snapshottime('sudo tdnf -q list updates <SNAPSHOTTIME>', self.snapshot_posix_time)
+        self.single_package_check_versions = self.__generate_command_with_snapshottime('sudo tdnf list available <PACKAGE-NAME> <SNAPSHOTTIME>', self.snapshot_posix_time)
+        self.single_package_check_installed = self.__generate_command_with_snapshottime('sudo tdnf list installed <PACKAGE-NAME> <SNAPSHOTTIME>', self.snapshot_posix_time)
+        self.single_package_upgrade_simulation_cmd = self.__generate_command_with_snapshottime('sudo tdnf install --assumeno --skip-broken <SNAPSHOTTIME>', self.snapshot_posix_time)
 
         # Install update
-        self.single_package_upgrade_cmd = 'sudo tdnf -y install --skip-broken '
+        self.single_package_upgrade_cmd = self.__generate_command_with_snapshottime('sudo tdnf -y install --skip-broken <SNAPSHOTTIME>', self.snapshot_posix_time)
 
         # Package manager exit code(s)
         self.tdnf_exitcode_ok = 0
@@ -89,6 +94,31 @@ def refresh_repo(self):
         self.invoke_package_manager(self.cmd_clean_cache)
         self.invoke_package_manager(self.cmd_repo_refresh)
 
+    # region Strict SDP using SnapshotTime
+    def __get_posix_time(self, datetime_to_convert, env_layer):
+        """Converts date str received to POSIX time string"""
+        posix_time = str()
+        datetime_to_convert_format = '%Y%m%dT%H%M%SZ'
+        self.composite_logger.log_debug("[TDNF] Getting POSIX time from given datetime. [DateTimeToConvert={0}][DateTimeStringFormat={1}]".format(str(datetime_to_convert), datetime_to_convert_format))
+        try:
+            if datetime_to_convert != str():
+                posix_time = env_layer.datetime.datetime_string_to_posix_time(datetime_to_convert, datetime_to_convert_format)
+        except Exception as error:
+            self.composite_logger.log_debug("[TDNF] Could not fetch POSIX time from given datetime. [DateTimeToConvert={0}][DateTimeStringFormat={1}][ComputedPosixTime={2}][Error={3}]".format(str(datetime_to_convert), datetime_to_convert_format, posix_time, repr(error)))
+
+        self.composite_logger.log_debug("[TDNF] Computed POSIX time from given datetime. [DateTimeToConvert={0}][DateTimeStringFormat={1}][ComputedPosixTime={2}]".format(str(datetime_to_convert), datetime_to_convert_format, posix_time))
+        return posix_time
+
+    @staticmethod
+    def __generate_command_with_snapshottime(command_template, snapshotposixtime=str()):
+        # type: (str, str) -> str
+        """ Prepares a standard command to use snapshottime."""
+        if snapshotposixtime == str():
+            return command_template.replace('<SNAPSHOTTIME>', str())
+        else:
+            return command_template.replace('<SNAPSHOTTIME>', ('--snapshottime={0}'.format(str(snapshotposixtime))))
+    # endregion
+
     # region Get Available Updates
     def invoke_package_manager_advanced(self, command, raise_on_exception=True):
         """Get missing updates using the command input"""
@@ -318,14 +348,15 @@ def is_arch_in_package_details(package_detail, package_arch_to_look_for):
 
     def get_dependent_list(self, packages):
         """Returns dependent List for the list of packages"""
+        cmd = self.single_package_upgrade_simulation_cmd
         package_names = ""
         for index, package in enumerate(packages):
             if index != 0:
                 package_names += ' '
             package_names += package
 
-        self.composite_logger.log_verbose("[TDNF] Resolving dependencies. [Command={0}]".format(str(self.single_package_upgrade_simulation_cmd + package_names)))
-        output = self.invoke_package_manager(self.single_package_upgrade_simulation_cmd + package_names)
+        self.composite_logger.log_verbose("[TDNF] Resolving dependencies. [Command={0}]".format(str(cmd + package_names)))
+        output = self.invoke_package_manager(cmd + package_names)
         dependencies = self.extract_dependencies(output, packages)
         self.composite_logger.log_verbose("[TDNF] Resolved dependencies. [Packages={0}][DependencyCount={1}]".format(str(packages), len(dependencies)))
         return dependencies

src/core/tests/Test_TdnfPackageManager.py

@@ -41,6 +41,55 @@ def mock_write_with_retry_raise_exception(self, file_path_or_handle, data, mode=
         raise Exception
     # endregion
 
+    def test_invalid_datetime_to_convert_to_posix_time(self):
+        # health_store_id is None
+        self.runtime.stop()
+        argument_composer = ArgumentComposer()
+        argument_composer.health_store_id = None
+        self.runtime = RuntimeCompositor(argument_composer.get_composed_arguments(), True, Constants.TDNF)
+        self.container = self.runtime.container
+        package_manager = self.container.get('package_manager')
+        self.assertTrue(package_manager.snapshot_posix_time is str())
+
+        # health_store_id is empty string
+        self.runtime.stop()
+        argument_composer = ArgumentComposer()
+        argument_composer.health_store_id = ""
+        self.runtime = RuntimeCompositor(argument_composer.get_composed_arguments(), True, Constants.TDNF)
+        self.container = self.runtime.container
+        package_manager = self.container.get('package_manager')
+        self.assertTrue(package_manager.snapshot_posix_time is str())
+
+        # health_store_id is random string
+        self.runtime.stop()
+        argument_composer = ArgumentComposer()
+        argument_composer.health_store_id = "pub_offer_sku_20.312.543"
+        self.runtime = RuntimeCompositor(argument_composer.get_composed_arguments(), True, Constants.TDNF)
+        self.container = self.runtime.container
+        package_manager = self.container.get('package_manager')
+        self.assertTrue(package_manager.snapshot_posix_time is str())
+
+    def test_valid_datetime_to_convert_to_posix_time(self):
+        # valid health_store_id
+        self.runtime.stop()
+        argument_composer = ArgumentComposer()
+        argument_composer.health_store_id = "pub_offer_sku_2024.04.01"
+        self.runtime = RuntimeCompositor(argument_composer.get_composed_arguments(), True, Constants.TDNF)
+        self.container = self.runtime.container
+        package_manager = self.container.get('package_manager')
+        self.assertTrue(package_manager.snapshot_posix_time is not None)
+        self.assertEqual("1711954800", package_manager.snapshot_posix_time)
+
+        # health_store_id is in unexpected format
+        self.runtime.stop()
+        argument_composer = ArgumentComposer()
+        argument_composer.health_store_id = "pub_offer_sk_u_2024.04.01"
+        self.runtime = RuntimeCompositor(argument_composer.get_composed_arguments(), True, Constants.TDNF)
+        self.container = self.runtime.container
+        package_manager = self.container.get('package_manager')
+        self.assertTrue(package_manager.snapshot_posix_time is not None)
+        self.assertEqual("1711954800", package_manager.snapshot_posix_time)
+
     def test_do_processes_require_restart(self):
         """Unit test for tdnf package manager"""
         # Restart required

src/core/tests/library/LegacyEnvLayerExtensions.py

@@ -14,6 +14,7 @@
 #
 # Requires Python 2.7+
 import os
+import re
 import sys
 from core.src.bootstrap.Constants import Constants
 
@@ -606,6 +607,7 @@ def run_command_output(self, cmd, no_output=False, chk_err=True):
                                  "Error(1032) : Operation aborted.\n"
                     elif cmd.find("list installed") > -1:
                         code = 0
+                        cmd = re.sub(r"--snapshottime=\d+", '', cmd)
                         package = cmd.replace('sudo tdnf list installed ', '')
                         whitelisted_versions = [
                             '3.0-16.azl3', '3.0-3.azl3', '2.5.4-1.azl3', '3.12.3-6.azl3', '2.11.5-1.azl3', '102-7.azl3', '6.6.78.1-1.azl3']  # any list of versions you want to work for *any* package