Lines Matching refs:self
65 def __init__(self):
66 self.run_mode = RunnerMode.PIPELINE
67 self.run_list = None
68 self.no_run_list = None
69 self.running = None
70 self.configs = None
71 self.devices = None
72 self.log = Log
73 self.start_time = None
74 self.test_results = None
75 self.upload_result_handler = None
76 self.project = None
77 self.prepare = None
78 self.cur_case = None
79 self._repeat = 1
80 self._repeat_round = 1
82 def init_pipeline_runner(self, run_list, configs, devices, upload_result_handler):
83 self.run_list = run_list
84 self.no_run_list = copy.copy(self.run_list)
85 self.running = False
86 self.configs = configs
87 self.devices = devices
88 self.start_time = get_cst_time()
89 self.test_results = []
90 self.upload_result_handler = upload_result_handler
91 self.project = ProjectVariables(self.log)
92 self.prepare = None
93 self.__init_project_variables()
94 self.run_mode = RunnerMode.PIPELINE
95 self._repeat = self.configs.get("request").config.repeat
96 self._repeat_round = self.configs.get("request").get_repeat_round()
98 def init_case_runner(self, run_list: Union[BaseCase, list]):
100 self.run_list = run_list
101 self.run_mode = RunnerMode.DEBUG
102 self.log = platform_logger("TestRunner")
104 def __init_project_variables(self):
109 self.log.debug("configs:{}".format(self.configs))
110 testcases_path = self.configs.get('testcases_path', "")
111 testargs = self.configs.get("testargs", {})
112 self.__flash_run_list(testargs)
114 self.cur_case = CurCase(self.log)
115 self.project.set_project_path()
116 self.project.set_testcase_path(testcases_path)
117 self.project.set_task_report_dir(self.configs.get("report_path"))
118 self.project.set_resource_path(self.get_local_resource_path())
120 def get_local_resource_path(self):
122 self.project.project_path, "testcases", "DeviceTest", "resource")
125 def get_local_aw_path(self):
127 self.project.project_path, "testcases", "DeviceTest", "aw")
130 def __flash_run_list(self, testargs):
137 self.log.info("get test:{}".format(get_test))
138 retry_test_list = self.parse_retry_test_list(get_test)
140 self.run_list = retry_test_list
141 self.no_run_list = copy.copy(self.run_list)
142 self.log.info("retry test list:{}".format(retry_test_list))
144 def parse_retry_test_list(self, retry_test_list):
149 self.log.error(err_msg)
158 for abd_file_path in self.run_list:
164 self.log.debug("history case list:{}".format(history_case_list))
172 def parse_config(self, test_configs):
175 def add_value_to_configs(self):
176 self.configs["log"] = self.log
177 self.configs["devices"] = self.devices
178 self.configs["project"] = self.project
180 def run(self):
181 self._pipeline_run()
183 def _pipeline_run(self):
184 self.running = True
185 aw_path = self.add_aw_path_to_sys(self.project.aw_path)
186 self.log.info("Executing run list {}.".format(self.run_list))
188 self.add_value_to_configs()
190 self.prepare = PrepareHandler(self.log, self.cur_case,
191 self.project, self.configs,
192 self.devices, self.run_list)
195 for device in self.devices:
198 self.log.debug(DeviceRoot.is_root_device)
202 self.log.error('set branch api error.')
204 self.prepare.run_prepare()
206 for test_cls_name in self.run_list:
208 if self.project.record.is_shutdown(raise_exception=False):
210 self.log.info("Executing test class {}".format(test_cls_name))
211 self.project.execute_case_name = case_name
212 self.run_test_class(test_cls_name, case_name)
213 self.prepare.run_prepare(is_teardown=True)
217 def add_aw_path_to_sys(self, aw_path):
222 self.log.info("add {} to sys path.".format(sys_aw_path))
226 def run_test_class(self, case_path, case_name):
249 self.project.cur_case_full_path = case_path
250 DeccVariable.set_cur_case_obj(self.cur_case)
252 self.log.info("Success to import {}.".format(case_name))
253 with test_cls(self.configs) as test_cls_instance:
254 self.cur_case.set_case_instance(test_cls_instance)
265 self.log.error(error_msg)
266 self.log.error(traceback.format_exc())
269 self.log.error(error_msg)
270 self.log.error(traceback.format_exc())
274 self.log.debug("del test_cls_instance success.")
276 self.log.warning("del test_cls_instance exception. {}".format(exception))
281 environment = self.configs.get("request").config.environment
282 steps = self.cur_case.get_steps_info()
300 round_folder = f"round{self._repeat_round}" if self._repeat > 1 else ""
302 to_file = os.path.join(self.project.task_report_dir, report_path)
306 self.cur_case.set_case_instance(None)
307 self.record_current_case_result(
312 def record_current_case_result(self, case_name, tests, case_result,
314 test_result = self.record_cls_result(
316 self.log.debug("test result: {}".format(test_result))
317 self.test_results.append(test_result)
318 self.upload_result_handler.report_handler.test_results.append(test_result)
320 def stop(self):
325 if self.running:
326 self.running = False
348 def __init__(self, suite, configs, devices):
349 self.suite = suite
350 self.running = False
351 self.configs = configs
352 self.devices = devices
353 self.log = Log
354 self.start_time = get_cst_time()
355 self.listeners = self.configs["listeners"]
356 self.state_machine = StateRecorder()
357 self.suite_name = ""
359 def add_value_to_configs(self):
360 self.configs["log"] = self.log
361 self.configs["devices"] = self.devices
362 self.configs["suite_name"] = self.suite_name
364 def run(self):
365 self.running = True
366 self.log.info("Executing test suite: {}.".format(self.suite))
368 self.suite_name = get_base_name(self.suite)
369 self.add_value_to_configs()
370 self.run_test_suite(self.suite)
372 def run_test_suite(self, test_cls_name):
384 self.handle_suites_started()
385 self.handle_suite_started()
387 test_cls = import_from_file(suite_dir_path, self.suite_name)
388 self.log.info("Success to import {}.".format(self.suite_name))
389 self.configs["cur_suite"] = test_cls
390 with test_cls(self.configs, suite_dir_path) as test_cls_instance:
394 self.handle_suite_ended(test_cls_instance)
397 self.log.error(error_msg)
398 self.log.error(traceback.format_exc())
399 self.handle_suites_ended(error_msg)
400 result_path = os.path.join(self.configs["report_path"], "result")
401 report_file = os.path.join(result_path, "%s.xml" % self.suite_name)
407 self.log.debug("del test suite instance success.")
409 self.log.warning("del test suite instance exception. "
413 def stop(self):
418 if self.running:
419 self.running = False
421 def handle_suites_started(self):
422 self.state_machine.get_suites(reset=True)
423 test_suites = self.state_machine.get_suites()
424 test_suites.suites_name = self.suite_name
426 for listener in self.listeners:
430 def handle_suites_ended(self, error_msg):
431 suites = self.state_machine.get_suites()
434 for listener in self.listeners:
437 def handle_suite_started(self):
438 self.state_machine.suite(reset=True)
439 self.state_machine.running_test_index = 0
440 test_suite = self.state_machine.suite()
441 test_suite.suite_name = self.suite_name
443 for listener in self.listeners:
447 def handle_suite_ended(self, testsuite_cls):
448 suite = self.state_machine.suite()
449 suites = self.state_machine.get_suites()
450 self.handle_one_case_result(testsuite_cls)
454 for listener in self.listeners:
458 def handle_one_case_result(self, testsuite_cls):
470 test_result = self.state_machine.test(reset=True)
471 test_suite = self.state_machine.suite()
480 test_result.current = self.state_machine.running_test_index + 1
482 self.state_machine.suite().run_time += run_time
483 for listener in self.listeners:
486 test_suites = self.state_machine.get_suites()
488 for listener in self.listeners:
491 self.state_machine.running_test_index += 1