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 copy 20import os 21import sys 22import traceback 23 24from xdevice import StateRecorder 25from xdevice import LifeCycle 26from xdevice import ResultCode 27from xdevice import get_cst_time 28from xdevice import Scheduler 29from xdevice import LifeStage 30 31from devicetest.runner.prepare import PrepareHandler 32from devicetest.core.error_message import ErrorMessage 33from devicetest.core.constants import RunResult 34from devicetest.utils.util import calculate_execution_time 35from devicetest.utils.util import clean_sys_resource 36from devicetest.utils.util import get_base_name 37from devicetest.utils.util import get_dir_path 38from devicetest.utils.util import import_from_file 39from devicetest.core.variables import DeccVariable 40from devicetest.core.variables import ProjectVariables 41from devicetest.core.variables import CurCase 42from devicetest.core.exception import DeviceTestError 43from devicetest.core.test_case import DeviceRoot 44from devicetest.report.generation import add_log_caching_handler 45from devicetest.report.generation import del_log_caching_handler 46from devicetest.report.generation import get_caching_logs 47from devicetest.report.generation import generate_report 48 49 50class TestRunner: 51 """executes test cases and 52 """ 53 54 def __init__(self): 55 self.run_list = None 56 self.no_run_list = None 57 self.running = None 58 self.configs = None 59 self.devices = None 60 self.log = None 61 self.start_time = None 62 self.test_results = None 63 self.upload_result_handler = None 64 self.project = None 65 self.prepare = None 66 self.cur_case = None 67 68 def init_pipeline_runner(self, run_list, configs, devices, log, upload_result_handler): 69 self.run_list = run_list 70 self.no_run_list = copy.copy(self.run_list) 71 self.running = False 72 self.configs = configs 73 self.devices = devices 74 self.log = log 75 self.start_time = get_cst_time() 76 self.test_results = [] 77 self.upload_result_handler = upload_result_handler 78 self.project = ProjectVariables(self.log) 79 self.prepare = None 80 self.__init_project_variables() 81 82 def __init_project_variables(self): 83 """ 84 testargs：为xDevice透传过来的数据,用户调用CONFIG可获取 85 :return: 86 """ 87 self.log.debug("configs:{}".format(self.configs)) 88 testcases_path = self.configs.get('testcases_path', "") 89 testargs = self.configs.get("testargs", {}) 90 self.__flash_run_list(testargs) 91 92 self.cur_case = CurCase(self.log) 93 self.project.set_project_path() 94 self.project.set_testcase_path(testcases_path) 95 self.project.set_task_report_dir( 96 self.get_report_dir_path(self.configs)) 97 self.project.set_resource_path(self.get_local_resource_path()) 98 99 def get_local_resource_path(self): 100 local_resource_path = os.path.join( 101 self.project.project_path, "testcases", "DeviceTest", "resource") 102 return local_resource_path 103 104 def get_report_dir_path(self, configs): 105 report_path = configs.get("report_path") 106 if report_path and isinstance(report_path, str): 107 report_dir_path = report_path.split(os.sep.join(["", "temp", "task"]))[0] 108 if os.path.exists(report_dir_path): 109 return report_dir_path 110 self.log.debug("report dir path:{}".format(report_dir_path)) 111 self.log.debug("report path:{}".format(report_path)) 112 self.log.error(ErrorMessage.Error_01433.Message.en, 113 Error_no=ErrorMessage.Error_01433.Code) 114 raise DeviceTestError(ErrorMessage.Error_01433.Topic) 115 116 def get_local_aw_path(self): 117 local_aw_path = os.path.join( 118 self.project.project_path, "testcases", "DeviceTest", "aw") 119 return local_aw_path 120 121 def __flash_run_list(self, testargs): 122 """ 123 retry 场景更新run list 124 :param testargs: 125 :return: 126 """ 127 get_test = testargs.get("test") 128 self.log.info("get test:{}".format(get_test)) 129 retry_test_list = self.parse_retry_test_list(get_test) 130 if retry_test_list is not None: 131 self.run_list = retry_test_list 132 self.no_run_list = copy.copy(self.run_list) 133 self.log.info("retry test list:{}".format(retry_test_list)) 134 135 def parse_retry_test_list(self, retry_test_list): 136 if retry_test_list is None: 137 return None 138 elif not isinstance(retry_test_list, list): 139 self.log.error(ErrorMessage.Error_01430.Message.en, 140 error_no=ErrorMessage.Error_01430.Code) 141 raise DeviceTestError(ErrorMessage.Error_01430.Topic) 142 143 elif len(retry_test_list) == 1 and "#" not in str(retry_test_list[0]): 144 return None 145 else: 146 history_case_list = [] 147 history_case_dict = dict() 148 retry_case_list = [] 149 for abd_file_path in self.run_list: 150 base_file_name = get_base_name(abd_file_path) 151 if base_file_name not in history_case_dict.keys(): 152 history_case_dict.update({base_file_name: []}) 153 history_case_dict.get(base_file_name).append(abd_file_path) 154 history_case_list.append(base_file_name) 155 self.log.debug("history case list:{}".format(history_case_list)) 156 157 for _value in retry_test_list: 158 case_id = str(_value).split("#")[0] 159 if case_id in history_case_dict.keys(): 160 retry_case_list.append(history_case_dict.get(case_id)[0]) 161 return retry_case_list 162 163 def parse_config(self, test_configs): 164 pass 165 166 def add_value_to_configs(self): 167 self.configs["log"] = self.log 168 self.configs["devices"] = self.devices 169 self.configs["project"] = self.project 170 171 def run(self): 172 self._pipeline_run() 173 174 def _pipeline_run(self): 175 self.running = True 176 aw_path = self.add_aw_path_to_sys(self.project.aw_path) 177 self.log.info("Executing run list {}.".format(self.run_list)) 178 179 self.add_value_to_configs() 180 181 self.prepare = PrepareHandler(self.log, self.cur_case, 182 self.project, self.configs, 183 self.devices, self.run_list) 184 # **********混合root和非root************** 185 try: 186 for device in self.devices: 187 if hasattr(device, "is_hw_root"): 188 DeviceRoot.is_root_device = device.is_hw_root 189 self.log.debug(DeviceRoot.is_root_device) 190 setattr(device, "is_device_root", DeviceRoot.is_root_device) 191 192 except Exception as e: 193 self.log.error(f'set branch api error. {e}') 194 # **************混合root和非root end********************** 195 self.prepare.run_prepare() 196 197 for test_cls_name in self.run_list: 198 case_name = get_base_name(test_cls_name) 199 if self.project.record.is_shutdown(raise_exception=False): 200 break 201 self.log.info("Executing test class {}".format(test_cls_name)) 202 self.project.execute_case_name = case_name 203 self.run_test_class(test_cls_name, case_name) 204 self.prepare.run_prepare(is_teardown=True) 205 clean_sys_resource(file_path=aw_path) 206 DeccVariable.reset() 207 208 def add_aw_path_to_sys(self, aw_path): 209 210 sys_aw_path = os.path.dirname(aw_path) 211 if os.path.exists(sys_aw_path): 212 sys.path.insert(1, sys_aw_path) 213 self.log.info("add {} to sys path.".format(sys_aw_path)) 214 return sys_aw_path 215 return None 216 217 def run_test_class(self, test_cls_name, case_name, time_out=0): 218 """Instantiates and executes a test class. 219 If the test cases list is not None, all the test cases in the test 220 class should be executed. 221 Args: 222 test_cls_name: Name of the test class to execute. 223 Returns: 224 A tuple, with the number of cases passed at index 0, and the total 225 number of test cases at index 1. 226 """ 227 # 开始收集日志 228 case_log_buffer_hdl = add_log_caching_handler() 229 230 tests = "__init__" 231 case_result = RunResult.FAILED 232 start_time = get_cst_time() 233 case_dir_path = get_dir_path(test_cls_name) 234 test_cls_instance = None 235 try: 236 self.project.cur_case_full_path = test_cls_name 237 DeccVariable.set_cur_case_obj(self.cur_case) 238 test_cls = import_from_file(case_dir_path, case_name) 239 self.log.info("Success to import {}.".format(case_name)) 240 with test_cls(self.configs) as test_cls_instance: 241 self.cur_case.set_case_instance(test_cls_instance) 242 test_cls_instance.run() 243 244 tests = test_cls_instance.tests 245 start_time = test_cls_instance.start_time 246 247 case_result = test_cls_instance.result 248 error_msg = test_cls_instance.error_msg 249 250 except Exception as exception: 251 error_msg = "{}: {}".format(ErrorMessage.Error_01100.Topic, exception) 252 self.log.error(ErrorMessage.Error_01100.Message.en, error_no=ErrorMessage.Error_01100.Code) 253 self.log.error(error_msg) 254 self.log.error(traceback.format_exc()) 255 if test_cls_instance: 256 try: 257 del test_cls_instance 258 self.log.debug("del test_cls_instance success.") 259 except Exception as exception: 260 self.log.warning("del test_cls_instance exception. {}".format(exception)) 261 262 Scheduler.call_life_stage_action(stage=LifeStage.case_end, 263 case_name=case_name, 264 case_result=case_result, error_msg=error_msg) 265 266 end_time = get_cst_time() 267 extra_head = self.cur_case.get_steps_extra_head() 268 steps = self.cur_case.get_steps_info() 269 # 停止收集日志 270 del_log_caching_handler(case_log_buffer_hdl) 271 # 生成报告 272 logs = get_caching_logs(case_log_buffer_hdl) 273 case_info = { 274 "name": case_name, 275 "result": case_result, 276 "begin": start_time.strftime("%Y-%m-%d %H:%M:%S"), 277 "end": end_time.strftime("%Y-%m-%d %H:%M:%S"), 278 'elapsed': calculate_execution_time(start_time, end_time), 279 "error": error_msg, 280 "logs": logs, 281 "extra_head": extra_head, 282 "steps": steps 283 } 284 report_path = os.path.join("details", case_name + ".html") 285 to_path = os.path.join(self.project.task_report_dir, report_path) 286 generate_report(to_path, case=case_info) 287 steps.clear() 288 del case_log_buffer_hdl 289 self.cur_case.set_case_instance(None) 290 self.record_current_case_result( 291 case_name, tests, case_result, start_time, error_msg, report_path) 292 return case_result, error_msg 293 294 def record_current_case_result(self, case_name, tests, case_result, 295 start_time, error_msg, report): 296 test_result = self.record_cls_result( 297 case_name, tests, case_result, start_time, error_msg, report) 298 self.log.debug("test result: {}".format(test_result)) 299 self.test_results.append(test_result) 300 self.upload_result_handler.report_handler.test_results.append(test_result) 301 302 def stop(self): 303 """ 304 Releases resources from test run. Should be called right after run() 305 finishes. 306 """ 307 if self.running: 308 self.running = False 309 310 @staticmethod 311 def record_cls_result(case_name=None, tests_step=None, result=None, 312 start_time=None, error="", report=""): 313 dict_result = { 314 "case_name": case_name, 315 "tests_step": tests_step or "__init__", 316 "result": result or RunResult.FAILED, 317 "start_time": start_time or get_cst_time(), 318 "error": error, 319 "end_time": get_cst_time(), 320 "report": report 321 } 322 return dict_result 323 324 325class TestSuiteRunner: 326 """ 327 executes test suite cases 328 """ 329 330 def __init__(self, suite, configs, devices, log): 331 self.suite = suite 332 self.running = False 333 self.configs = configs 334 self.devices = devices 335 self.log = log 336 self.start_time = get_cst_time() 337 self.listeners = self.configs["listeners"] 338 self.state_machine = StateRecorder() 339 self.suite_name = "" 340 341 def add_value_to_configs(self): 342 self.configs["log"] = self.log 343 self.configs["devices"] = self.devices 344 self.configs["suite_name"] = self.suite_name 345 346 def run(self): 347 self.running = True 348 self.log.info("Executing test suite: {}.".format(self.suite)) 349 350 self.suite_name = get_base_name(self.suite) 351 self.add_value_to_configs() 352 self.run_test_suite(self.suite) 353 354 def run_test_suite(self, test_cls_name): 355 """Instantiates and executes a test class. 356 If the test cases list is not None, all the test cases in the test 357 class should be executed. 358 Args: 359 test_cls_name: Name of the test class to execute. 360 Returns: 361 A tuple, with the number of cases passed at index 0, and the total 362 number of test cases at index 1. 363 """ 364 error_msg = "Test resource loading error" 365 suite_dir_path = get_dir_path(test_cls_name) 366 test_cls_instance = None 367 try: 368 test_cls = import_from_file(suite_dir_path, self.suite_name) 369 self.handle_suites_started() 370 self.handle_suite_started() 371 self.log.info("Success to import {}.".format(self.suite_name)) 372 self.configs["cur_suite"] = test_cls 373 with test_cls(self.configs, suite_dir_path) as test_cls_instance: 374 test_cls_instance.run() 375 376 error_msg = test_cls_instance.error_msg 377 378 self.handle_suite_ended(test_cls_instance) 379 self.handle_suites_ended() 380 381 except Exception as e: 382 self.log.debug(traceback.format_exc()) 383 self.log.error("run suite error! Exception: {}".format(e)) 384 if test_cls_instance: 385 try: 386 del test_cls_instance 387 self.log.debug("del test suite instance success.") 388 except Exception as e: 389 self.log.warning("del test suite instance exception. " 390 "Exception: {}".format(e)) 391 return error_msg 392 393 def stop(self): 394 """ 395 Releases resources from test run. Should be called right after run() 396 finishes. 397 """ 398 if self.running: 399 self.running = False 400 401 def handle_suites_started(self): 402 self.state_machine.get_suites(reset=True) 403 test_suites = self.state_machine.get_suites() 404 test_suites.suites_name = self.suite_name 405 test_suites.test_num = 0 406 for listener in self.listeners: 407 suite_report = copy.copy(test_suites) 408 listener.__started__(LifeCycle.TestSuites, suite_report) 409 410 def handle_suites_ended(self): 411 suites = self.state_machine.get_suites() 412 suites.is_completed = True 413 for listener in self.listeners: 414 listener.__ended__(LifeCycle.TestSuites, test_result=suites, 415 suites_name=suites.suites_name) 416 417 def handle_suite_started(self): 418 self.state_machine.suite(reset=True) 419 self.state_machine.running_test_index = 0 420 test_suite = self.state_machine.suite() 421 test_suite.suite_name = self.suite_name 422 test_suite.test_num = 0 423 for listener in self.listeners: 424 suite_report = copy.copy(test_suite) 425 listener.__started__(LifeCycle.TestSuite, suite_report) 426 427 def handle_suite_ended(self, testsuite_cls): 428 suite = self.state_machine.suite() 429 suites = self.state_machine.get_suites() 430 self.handle_one_case_result(testsuite_cls) 431 suite.is_completed = True 432 # 设置测试套的报告路径 433 suite.report = testsuite_cls.suite_report_path 434 for listener in self.listeners: 435 listener.__ended__(LifeCycle.TestSuite, copy.copy(suite), is_clear=True) 436 suites.run_time += suite.run_time 437 438 def handle_one_case_result(self, testsuite_cls): 439 status_dict = {RunResult.PASSED: ResultCode.PASSED, 440 RunResult.FAILED: ResultCode.FAILED, 441 RunResult.BLOCKED: ResultCode.BLOCKED, 442 "ignore": ResultCode.SKIPPED, 443 RunResult.FAILED: ResultCode.FAILED} 444 for case_name, case_result in testsuite_cls.case_result.items(): 445 result = case_result.get("result") 446 error = case_result.get("error") 447 run_time = case_result.get("run_time") 448 report = case_result.get("report") 449 450 test_result = self.state_machine.test(reset=True) 451 test_suite = self.state_machine.suite() 452 test_result.test_class = test_suite.suite_name 453 test_result.test_name = case_name 454 test_result.code = status_dict.get(result).value 455 test_result.stacktrace = error 456 test_result.run_time = run_time 457 test_result.report = report 458 test_result.current = self.state_machine.running_test_index + 1 459 460 self.state_machine.suite().run_time += run_time 461 for listener in self.listeners: 462 listener.__started__( 463 LifeCycle.TestCase, copy.copy(test_result)) 464 test_suites = self.state_machine.get_suites() 465 test_suites.test_num += 1 466 for listener in self.listeners: 467 listener.__ended__( 468 LifeCycle.TestCase, copy.copy(test_result)) 469 self.state_machine.running_test_index += 1 470