1#!/usr/bin/env python3 2# -*- coding: utf-8 -*- 3# 4# Copyright (c) 2024 Huawei Device Co., Ltd. 5# Licensed under the Apache License, Version 2.0 (the "License"); 6# you may not use this file except in compliance with the License. 7# You may obtain a copy of the License at 8# 9# http://www.apache.org/licenses/LICENSE-2.0 10# 11# Unless required by applicable law or agreed to in writing, software 12# distributed under the License is distributed on an "AS IS" BASIS, 13# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 14# See the License for the specific language governing permissions and 15# limitations under the License. 16# 17 18import json 19import os 20import re 21import fnmatch 22import sys 23from abc import ABC, abstractmethod 24import xml.etree.ElementTree as ET 25 26from Utils import ChangeFileEntity, XTSTargetUtils, PathUtils, MatchConfig, HOME 27 28 29class Ci_Manager(ABC): 30 31 @abstractmethod 32 def get_targets_from_change(self, change_list: list): 33 pass 34 35 def write_result(self, target_path_set: set, target_set: set): 36 print(f"{self.__class__.__name__} append build_targets : {self._build_targets}") 37 print(f"{self.__class__.__name__} append build_paths : {self._build_paths}") 38 39 xts_root_target = PathUtils.get_root_target(self._xts_root_dir) 40 if xts_root_target in target_set: 41 print("compile full testsuites") 42 return 43 if xts_root_target in self._build_targets: 44 target_set.add(xts_root_target) 45 target_path_set.clear() 46 print("compile full testsuites") 47 return 48 49 target_set.update(set(self._build_targets)) 50 target_path_set.update(set(self._build_paths)) 51 52 53class ComponentManager(Ci_Manager): 54 55 def __init__(self, xts_root_dir, code_root_dir): 56 self._xts_root_dir = xts_root_dir 57 self._code_root_dir = code_root_dir 58 self._build_paths = [] 59 self._build_targets = [] 60 61 def get_targets_from_change(self, change_list): 62 for changeFileEntity in change_list: 63 if changeFileEntity.path not in MatchConfig.get_xts_path_list(): 64 ret = self.getTargetsPaths(changeFileEntity) 65 if ret == 1: 66 continue 67 return 0 68 69 def getTargetsPaths(self, change_file_entity: ChangeFileEntity): 70 # 获取部件名 71 try: 72 bundle_name = self.getBundleName(change_file_entity.path) 73 except Exception as e: 74 print("warning: Get bundle_name failed from cmoponent repository {}".format(change_file_entity.name)) 75 return 1 76 print(f"{self.__class__.__name__} append bundle_name : {bundle_name}") 77 # 部件名(partname)获取paths 78 paths = XTSTargetUtils.getPathsByBundle([bundle_name], self._xts_root_dir) 79 if paths: 80 change_file_entity.set_already_match_utils(True) 81 self._build_paths += paths 82 return 0 83 84 def getBundleName(self, path) -> str: 85 with open(os.path.join(HOME, path, "bundle.json"), 'r') as file: 86 data = json.load(file) 87 bundle_name = data['component']['name'] 88 return bundle_name 89 90 91class XTSManager(Ci_Manager): 92 93 def __init__(self, xts_root_dir, code_root_dir): 94 self._xts_root_dir = xts_root_dir 95 self._code_root_dir = code_root_dir 96 self._build_paths = [] 97 self._build_targets = [] 98 self._need_all = False 99 100 def get_targets_from_change(self, change_list): 101 for changeFileEntity in change_list: 102 # tools仓修改，编译全量 103 if changeFileEntity.path == "test/xts/tools": 104 self._need_all = True 105 changeFileEntity.set_already_match_utils(True) 106 if changeFileEntity.path in self._xts_root_dir: 107 # 只有当前编译的xts仓修改参与计算 108 ret = self.getTargetsPaths(changeFileEntity) 109 if ret == 1: 110 print("warning: failed to parse modification of repository {}".format(self.__class__.__name__)) 111 return 1 112 if self._need_all: 113 self._build_targets += PathUtils.get_all_build_target(self._xts_root_dir) 114 return 0 115 116 # 获取path接口 117 def getTargetsPaths(self, changeFileEntity: ChangeFileEntity): 118 # 修改和新增 119 for file in changeFileEntity.add + changeFileEntity.modified: 120 # file转为绝对路径 121 file = os.path.join(self._code_root_dir, file) 122 # 筛选掉例外的目录 123 if PathUtils.isMatchRules(file, MatchConfig.get_exception_path()): 124 continue 125 # 当前文件路径或父已存在,跳过 126 if PathUtils.isTargetContains(self._build_paths, file): 127 continue 128 # 当前file对应BUILD.gn路径 129 build_File = XTSTargetUtils.get_current_Build(self._xts_root_dir, file) 130 # 计算到根目录或指定目录,直接编译全量 131 if (os.path.dirname(build_File) == self._xts_root_dir or 132 PathUtils.isMatchRules(file, MatchConfig.get_all_com_path())): 133 self._need_all = True 134 else: 135 self._build_paths.append(os.path.dirname(build_File)) 136 # 删除 137 for file in changeFileEntity.delete: 138 # file转为绝对路径 139 file = os.path.join(self._code_root_dir, file) 140 # 筛选掉例外的目录 141 if PathUtils.isMatchRules(file, MatchConfig.get_exception_path()): 142 continue 143 # 当前文件路径或父已存在,跳过 144 if PathUtils.isTargetContains(self._build_paths, file): 145 continue 146 # 当前存在的最外层路径 147 exist_path = PathUtils.get_current_exist(self._xts_root_dir, os.path.dirname(file)) 148 build_File = XTSTargetUtils.get_current_Build(self._xts_root_dir, exist_path) 149 # 计算到根目录或指定目录,直接编译全量 150 if (os.path.dirname(build_File) == self._xts_root_dir or 151 PathUtils.isMatchRules(file, MatchConfig.get_all_com_path())): 152 self._need_all = True 153 else: 154 self._build_paths.append(os.path.dirname(build_File)) 155 return 0 156 157 158class WhitelistManager(Ci_Manager): 159 160 def __init__(self, xts_root_dir, code_root_dir): 161 self._xts_root_dir = xts_root_dir 162 self._code_root_dir = code_root_dir 163 self._build_paths = [] 164 self._build_targets = [] 165 self.full_impact_flag = "FULL_IMPACT" 166 self.full_impact_flag_totally = "FULL_IMPACT_TOTALLY" 167 168 def get_targets_from_change(self, change_list): 169 for changeFileEntity in change_list: 170 if changeFileEntity.path not in MatchConfig.get_xts_path_list(): 171 ret = self.getTargetsandPaths(changeFileEntity) 172 if ret == 1: 173 return 1 174 return 0 175 176 def getTargetsandPaths(self, change_file_entity): 177 white_list = MatchConfig.get_white_list_repo() 178 if change_file_entity.path not in white_list: 179 return 0 180 bundles = white_list[change_file_entity.path]["add_bundle"] 181 add_targets = white_list[change_file_entity.path]["add_target"] 182 targets = [] 183 if add_targets: 184 if add_targets[0] == self.full_impact_flag: 185 targets = PathUtils.get_all_build_target(self._xts_root_dir, 0) 186 if add_targets[0] == self.full_impact_flag_totally: 187 targets = PathUtils.get_all_build_target(self._xts_root_dir, 1) 188 else: 189 xts_parts = "/".join([p for p in os.path.normpath(self._xts_root_dir).split(os.sep) if p][-3:]) 190 for target in add_targets: 191 if xts_parts in target: 192 targets.append(target) 193 change_file_entity.set_already_match_utils(True) 194 if bundles: 195 paths = XTSTargetUtils.getPathsByBundle(bundles, self._xts_root_dir) 196 if paths: 197 self._build_paths += paths 198 if targets: 199 self._build_targets += targets 200 return 0 201 202 203class OldPreciseManager(Ci_Manager): 204 205 def __init__(self, xts_root_dir, code_root_dir): 206 self._xts_root_dir = xts_root_dir 207 self._code_root_dir = code_root_dir 208 self._build_paths = [] 209 self._build_targets = [] 210 self._precise_compilation_file = os.path.join(HOME, "test", "xts", "tools", "config", 211 "precise_compilation.json") 212 self._init_old_precise_map() 213 214 def _init_old_precise_map(self): 215 self._old_precise_map = {} 216 with open(self._precise_compilation_file, 'r') as file: 217 data_list = json.load(file) 218 for item in data_list: 219 name = item['name'] 220 build_target = item['buildTarget'] 221 self._old_precise_map[name] = build_target 222 223 def get_targets_from_change(self, change_list): 224 for changeFileEntity in change_list: 225 if changeFileEntity.path not in MatchConfig.get_xts_path_list() and \ 226 not changeFileEntity.get_already_match_utils(): 227 if self._xts_root_dir.endswith("acts"): 228 ret = self.getTargets(changeFileEntity) 229 if ret == 1: 230 pass 231 else: 232 self._build_targets += PathUtils.get_all_build_target(self._xts_root_dir) 233 return 0 234 235 # 获取path接口 236 def getTargets(self, changeFileEntity: ChangeFileEntity): 237 # 获取开源仓名 238 repo_name = self.search_repo_name(changeFileEntity.path) 239 # precise_compilation.json配置文件中获取对应目标 240 if repo_name in self._old_precise_map: 241 self._build_targets.append(self._old_precise_map[repo_name]) 242 return 0 243 244 def getTargetsbyRepoName(self, repo_name): 245 with open(repo_name, 'r') as file: 246 data_list = json.load(file) 247 # 遍历列表中的每个字典 248 for item in data_list: 249 # 获取 name 和 buildTarget 的值 250 name = item['name'] 251 build_target = item['buildTarget'] 252 # 打印结果 253 print(f'Name: {name}, Build Target: {build_target}') 254 255 def search_repo_name(self, repo_path, directory=os.path.join(HOME, ".repo", "manifests")): 256 for root, dirs, files in os.walk(directory): 257 for filename in fnmatch.filter(files, '*.xml'): 258 file_path = os.path.join(root, filename) 259 for child in ET.parse(file_path).getroot().findall('project'): 260 if 'path' in child.attrib and child.attrib['path'] == repo_path: 261 if 'gitee_name' in child.attrib: 262 return child.attrib['gitee_name'] 263 if 'name' in child.attrib: 264 return child.attrib['name'] 265 return None 266 267class GetInterfaceData(Ci_Manager): 268 269 def __init__(self, xts_root_dir, code_root_dir): 270 self._xts_root_dir = xts_root_dir 271 self._code_root_dir = code_root_dir 272 self._build_paths = [] 273 self._build_targets = [] 274 self.sum_change_list_path = [] 275 self.bundle_name_list = [] 276 self.match_path_list = [] 277 self.no_match_path_list = [] 278 279 # 截取路径 280 def get_first_levels_path(self, path, num): 281 normalized_path = os.path.normpath(path) 282 parts = normalized_path.split(os.sep) 283 return os.sep.join(parts[:num]) 284 285 def path_error(self, path_list): 286 if len(path_list) > 0: 287 raise Exception('error: Failed to obtain interface file ownership, Please config in test/xts/tools/config/ci_api_part_name.json') 288 289 def get_targets_from_change(self, change_list): 290 if "acts" in self._xts_root_dir: 291 # 分开处理三个 interface 仓 292 self.get_c_bundle_name(change_list, MatchConfig.get_interface_json_c_data()) 293 self.get_driver_interface_bundle_name(change_list, MatchConfig.get_interface_json_driver_interface_data()) 294 self.get_js_bundle_name(change_list, MatchConfig.get_interface_json_js_data()) 295 # 筛选出未匹配路径 296 for path in self.sum_change_list_path: 297 if path not in self.match_path_list: 298 # feature 分支.d.ets后缀无需编译用例 299 if not path.endswith(".d.ets"): 300 self.no_match_path_list.append(path) 301 302 self.bundle_name_list = list(set(self.bundle_name_list)) 303 self._build_targets = list(set(self._build_targets)) 304 print('INTERFACE_BUNDLE_NAME = ', self.bundle_name_list) 305 306 # 抛出未匹配到 bundle_name 的路径 307 try: 308 self.path_error(self.no_match_path_list) 309 except Exception as e: 310 print(e) 311 for path in self.no_match_path_list: 312 print("error: Cannot match path {}".format(path)) 313 sys.exit(1) 314 315 # 根据bundle_name 查找对应 build_paths 316 self._build_paths = XTSTargetUtils.getPathsByBundle(self.bundle_name_list, self._xts_root_dir) 317 else: 318 for change in change_list: 319 if change.path in MatchConfig.get_interface_path_list(): 320 self._build_targets += PathUtils.get_all_build_target(self._xts_root_dir) 321 return 322 323 # 处理 interface/sdk-js 仓 324 def get_js_bundle_name(self, change_list, js_json_data): 325 # 获取 change_list interface/sdk-js 仓数据 326 for store in change_list: 327 if store.path != MatchConfig.get_interface_path_list()[0]: 328 continue 329 paths = store.add + store.modified + store.delete 330 self.sum_change_list_path += paths 331 targete_data = [data for data in js_json_data if data['path'] in paths] 332 for _data in targete_data: 333 store.set_already_match_utils(True) 334 self.match_path_list.append(_data.get('path')) 335 # 找出 bundle_name 加入 self.bundle_name_list 336 self.bundle_name_list += _data.get('bundle_name') 337 # 针对 interface/sdk-js/kits、interface/sdk-js/arkts 在配置文件中查找 build_target 338 self._build_targets += _data.get('build_targets') 339 # 根据路径匹配 340 for path in paths: 341 for source_path in js_json_data: 342 if PathUtils.is_parent_path(source_path.get('path'), path): 343 self.match_path_list.append(path) 344 store.set_already_match_utils(True) 345 self.bundle_name_list += source_path.get('bundle_name') 346 347 # 处理 driver_interface 仓 348 def get_driver_interface_bundle_name(self, change_list, driver_interface_json_data): 349 add_bundle_json_path = [] 350 for store in change_list: 351 # 获取 change_list driver_interface 仓数据 352 if store.path != MatchConfig.get_interface_path_list()[2]: 353 continue 354 for path in store.add + store.modified + store.delete: 355 self.sum_change_list_path.append(path) 356 for source_path in driver_interface_json_data: 357 # 从 drivers/interface 下第一级目录截取路径在配置文件中查找 bundle_name 358 if PathUtils.is_parent_path(source_path.get('path'), path): 359 store.set_already_match_utils(True) 360 self.bundle_name_list += source_path.get('bundle_name') 361 self.match_path_list.append(path) 362 363 # 查看新增目录下是否有 bundle.json 364 for path in store.add: 365 if os.path.basename(path) == 'bundle.json': 366 try: 367 # 发现 bundle.json 后进去文件寻找 bundle_name 368 with open(os.path.abspath(__file__).split('/test/')[0] + '/' + path, 'r') as d: 369 for k, v in json.load(d).items(): 370 if k == 'component': 371 add_bundle_json_path.append([path, v['name']]) 372 except FileNotFoundError: 373 print("error: The repository drivers/interface has a new directory and no bundle.json is found under the directory, Please add bundle.json") 374 sys.exit(1) 375 376 # 处理新增目录下其他文件 377 for path in store.add: 378 for path_name in add_bundle_json_path: 379 if self.get_first_levels_path(path, 3) == self.get_first_levels_path(path_name[0], 3): 380 self.match_path_list.append(path) 381 store.set_already_match_utils(True) 382 self.bundle_name_list.append(path_name[1]) 383 384 # 处理 interface/sdk_c 仓 385 def get_c_bundle_name(self, change_list, c_json_data): 386 for store in change_list: 387 # 获取 change_list interface/sdk_c 仓数据 388 if store.path != MatchConfig.get_interface_path_list()[1]: 389 continue 390 for path in store.add + store.modified + store.delete: 391 self.sum_change_list_path.append(path) 392 for source_path in c_json_data: 393 # 根据目录匹配 394 if PathUtils.is_parent_path(source_path.get('path'), path): 395 self.match_path_list.append(path) 396 store.set_already_match_utils(True) 397 self.bundle_name_list += source_path.get('bundle_name') 398 # 根据文件匹配 399 elif path == source_path.get('path'): 400 self.match_path_list.append(path) 401 store.set_already_match_utils(True) 402 self.bundle_name_list += source_path.get('bundle_name')