SG-40996 prevent tk component autoupdate

python/tank/descriptor/io_descriptor/appstore.py

@@ -13,12 +13,16 @@
 """
 
 import fnmatch
+import functools
 import http.client
 import json
 import os
+import sys
 import urllib.parse
 import urllib.request
+from typing import Dict, Iterator, List, Optional, Set, Tuple
 
+from tank.util.version import is_version_newer
 from tank_vendor import shotgun_api3
 from tank_vendor.shotgun_api3.lib import httplib2
 
@@ -37,6 +41,7 @@
     TankAppStoreConnectionError,
     TankAppStoreError,
     TankDescriptorError,
+    TankMissingManifestError,
 )
 from .downloadable import IODescriptorDownloadable
 
@@ -125,9 +130,7 @@ def __init__(self, descriptor_dict, sg_connection, bundle_type):
         :param bundle_type: Either Descriptor.APP, CORE, ENGINE or FRAMEWORK or CONFIG
         :return: Descriptor instance
         """
-        super().__init__(
-            descriptor_dict, sg_connection, bundle_type
-        )
+        super().__init__(descriptor_dict, sg_connection, bundle_type)
 
         self._validate_descriptor(
             descriptor_dict, required=["type", "name", "version"], optional=["label"]
@@ -538,6 +541,180 @@ def get_latest_cached_version(self, constraint_pattern=None):
         log.debug("Latest cached version resolved to %r" % desc)
         return desc
 
+    def _filter_versions_by_label(self, all_versions: Dict[str, str]) -> Iterator[str]:
+        """
+        Filter cached versions by label if one is specified.
+
+        :param all_versions: Dictionary of {version_string: path}
+        :returns: Iterator of version strings that match the label criteria
+        """
+        if not self._label:
+            yield from all_versions.keys()
+            return
+
+        log.debug(f"Filtering cached versions by label '{self._label}'...")
+        for version_str, path in all_versions.items():
+            metadata = self.__load_cached_app_store_metadata(path)
+            try:
+                tags = [x["name"] for x in metadata["sg_version_data"]["tags"]]
+                if self.__match_label(tags):
+                    yield version_str
+            except (KeyError, TypeError, AttributeError) as e:
+                log.debug(
+                    f"Could not determine label metadata for {path}. Ignoring. Details: {e}"
+                )
+
+    def _filter_versions_by_app_store(
+        self, version_numbers: Iterator[str], sg_matching_records: Optional[List[Dict]]
+    ) -> Iterator[str]:
+        """
+        Filter versions to only those available in the App Store.
+
+        :param version_numbers: Iterator of version strings
+        :param sg_matching_records: List of Shotgun version records from App Store
+        :returns: Iterator of version strings that exist in App Store, or original iterator if no records provided
+        """
+        if not sg_matching_records:
+            yield from version_numbers
+            return
+
+        app_store_versions: Set[str] = {rec["code"] for rec in sg_matching_records}
+        for v in version_numbers:
+            if v in app_store_versions:
+                yield v
+
+    def _sort_versions_descending(self, version_numbers: Iterator[str]) -> List[str]:
+        """
+        Sort version numbers in descending order (newest first).
+
+        :param version_numbers: Iterator of version strings
+        :returns: Sorted list with newest versions first
+        """
+        # Need to materialize the iterator for sorting
+        versions_list = list(version_numbers)
+
+        def version_compare(a: str, b: str) -> int:
+            """
+            Compare two version strings for sorting.
+            Returns: -1 if a < b, 0 if a == b, 1 if a > b
+            """
+            try:
+                if is_version_newer(a, b):
+                    return 1  # a is newer than b
+                elif is_version_newer(b, a):
+                    return -1  # b is newer than a
+                else:
+                    return 0  # equal
+            except Exception:
+                # If comparison fails, treat as equal
+                return 0
+
+        try:
+            # Use sorted with cmp_to_key to sort in descending order (newest first)
+            return sorted(
+                versions_list, key=functools.cmp_to_key(version_compare), reverse=True
+            )
+        except Exception:
+            # Fallback to simple reverse sort if version comparison fails
+            return sorted(versions_list, reverse=True)
+
+    def _is_version_python_compatible(
+        self, version_str: str
+    ) -> Optional["IODescriptorAppStore"]:
+        """
+        Check if a specific version is compatible with current Python.
+
+        :param version_str: Version string to check
+        :returns: IODescriptorAppStore instance if compatible, None otherwise
+        """
+        try:
+            # Create descriptor for this cached version
+            temp_descriptor_dict = {
+                "type": "app_store",
+                "name": self._name,
+                "version": version_str,
+            }
+            if self._label:
+                temp_descriptor_dict["label"] = self._label
+
+            temp_desc = IODescriptorAppStore(
+                temp_descriptor_dict, self._sg_connection, self._bundle_type
+            )
+            temp_desc.set_cache_roots(self._bundle_cache_root, self._fallback_roots)
+
+            # Check if this version exists locally
+            if not temp_desc.exists_local():
+                log.debug(
+                    f"Version {version_str} not found in cache (expected at {temp_desc.get_path()})"
+                )
+                return
+
+            # Check Python compatibility
+            manifest = temp_desc.get_manifest(constants.BUNDLE_METADATA_FILE)
+            if self._check_minimum_python_version(manifest):
+                log.debug(f"Found compatible cached version: {version_str}")
+                return temp_desc
+
+            # Not compatible - log and return
+            min_py_ver = manifest.get("minimum_python_version", "not specified")
+            log.debug(
+                f"Cached version {version_str} requires Python {min_py_ver} - not compatible"
+            )
+            return
+
+        except (TankDescriptorError, TankMissingManifestError, IOError, OSError) as e:
+            log.debug(
+                f"Could not check compatibility for cached version {version_str}: {e}"
+            )
+            return
+
+    def _find_compatible_cached_version(
+        self, sg_matching_records: Optional[List[Dict]]
+    ) -> Optional["IODescriptorAppStore"]:
+        """
+        Searches locally cached versions for the highest version compatible
+        with the current Python version.
+
+        :param sg_matching_records: List of Shotgun version records from App Store
+        :returns: IODescriptorAppStore instance or None if no compatible version found
+        """
+        log.debug(
+            "Searching for compatible cached version of %r in bundle cache..." % self
+        )
+
+        # Get all locally cached versions
+        all_versions = self._get_locally_cached_versions()
+        if not all_versions:
+            log.debug("No cached versions found in bundle cache")
+            return
+
+        log.debug("Found %d cached versions to check" % len(all_versions))
+
+        # Apply filters: label and App Store availability
+        version_numbers = self._filter_versions_by_label(all_versions)
+        version_numbers = self._filter_versions_by_app_store(
+            version_numbers, sg_matching_records
+        )
+
+        # Sort versions (newest first)
+        sorted_versions = self._sort_versions_descending(version_numbers)
+
+        if not sorted_versions:
+            log.debug("No cached versions match criteria")
+            return
+
+        log.debug(
+            f"Checking {len(sorted_versions)} cached versions for Python compatibility (newest first)"
+        )
+
+        # Find first compatible version
+        for version_str in sorted_versions:
+            descriptor = self._is_version_python_compatible(version_str)
+            if descriptor:
+                return descriptor
+
+        log.debug("No compatible cached version found")
+
     @LogManager.log_timing
     def get_latest_version(self, constraint_pattern=None):
         """
@@ -556,8 +733,7 @@ def get_latest_version(self, constraint_pattern=None):
         :returns: IODescriptorAppStore object
         """
         log.debug(
-            "Determining latest version for %r given constraint pattern %s"
-            % (self, constraint_pattern)
+            f"Determining latest version for {self} given constraint pattern {constraint_pattern}"
         )
 
         # connect to the app store
@@ -582,7 +758,7 @@ def get_latest_version(self, constraint_pattern=None):
 
             if sg_bundle_data is None:
                 raise TankDescriptorError(
-                    "App store does not contain an item named '%s'!" % self._name
+                    f"App store does not contain an item named '{self._name}'!"
                 )
 
             # now get all versions
@@ -613,8 +789,7 @@ def get_latest_version(self, constraint_pattern=None):
         )
 
         log.debug(
-            "Downloaded data for %d versions from Flow Production Tracking."
-            % len(sg_versions)
+            f"Downloaded data for {len(sg_versions)} versions from Flow Production Tracking."
         )
 
         # now filter out all labels that aren't matching
