#!/usr/bin/env python3 # -*- coding: utf-8 -*- # # Copyright (c) 2024 Huawei Device Co., Ltd. # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # import json import os import sys from pathlib import Path from abc import ABC, abstractmethod from utils import ( ChangeFileEntity, XTSTargetUtils, PathUtils, MatchConfig, CODEBASE, InterfaceRepo, WhitelistProcessor ) class Ci_Manager(ABC): @abstractmethod def get_targets_from_change(self, change_list: list): pass def write_result(self, target_path_set: set, target_set: set): print(f"{self.__class__.__name__} append build_targets : {self._build_targets}") print(f"{self.__class__.__name__} append build_paths : {self._build_paths}") xts_root_target = PathUtils.get_root_target(self._xts_root_dir) if xts_root_target in target_set: print("compile full testsuites") return if xts_root_target in self._build_targets: target_set.add(xts_root_target) target_path_set.clear() print("compile full testsuites") return target_set.update(set(self._build_targets)) target_path_set.update(set(self._build_paths)) class ComponentManager(Ci_Manager): def __init__(self, xts_root_dir, code_root_dir): self._xts_root_dir = xts_root_dir self._code_root_dir = code_root_dir self._build_paths = [] self._build_targets = [] self._ieyes_xts_cfg_file = os.path.join(self._code_root_dir, 'ieyes_xts.json') self._ieyes_xts_map = {} def get_targets_from_change(self, change_list): if not os.path.exists(self._ieyes_xts_cfg_file): return 0 with open(self._ieyes_xts_cfg_file, 'r') as f: self._ieyes_xts_map = json.load(f) for changeFileEntity in change_list: if changeFileEntity.path not in MatchConfig.get_xts_path_list(): ret = self.getTargetsPaths(changeFileEntity) if ret == 1: continue return 0 def getTargetsPaths(self, change_file_entity: ChangeFileEntity): # 获取部件名 try: bundle_name = self.getBundleName(change_file_entity.path) except Exception as e: print("warning: Get bundle_name failed from cmoponent repository {}".format(change_file_entity.name)) return 1 print(f"{self.__class__.__name__} append bundle_name : {bundle_name}") # 部件名(partname)获取paths paths = XTSTargetUtils.getPathsByBundle([bundle_name], self._xts_root_dir, self._ieyes_xts_map) if paths: change_file_entity.set_already_match_utils(True) self._build_paths += paths return 0 def getBundleName(self, path) -> str: with open(os.path.join(CODEBASE, path, "bundle.json"), 'r') as file: data = json.load(file) bundle_name = data['component']['name'] return bundle_name class XTSManager(Ci_Manager): def __init__(self, xts_root_dir, code_root_dir): self._xts_root_dir = xts_root_dir self._code_root_dir = code_root_dir self._build_paths = [] self._build_targets = [] self._need_all = False def get_targets_from_change(self, change_list): for changeFileEntity in change_list: if changeFileEntity.path in self._xts_root_dir: # 只有当前编译的xts仓修改参与计算 ret = self.getTargetsPaths(changeFileEntity) if ret == 1: print("warning: failed to parse modification of repository {}".format(self.__class__.__name__)) return 1 if self._need_all: self._build_targets += PathUtils.get_all_build_target(self._xts_root_dir) return 0 # 获取path接口 def getTargetsPaths(self, changeFileEntity: ChangeFileEntity): # 修改和新增 for file in changeFileEntity.add + changeFileEntity.modified: # file转为绝对路径 file = os.path.join(self._code_root_dir, file) # 筛选掉例外的目录 if PathUtils.isMatchRules(file, MatchConfig.get_exception_path()): continue # 当前文件路径或父已存在,跳过 if PathUtils.isTargetContains(self._build_paths, file): continue # 当前file对应BUILD.gn路径 build_File = XTSTargetUtils.get_current_Build(self._xts_root_dir, file) # 计算到根目录或指定目录,直接编译全量 if (os.path.dirname(build_File) == self._xts_root_dir or PathUtils.isMatchRules(file, MatchConfig.get_all_com_path())): self._need_all = True else: self._build_paths.append(os.path.dirname(build_File)) # 删除 for file in changeFileEntity.delete: # file转为绝对路径 file = os.path.join(self._code_root_dir, file) # 筛选掉例外的目录 if PathUtils.isMatchRules(file, MatchConfig.get_exception_path()): continue # 当前文件路径或父已存在,跳过 if PathUtils.isTargetContains(self._build_paths, file): continue # 当前存在的最外层路径 exist_path = PathUtils.get_current_exist(self._xts_root_dir, os.path.dirname(file)) build_File = XTSTargetUtils.get_current_Build(self._xts_root_dir, exist_path) # 计算到根目录或指定目录,直接编译全量 if (os.path.dirname(build_File) == self._xts_root_dir or PathUtils.isMatchRules(file, MatchConfig.get_all_com_path())): self._need_all = True else: self._build_paths.append(os.path.dirname(build_File)) return 0 class WhitelistManager(Ci_Manager): def __init__(self, xts_root_dir, code_root_dir, suite_type): self._xts_root_dir = xts_root_dir self._code_root_dir = code_root_dir self._build_paths = [] self._build_targets = [] self.full_impact_flag = "FULL_IMPACT" self.full_impact_flag_totally = "FULL_IMPACT_TOTALLY" self._suite_type = suite_type self._suite_name = XTSTargetUtils.get_suite_name(xts_root_dir) def get_targets_from_change(self, change_list): for chg_file_entity in change_list: if chg_file_entity.path not in MatchConfig.get_xts_path_list(): ret = self.get_targets_and_paths(chg_file_entity) if ret == 1: return 1 return 0 def get_targets_and_paths(self, chg_file_entity): white_list = MatchConfig.get_white_list_repo() if chg_file_entity.path not in white_list: return 0 chg_file_entity.set_already_match_utils(True) whitelist_item = white_list.get(chg_file_entity.path) subdirs = whitelist_item.get("subdirectory") if subdirs: self.get_targets_and_paths_from_subdirs(chg_file_entity, subdirs) else: self.get_targets_and_paths_by_cfg( whitelist_item.get("suite_type"), whitelist_item.get("add_bundle"), whitelist_item.get("add_target")) return 0 def get_targets_and_paths_from_subdirs(self, chg_file_entity, subdirs): chg_files = chg_file_entity.add + chg_file_entity.modified + chg_file_entity.delete for dir in subdirs: matched = False for file in chg_files: if file.startswith(dir): matched = True break if not matched: continue config = subdirs.get(dir) self.get_targets_and_paths_by_cfg(config.get("suite_type"), config.get("add_bundle"), config.get("add_target")) def _check_conflict_targets(self, suite_targets): targets = set(suite_targets) x_flags = {self.full_impact_flag, self.full_impact_flag_totally} if set(x_flags) & targets and len(targets) != 1: print(f"[ERROR] These flags are exclusive: {x_flags}, " f"when set, do make sure the flag is the only target in the configuration.") sys.exit(1) def get_targets_and_paths_by_cfg(self, suite_type, add_bundle, add_target): if suite_type and not list(set(suite_type) & set(self._suite_type)): return targets = [] suite_targets = add_target.get(self._suite_name) if add_target else None if suite_targets: self._check_conflict_targets(suite_targets) if self.full_impact_flag in suite_targets: targets = PathUtils.get_all_build_target(self._xts_root_dir, 0) elif self.full_impact_flag_totally in suite_targets: targets = PathUtils.get_all_build_target(self._xts_root_dir, 1) else: targets += XTSTargetUtils.filter_suite_targets(self._suite_name, suite_targets) if add_bundle: paths = XTSTargetUtils.getPathsByBundle(add_bundle, self._xts_root_dir) if paths: self._build_paths += paths if targets: self._build_targets += targets class GetInterfaceData(Ci_Manager): def __init__(self, xts_root_dir, code_root_dir, suite_type): self._xts_root_dir = xts_root_dir self._code_root_dir = code_root_dir self._build_paths = [] self._build_targets = [] self.sum_change_list_path = [] self.bundle_name_list = [] self.match_path_list = [] self.no_match_path_list = [] self._suite_type = suite_type self._suite_name = XTSTargetUtils.get_suite_name(xts_root_dir) # 截取路径 def get_first_levels_path(self, path, num): normalized_path = os.path.normpath(path) parts = normalized_path.split(os.sep) return os.sep.join(parts[:num]) def path_error(self, path_list): if len(path_list) > 0: print("[ERROR] Failed to obtain interface file ownership. " "Please config in test/xts/tools/config/ci_api_part_name.json") for path in path_list: print("[ERROR] Cannot match path {}".format(path)) raise RuntimeError("Interface file ownership mismatch") def get_targets_from_change(self, change_list): for store in change_list: if "hap_static" in self._suite_type and \ store.path in MatchConfig.get_interface_path_mapping().values(): self._build_targets = ["xts_{}".format(os.path.basename(self._xts_root_dir))] return candidates = self._filter_by_whitelist(change_list) # 处理所有 interface 仓 for repo_name in InterfaceRepo: self.append_bundles_and_targets(candidates, MatchConfig.get_interface_path(repo_name), MatchConfig.get_interface_bundle_config(repo_name)) self.append_extra_from_driver_interface(candidates) # 筛选出未匹配路径 for path in self.sum_change_list_path: if path not in self.match_path_list: self.no_match_path_list.append(path) self.bundle_name_list = list(set(self.bundle_name_list)) self._build_targets = list(set(self._build_targets)) print(f"INTERFACE_BUNDLE_NAME = {self.bundle_name_list}") # 抛出未匹配到 bundle_name 的路径 self.path_error(self.no_match_path_list) # 根据bundle_name 查找对应 build_paths self._build_paths = XTSTargetUtils.getPathsByBundle(self.bundle_name_list, self._xts_root_dir) def _filter_by_whitelist(self, change_list: list[ChangeFileEntity]): """ Keep only interface repos and filter out whitelist files. """ result = [] repo_map = MatchConfig.get_interface_path_mapping() interface_paths = repo_map.values() processor = WhitelistProcessor( MatchConfig.get_interface_whitelist_config(), repo_map, self._code_root_dir ) def rm_path_prefix(paths: list[str], prefix: str) -> list[str]: prefix_path = Path(prefix) return [ str(Path(p).relative_to(prefix_path)) for p in paths if Path(p).is_relative_to(prefix_path) ] for chg in change_list: if chg.path in interface_paths: _chg = ChangeFileEntity(chg.name, chg.path) _chg.addAddPaths(rm_path_prefix(processor.filter_files(chg.add), _chg.path)) _chg.addModifiedPaths(rm_path_prefix(processor.filter_files(chg.modified), _chg.path)) _chg.addDeletePaths(rm_path_prefix(processor.filter_files(chg.delete), _chg.path)) result.append(_chg) return result def _validate_config_entry(self, conf, early_exit = False): msg = "Invalid interface configuration" err = RuntimeError(msg) if not isinstance(conf, dict): print(f"[ERROR] {msg}: config entry must be a dictionary") raise err path = conf.get("path") if not path or not isinstance(path, str): print(f"[ERROR] {msg}: 'path' is required as a non-empty string") raise err if early_exit: return bundle_names = conf.get("bundle_name") if not ( isinstance(bundle_names, list) and bundle_names and all(isinstance(b, str) and b for b in bundle_names) ): print(f"[ERROR] {msg}: 'bundle_name' must be a non-empty list of non-empty strings.") raise err build_targets = conf.get("build_targets") if not ( isinstance(build_targets, dict) and ( not build_targets # valid for {} or all( isinstance(targets, list) and targets and all(isinstance(t, str) and t for t in targets) for targets in build_targets.values() ) ) ): print(f"[ERROR] {msg}: 'build_targets' must be a dictionary. " f"If not empty, every value must be a non-empty list of non-empty strings.") raise err # 接口仓统一处理逻辑 def append_bundles_and_targets(self, change_list, interface_path, interface_configs): for chg in change_list: if chg.path != interface_path: continue for path in chg.add + chg.modified + chg.delete: self.sum_change_list_path.append(path) for conf in interface_configs: # 路径 && 目录均不匹配 if path != conf.get("path") and not \ PathUtils.is_parent_path(conf.get("path"), path): continue self.match_path_list.append(path) chg.set_already_match_utils(True) bundle_names = conf.get("bundle_name", []) build_targets = XTSTargetUtils.filter_suite_targets(self._suite_name, conf.get("build_targets", {}).get(self._suite_name, [])) self.bundle_name_list += bundle_names self._build_targets += build_targets # driver_interface 仓额外处理逻辑 def append_extra_from_driver_interface(self, change_list): add_bundle_json_path = [] interface_path = MatchConfig.get_interface_path(InterfaceRepo.DRIVERS) for chg in change_list: if chg.path != interface_path: continue # 查看新增目录下是否有 bundle.json for path in chg.add: if os.path.basename(path) == 'bundle.json': try: # 发现 bundle.json 后进去文件寻找 bundle_name with open(os.path.abspath(__file__).split('/test/')[0] + '/' + path, 'r') as d: for k, v in json.load(d).items(): if k == 'component': add_bundle_json_path.append([path, v['name']]) except FileNotFoundError: print("[ERROR] The repository drivers/interface has a new directory " "but no bundle.json is found under the directory, Please add bundle.json") sys.exit(1) # 处理新增目录下其他文件 for path in chg.add: if path in self.match_path_list: continue for path_name in add_bundle_json_path: if self.get_first_levels_path(path, 3) == self.get_first_levels_path(path_name[0], 3): self.match_path_list.append(path) chg.set_already_match_utils(True) self.bundle_name_list.append(path_name[1])