feat: 安装预设插件锁定版本 (closed TencentBlueKing#2482)

apps/backend/subscription/steps/adapter.py

@@ -285,19 +285,19 @@ def max_ids_by_key(self, contained_os_cpu_items: List[Dict[str, Any]]) -> List[i
     def format2policy_packages_new(
         self, plugin_id: int, plugin_name: str, plugin_version: str, config_templates: List[Dict[str, Any]]
     ) -> List[Dict[str, Any]]:
-        latest_flag: str = "latest"
-        is_tag: bool = Tag.objects.filter(
-            target_id=plugin_id, name=latest_flag, target_type=TargetType.PLUGIN.value
-        ).exists()
+        tags: List[str] = ["latest", "stable"]
+        is_tag: Dict[str, bool] = {
+            tag: Tag.objects.filter(target_id=plugin_id, name=tag, target_type=TargetType.PLUGIN.value).exists()
+            for tag in tags
+        }
 
-        if plugin_version != latest_flag or is_tag:
-            # 如果 latest 是 tag，走取指定版本的逻辑
-            packages = self.get_packages(plugin_name, plugin_version)
-        else:
+        if plugin_version in tags and not is_tag[plugin_version]:
             max_pkg_ids: List[int] = self.max_ids_by_key(
                 list(models.Packages.objects.filter(project=plugin_name).values("id", "os", "cpu_arch"))
             )
             packages = models.Packages.objects.filter(id__in=max_pkg_ids)
+        else:
+            packages = self.get_packages(plugin_name, plugin_version)
 
         if not packages:
             raise errors.PluginValidationError(
@@ -307,11 +307,11 @@ def format2policy_packages_new(
         os_cpu__config_templates_map = defaultdict(list)
         for template in config_templates:
             is_main_template = template["is_main"]
-            if template["version"] != latest_flag or is_tag:
-                plugin_version_set = {plugin_version, "*"}
+            if plugin_version in tags and not is_tag[plugin_version]:
+                tag_packages_version_set = set(packages.values_list("version", flat=True))
+                plugin_version_set = tag_packages_version_set | {"*"}
             else:
-                latest_packages_version_set = set(packages.values_list("version", flat=True))
-                plugin_version_set = latest_packages_version_set | {"*"}
+                plugin_version_set = {plugin_version, "*"}
 
             max_config_tmpl_ids: typing.List[int] = self.max_ids_by_key(
                 list(
@@ -479,6 +479,7 @@ def get_matching_package_obj(self, os_type: str, cpu_arch: str, bk_biz_id: int)
     def get_matching_config_tmpl_objs(
         self, os_type: str, cpu_arch: str, package: models.Packages, bk_biz_id: int, config: Dict
     ) -> List[models.PluginConfigTemplate]:
+        # 如果当前业务设定了业务最大版本，重新从数据库中获取
         if str(bk_biz_id) in self.plugin_version_config():
             config_tmpl = (
                 models.PluginConfigTemplate.objects.filter(
@@ -494,6 +495,7 @@ def get_matching_config_tmpl_objs(
         return self.config_tmpl_obj_gby_os_key.get(self.get_os_key(os_type, cpu_arch), [])
 
     def get_biz_package(self, plugin_name: str, os_type: str, cpu_arch: str, biz_version: str):
+        """获取业务锁定版本的插件包"""
         packages_all = self.get_packages(plugin_name, biz_version)
         packages = packages_all.filter(
             id__in=[pkg.id for pkg in packages_all if version.Version(pkg.version) <= version.Version(biz_version)]
@@ -508,13 +510,15 @@ def get_biz_package(self, plugin_name: str, os_type: str, cpu_arch: str, biz_ver
 
     @staticmethod
     def plugin_version_config():
+        """业务锁定版本配置"""
         plugin_version_config: Dict[str, Dict[str, str]] = models.GlobalSettings.get_config(
             models.GlobalSettings.KeyEnum.PLUGIN_VERSION_CONFIG.value, default={}
         )
         return plugin_version_config
 
     @staticmethod
     def get_biz_version(package: models.Packages, bk_biz_id: int):
+        """获取业务锁定版本"""
         plugin_version_config = PolicyStepAdapter.plugin_version_config()
         biz_version = None
         if str(bk_biz_id) in plugin_version_config:
@@ -530,7 +534,7 @@ def get_biz_version(package: models.Packages, bk_biz_id: int):
         return biz_version
 
     def get_packages(self, plugin_name: str, plugin_version: str, biz_version: str = None):
-        """如果不存在某个版本，则返回最大id的版本"""
+        """如果不存在某个版本，则取最大id的版本"""
         all_packages = models.Packages.objects.filter(project=plugin_name).values("id", "os", "cpu_arch", "version")
         version_packages = {pkg["id"]: pkg for pkg in all_packages if pkg["version"] == plugin_version}
         package_ids = set(version_packages.keys())