@@ -625,12 +800,12 @@ def get_latest_version(self, constraint_pattern=None):
                 matching_records.append(sg_version_entry)
 
         log.debug(
-            "After applying label filters, %d records remain." % len(matching_records)
+            f"After applying label filters, {len(matching_records)} records remain."
         )
 
         if len(matching_records) == 0:
             raise TankDescriptorError(
-                "Cannot find any versions for %s in the App store!" % self
+                f"Cannot find any versions for {self} in the App store!"
             )
 
         # and filter out based on version constraint
@@ -642,24 +817,83 @@ def get_latest_version(self, constraint_pattern=None):
             )
             if version_to_use is None:
                 raise TankDescriptorError(
-                    "'%s' does not have a version matching the pattern '%s'. "
-                    "Available versions are: %s"
-                    % (
-                        self.get_system_name(),
-                        constraint_pattern,
-                        ", ".join(version_numbers),
-                    )
+                    f"'{self.get_system_name()}' does not have a version matching the pattern '{constraint_pattern}'. "
+                    f"Available versions are: {', '.join(version_numbers)}"
                 )
             # get the sg data for the given version
             sg_data_for_version = [
                 d for d in matching_records if d["code"] == version_to_use
             ][0]
 
         else:
-            # no constraints applied. Pick first (latest) match
+            # no constraints applied. Check if latest version meets compatibility requirements
             sg_data_for_version = matching_records[0]
             version_to_use = sg_data_for_version["code"]
 
