1#!/usr/bin/env python3 2# coding=utf-8 3 4# 5# Copyright (c) 2022 Huawei Device Co., Ltd. 6# Licensed under the Apache License, Version 2.0 (the "License"); 7# you may not use this file except in compliance with the License. 8# You may obtain a copy of the License at 9# 10# http://www.apache.org/licenses/LICENSE-2.0 11# 12# Unless required by applicable law or agreed to in writing, software 13# distributed under the License is distributed on an "AS IS" BASIS, 14# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 15# See the License for the specific language governing permissions and 16# limitations under the License. 17# 18 19import os 20import threading 21import time 22 23from devicetest.core.constants import RunResult 24from devicetest.core.constants import RunSection 25from devicetest.core.error_message import ErrorMessage 26from devicetest.core.exception import DeviceTestError 27from devicetest.core.record import ProjectRecord 28from devicetest.log.logger import DeviceTestLog as log 29from devicetest.utils.util import get_base_name 30from xdevice import is_env_pool_run_mode 31from xdevice import Variables 32 33 34def get_decrypt_resource_path(): 35 return Variables.res_dir 36 37 38def set_resource_path(resource_path): 39 DeccVariable.project.resource_path = resource_path 40 41 42def get_testsuit_path(): 43 return DeccVariable.project.test_suite_path 44 45 46def get_project_path(): 47 """ 48 get project path 49 :return: prcject_path 50 """ 51 try: 52 if DeccVariable.project.project_path: 53 return DeccVariable.project.project_path 54 55 project_path = os.path.dirname(Variables.top_dir) 56 if project_path is None: 57 log.info("project path is None.") 58 raise Exception("") 59 if not os.path.exists(project_path): 60 log.info("project path not exists.") 61 log.debug("project path:{}".format(project_path)) 62 raise Exception("") 63 return project_path 64 65 except Exception as error: 66 log.error(ErrorMessage.Error_01428.Message.en, 67 error_no=ErrorMessage.Error_01428.Code, 68 is_traceback=True) 69 70 raise DeviceTestError(ErrorMessage.Error_01428.Topic) from error 71 72 73class CurCase: 74 75 def __init__(self, _log): 76 # 用例级别参数 77 self.log = _log 78 self.step_total = 0 # tests 数 79 self.run_section = "" # RunSection.SETUP 80 self.case_result = RunResult.PASSED # 当前用例执行结果 81 self.name = '' # 类方法名，即：用例名case_id 82 self.suite_name = "" # 用例对应哪个测试套 83 self.error_msg = '' # 用例失败信息 84 self.case_screenshot_dir = None # 当前用例失败截图的图片保存路径 85 self.case_flash_error_msg = False # 记录当前y用例是否更新了errorMsg 86 self.is_upload_method_result = False # 记录当前用例是否上报过第一个失败步骤 87 88 self.step_section = RunSection.SETUP # 用例方法setup/test/teardown 89 self.step_index = -1 # 当前步骤序号 90 self.step_error_msg = '' # 当前步骤的失败信息 91 self.step_fail_msg = '' # 用户指定的失败信息 92 self.step_result = RunResult.PASSED # 当前步骤执行结果 93 self.steps_info = [] # 记录测试用例（含测试套子用例）的步骤信息，如步骤名称、执行结果、耗时、截图等 94 self.suite_steps_info = [] # 记录测试套的的步骤信息，如步骤名称、执行结果、耗时、截图等 95 self.auto_record_steps_info = False # 默认记录记录用例操作步骤的信息，设为False，需人工调用record_step添加 96 97 self.test_method = TestMethod(self.log) 98 self.cur_check_cmd = CurCheckCmd() 99 100 # 失败截图相关 101 self.checkepr = False # 啥含义？ 102 self.image_num = 0 103 self.video_num = 0 104 self.dump_xml_num = 0 105 106 # prepare相关 107 self.status = 0 108 self.description = '' # VAR.CurCase.Description memoryLeakReport.py用到 109 self.log_details_path = "./log/test_run_details.log" 110 self.log_path = "./log/test_run_summary.log" 111 self.report_path = '' 112 self.iperf_path = None # wifi相关 113 114 # windows 115 self.win_capture_path = '' # WinCapturePath 116 self.exact_start_time = '' # memoryLeakReport.py用到VAR.CurCase.ExactStartTime 117 self.start_time = '' # memoryLeakReport.py用到VAR.CurCase.StartTime 118 119 self.case_name_file_path = '' # VAR.CurCase.CaseName.FilePath 120 self.device_log = DeviceLog() 121 122 self.case_instance = None 123 self.suite_instance = None 124 125 self.devices = list() 126 self.is_capture_step_screen = False 127 128 @property 129 def testcase(self): 130 return self.case_instance 131 132 @property 133 def testsuite(self): 134 return self.suite_instance 135 136 def set_case_instance(self, case_obj): 137 self.case_instance = case_obj 138 self.set_capture_step_screen_flag(case_obj) 139 140 def set_suite_instance(self, suite_obj): 141 self.suite_instance = suite_obj 142 self.set_capture_step_screen_flag(suite_obj) 143 144 def set_capture_step_screen_flag(self, instance): 145 if instance is None: 146 return 147 configs = instance.configs 148 if configs.get("testargs", None) and configs["testargs"].get("screenshot", [""])[0].lower() == "true": 149 self.is_capture_step_screen = True 150 self.devices = instance.devices 151 152 def set_error_msg(self, error_msg): 153 self.log.debug("set CurCase error msg as: {}".format(error_msg)) 154 self.error_msg = error_msg 155 156 def set_run_section(self, run_section): 157 self.log.debug("set CurCase run section as: {}".format(run_section)) 158 self.run_section = run_section 159 160 def set_case_result(self, case_result): 161 self.case_result = case_result 162 self.log.debug( 163 "set CurCase case result as: {}".format(self.case_result)) 164 165 def set_step_total(self, step_total): 166 self.step_total = step_total 167 self.log.debug( 168 "set CurCase step total as: {}".format(self.step_total)) 169 170 def set_name(self, name): 171 self.name = name 172 self.log.debug("set CurCase name as: {}".format(self.name)) 173 174 def set_step_section(self, step_section): 175 self.step_section = step_section 176 self.log.debug( 177 "set CurCase step section as: {}".format(self.step_section)) 178 179 def set_case_screenshot_dir(self, test_suite_path, task_report_dir, cur_case_full_path): 180 case_screenshot_dir = task_report_dir if is_env_pool_run_mode() else os.path.join(task_report_dir, "script") 181 case_abs_path_base_name = get_base_name(cur_case_full_path, is_abs_name=True) 182 if case_abs_path_base_name and test_suite_path: 183 self.log.debug("case_abs_path_base_name:{}, test_suite_path:{}" 184 .format(case_abs_path_base_name, test_suite_path)) 185 _list = case_abs_path_base_name.split(test_suite_path) 186 if len(_list) == 2: 187 case_screenshot_dir = os.path.abspath( 188 os.path.join(task_report_dir, _list[1].strip(os.sep))) 189 self.case_screenshot_dir = case_screenshot_dir 190 self.log.debug("set case screenshot dir path as: {}".format( 191 self.case_screenshot_dir)) 192 193 def init_stage_var(self, test_method_name, 194 run_section=None, is_error_msg=False): 195 if run_section: 196 self.set_run_section(run_section) 197 self.test_method.init_test_method(test_method_name, 198 is_error_msg=is_error_msg) 199 200 def set_checkepr(self, checkepr): 201 self.checkepr = checkepr 202 self.log.debug("set project checkepr as: {}".format(self.checkepr)) 203 204 def flash_error_msg_and_result(self, error_msg): 205 if not self.error_msg: 206 self.set_error_msg(error_msg) 207 if not self.test_method.error_msg: 208 self.test_method.set_error_msg(error_msg) 209 if self.case_result == RunResult.PASSED: 210 self.set_case_result(RunResult.FAILED) 211 if self.test_method.result == RunResult.PASSED: 212 self.test_method.set_result(RunResult.FAILED) 213 214 def set_step_index(self, index): 215 self.step_index = index 216 217 def set_step_info(self, name, **kwargs): 218 # 不允许外部的同名参数修改内部的记录 219 for builtin_key in ["cost", "result", "screenshot", "_timestamp"]: 220 if builtin_key in kwargs: 221 kwargs.pop(builtin_key) 222 # 耗时为前后Step的记录时间差 223 steps_info = self._get_steps_info_obj() 224 index = len(steps_info) 225 log.info(f'<div class="aw" id="{index}">{name}</div>') 226 if index > 0: 227 last_step = steps_info[-1] 228 last_step["cost"] = round(time.time() - last_step.get("_timestamp"), 3) 229 shots = self._capture_step_screen(name) 230 step = {"name": name, "result": "pass", "cost": 0, "screenshot": shots, "_timestamp": time.time()} 231 step.update(kwargs) 232 steps_info.append(step) 233 self.set_step_index(index) 234 return index 235 236 def update_step_info(self, index, **kwargs): 237 steps_info = self._get_steps_info_obj() 238 max_index = len(steps_info) - 1 239 if not 0 <= index <= max_index: 240 log.warning(f"update step info failed, index must be in [0, {max_index}]") 241 return 242 step = steps_info[index] 243 step.update(kwargs) 244 245 def update_step_shots(self, path, link): 246 if path is None or not os.path.exists(path): 247 return 248 steps_info = self._get_steps_info_obj() 249 if len(steps_info) == 0: 250 return 251 save_name = os.path.basename(path) 252 steps_info[-1].get("screenshot", []).append(f'<a href="{link}" target="_blank">{save_name}</a>') 253 254 def get_steps_extra_head(self): 255 """获取步骤记录的额外表头字段（AZ order）""" 256 steps_info = self._get_steps_info_obj() 257 if len(steps_info) == 0: 258 return [] 259 # 默认表头，与self.set_step_info添加的记录对应 260 heads = ["name", "result", "cost", "screenshot"] 261 # 人工记录模式下，可拓展记录更多数据，获取最长的数据项的key做为表头 262 size, target = len(heads), None 263 for step in steps_info: 264 current_size = len(step) 265 if current_size >= size: 266 target = step 267 size = current_size 268 for name in target.keys(): 269 if name not in heads and not name.startswith("_"): 270 heads.append(name) 271 return sorted(heads[4:]) 272 273 def get_steps_info(self): 274 steps_info = self._get_steps_info_obj() 275 if len(steps_info) > 0: 276 last_step = steps_info[-1] 277 last_step["cost"] = round(time.time() - last_step.get("_timestamp"), 3) 278 return steps_info 279 280 def _get_steps_info_obj(self): 281 """返回测试套或测试用例的记录表""" 282 return self.steps_info if self.case_instance is not None else self.suite_steps_info 283 284 def _capture_step_screen(self, step_name): 285 """ 286 take a screenshot of each device after each step is performed 287 """ 288 shots = [] 289 if self.is_capture_step_screen: 290 from devicetest.controllers.tools.screen_agent import ScreenAgent 291 for device in self.devices: 292 path, link = ScreenAgent.capture_step_picture(device, step_name) 293 if path is None or not os.path.exists(path): 294 continue 295 save_name = os.path.basename(path) 296 shots.append(f'<a href="{link}" target="_blank">{save_name}</a>') 297 return shots 298 299 300class TestMethod: 301 def __init__(self, _log): 302 # 步骤级别参数 303 self.log = _log 304 self.name = 'setup' 305 self.result = RunResult.PASSED 306 self.level = '' 307 self.error_msg = '' 308 self.method_return = '' 309 self.func_ret = [] 310 self.step_flash_fail_msg = False 311 312 def set_result(self, result=None): 313 self.result = result or RunResult.PASSED 314 self.log.debug( 315 "set TestMethod result as: {}".format(self.result)) 316 317 def set_error_msg(self, error_msg): 318 self.error_msg = error_msg 319 self.log.debug( 320 "set TestMethod error msg as: {}".format(self.error_msg)) 321 322 def init_test_method(self, name, is_error_msg=False): 323 self.level = '', 324 self.name = name, 325 self.result = RunResult.PASSED 326 if is_error_msg: 327 self.error_msg = '' 328 self.func_ret.clear() 329 330 def init_aw_method(self): 331 self.error_msg = '' 332 self.result = RunResult.PASSED 333 self.step_flash_fail_msg = False # 记录当前步骤是否更新了failMsg 334 self.func_ret.clear() 335 self.log.debug("init aw method.") 336 337 338class CurStep: 339 pass 340 341 342class Prepare: 343 def __init__(self): 344 self.path = '' 345 self.config = {} 346 347 def set_prepare_path(self, path): 348 if path: 349 self.path = path 350 log.debug("prepare path:{}".format(path)) 351 352 353class Settings: 354 language = '' 355 product = '' # VAR.Settings.Product 356 357 358class Event: 359 configs = {} # VAR.Event.Configs 360 361 362class RedirectLog: 363 task_name = "" # VAR.Project.RedirectLog.TaskName 364 365 366class DeviceLog: 367 ftp_path = [] # VAR.CurCase.DeviceLog.FthPath 368 369 370class ProjectVariables: 371 def __init__(self, _log): 372 # 工程级别参数 373 self.log = _log 374 self.record = ProjectRecord(_log) 375 self.project_path = '' # xdevice工程路径 376 self.aw_path = '' # 测试套aw路径 377 self.testcase_path = '' 378 self.settings = None 379 self.resource_path = '' # 测试套工程resource路径 380 self.test_suite_path = '' # 测试套工程路径 381 self.task_report_dir = '' # 测试用例的框架日志路径 382 self.prepare = Prepare() # prepare 相关 383 self.cur_case_full_path = '' # 记录当前正执行用例全路径 384 self.execute_case_name = None # 记录当前正执行用例id 385 self.config_json = {} # 用户自定义的公共的参数 386 self.property_config = [] # 用户自定义的设备相关的参数 387 self.retry_test_list = [] 388 self.devicename = {} 389 self.step_debug = '' 390 391 self.monkey = False # extension/monkey/monkey.py VAR.Project.Monkey 392 self.task_id = "" # VAR.Project.TaskID memoryLeakReport.py中用到，先记录该字段 VAR。Projec.TaskId 393 self.settings = Settings() # target中用到：VAR。Settings.Language 394 self.total = 0 # VAR.Project.Total 395 self.start_time = '' # memoryLeakReport.py用到VAR.Project.StartTime 396 self.exact_start_time = '' # memoryLeakReport.py用到VAR.Project.ExactStartTime 397 self.finish = 0 # VAR.Project.Finish 398 399 self.config = { 400 } 401 402 self.event = Event() 403 self.test_file = '' # VAR.Project.TestFile 404 self.is_ticc_server = False # GlobalParam.IS_TICC_SERVER 405 self.redirect_log = RedirectLog() 406 407 def set_project_path(self, project_path=None): 408 self.project_path = project_path or get_project_path() 409 self.log.debug("project path is: {}".format(self.project_path)) 410 411 def set_aw_path(self, aw_path): 412 if aw_path: 413 self.aw_path = aw_path 414 self.log.debug("aw path is: {}".format(self.aw_path)) 415 416 def set_testcase_path(self, testcase_path): 417 self.testcase_path = testcase_path 418 self.log.debug("testcase path is: {}".format(self.testcase_path)) 419 420 def set_settings(self, settings): 421 if settings: 422 self.settings = settings 423 self.log.debug("settings is: {}".format(self.settings)) 424 425 def set_test_suite_path(self, test_suite_path): 426 if test_suite_path: 427 self.test_suite_path = test_suite_path 428 self.log.debug("test suite path is: {}".format(self.test_suite_path)) 429 430 def set_task_report_dir(self, task_report_dir): 431 if task_report_dir: 432 self.task_report_dir = task_report_dir 433 self.log.debug("task report dir: {}".format(self.task_report_dir)) 434 435 def set_resource_path(self, resource_path): 436 if resource_path: 437 self.resource_path = resource_path 438 self.log.debug("resource path is: {}".format(self.resource_path)) 439 440 def set_config_json(self, config_json): 441 self.config_json = config_json 442 self.log.debug("config json is: {}".format(self.config_json)) 443 444 def set_property_config(self, property_config): 445 self.property_config = property_config 446 self.log.debug("property config is: {}".format(self.property_config)) 447 448 def set_devicename(self, devicename): 449 if devicename: 450 self.devicename = devicename 451 self.log.debug("devicename is: {}".format(self.devicename)) 452 453 454class CurCheckCmd: 455 def __init__(self): 456 # 用例校验参数？ 457 self.through = "" 458 self.expect = "" 459 self.actual = "" 460 461 def get_cur_check_status(self): 462 if all([self.through, self.expect, self.actual]): 463 return True 464 return False 465 466 def get_cur_check_msg(self): 467 return "{}, expect:{}, actual:{}".format(self.through, self.expect, 468 self.actual) 469 470 471class DeccVariable: 472 __cur_case = {} 473 __thread_lock = threading.Lock() 474 project = ProjectVariables(log) 475 476 @classmethod 477 def set_project_obj(cls, project_obj): 478 log.info("init DeccVariable project object") 479 cls.project = project_obj 480 481 @classmethod 482 def set_cur_case_obj(cls, cur_case_obj): 483 log.info("init DeccVariable cur case object") 484 with cls.__thread_lock: 485 cls.__cur_case[cls.__cur_case_key()] = cur_case_obj 486 487 @classmethod 488 def cur_case(cls): 489 with cls.__thread_lock: 490 return cls.__cur_case.get(cls.__cur_case_key(), None) 491 492 @classmethod 493 def __cur_case_key(cls): 494 return threading.current_thread().ident 495 496 @classmethod 497 def reset(cls): 498 log.info("reset DeccVariable") 499 with cls.__thread_lock: 500 key = cls.__cur_case_key() 501 if key in cls.__cur_case: 502 cls.__cur_case.pop(key) 503