+            # Create a temporary descriptor to check its manifest
+            temp_descriptor_dict = {
+                "type": "app_store",
+                "name": self._name,
+                "version": version_to_use,
+            }
+            if self._label:
+                temp_descriptor_dict["label"] = self._label
+
+            temp_desc = IODescriptorAppStore(
+                temp_descriptor_dict, self._sg_connection, self._bundle_type
+            )
+            temp_desc.set_cache_roots(self._bundle_cache_root, self._fallback_roots)
+
+            # Check if latest version meets compatibility requirements
+            try:
+                # Download if needed to read manifest
+                if not temp_desc.exists_local():
+                    log.debug(
+                        f"Downloading {version_to_use} to check compatibility requirements"
+                    )
+                    temp_desc.download_local()
+
+                manifest = temp_desc.get_manifest(constants.BUNDLE_METADATA_FILE)
+                if not self._check_minimum_python_version(manifest):
+                    # Latest version is NOT compatible - search for compatible version
+                    current_py_ver = ".".join(str(x) for x in sys.version_info[:3])
+                    min_py_ver = manifest.get("minimum_python_version", "not specified")
+                    log.warning(
+                        f"Auto-update blocked: Latest version {version_to_use} requires Python {min_py_ver}, current is {current_py_ver}."
+                    )
+
+                    # Try to find highest compatible version in cache
+                    compatible_desc = self._find_compatible_cached_version(
+                        matching_records
+                    )
+                    if compatible_desc:
+                        log.info(
+                            f"Using cached compatible version {compatible_desc.get_version()} instead of latest {version_to_use}"
+                        )
+                        return compatible_desc
+                    else:
+                        log.warning(
+                            f"No compatible cached version found. Proceeding with latest version {version_to_use} "
+                            "(may cause compatibility issues)."
+                        )
+                        # Reset to use latest - user will need to upgrade Python
+                        # This maintains current behavior when no alternative exists
+                else:
+                    log.debug(
+                        f"Latest version {version_to_use} meets all compatibility requirements"
+                    )
+            except (
+                TankDescriptorError,
+                TankMissingManifestError,
+                TankAppStoreError,
+                ShotgunAttachmentDownloadError,
+                IOError,
+                OSError,
+            ) as e:
+                log.warning(
+                    f"Could not check compatibility requirements for {version_to_use}: {e}. Proceeding with auto-update."
+                )
+
         # make a descriptor dict
         descriptor_dict = {
             "type": "app_store",