Lines Matching full:self
44 def __init__(self): argument
45 self.state = None
46 self.type = None
47 self.regex = []
48 self.matches = OrderedDict()
49 self.ordered = True
50 self.repeat = 1
51 self.id = None
52 self.fail_on_fault = True
53 self.fault = False
54 self.capture_coverage = False
55 self.next_pattern = 0
56 self.record = None
57 self.recording = []
58 self.fieldnames = []
59 self.ztest = False
60 self.detected_suite_names = []
61 self.run_id = None
62 self.matched_run_id = False
63 self.run_id_exists = False
64 self.instance: TestInstance | None = None
65 self.testcase_output = ""
66 self._match = False
68 def configure(self, instance): argument
69 self.instance = instance
71 self.id = instance.testsuite.id
72 self.run_id = instance.run_id
74 self.fail_on_fault = False
77 self.type = config.get('type', None)
78 self.regex = config.get('regex', [])
79 self.repeat = config.get('repeat', 1)
80 self.ordered = config.get('ordered', True)
81 self.record = config.get('record', {})
83 def process_test(self, line): argument
85 runid_match = re.search(self.run_id_pattern, line)
88 self.run_id_exists = True
89 if run_id == str(self.run_id):
90 self.matched_run_id = True
92 if self.RUN_PASSED in line:
93 if self.fault:
94 self.state = "failed"
96 self.state = "passed"
98 if self.RUN_FAILED in line:
99 self.state = "failed"
101 if self.fail_on_fault:
102 if self.FAULT == line:
103 self.fault = True
105 if self.GCOV_START in line:
106 self.capture_coverage = True
107 elif self.GCOV_END in line:
108 self.capture_coverage = False
114 def configure(self, instance): argument
115 super(Robot, self).configure(instance)
116 self.instance = instance
120 self.path = config.get('robot_test_path', None)
122 def handle(self, line): argument
128 self.instance.state = "passed"
129 tc = self.instance.get_case_or_create(self.id)
132 def run_robot_test(self, command, handler): argument
136 env["ROBOT_FILES"] = self.path
139 stderr=subprocess.STDOUT, cwd=self.instance.build_dir, env=env) as cmake_proc:
142 self.instance.execution_time = time.time() - start_time
145 self.instance.status = "passed"
149 self.instance.testcases[0].status = "passed"
152 (handler.sourcedir, self.instance.platform.name))
153 self.instance.status = "failed"
154 self.instance.testcases[0].status = "failed"
157 with open(os.path.join(self.instance.build_dir, handler.log), "wt") as log:
163 def configure(self, instance): argument
164 super(Console, self).configure(instance)
165 if self.type == "one_line":
166 self.pattern = re.compile(self.regex[0])
167 self.patterns_expected = 1
168 elif self.type == "multi_line":
169 self.patterns = []
170 for r in self.regex:
171 self.patterns.append(re.compile(r))
172 self.patterns_expected = len(self.patterns)
174 def handle(self, line): argument
175 if self.type == "one_line":
176 if self.pattern.search(line):
177 …logger.debug(f"HARNESS:{self.__class__.__name__}:EXPECTED({self.next_pattern}):'{self.pattern.patt…
178 self.next_pattern += 1
179 self.state = "passed"
180 elif self.type == "multi_line" and self.ordered:
181 if (self.next_pattern < len(self.patterns) and
182 self.patterns[self.next_pattern].search(line)):
183 …logger.debug(f"HARNESS:{self.__class__.__name__}:EXPECTED({self.next_pattern}):'{self.patterns[sel…
184 self.next_pattern += 1
185 if self.next_pattern >= len(self.patterns):
186 self.state = "passed"
187 elif self.type == "multi_line" and not self.ordered:
188 for i, pattern in enumerate(self.patterns):
189 r = self.regex[i]
190 if pattern.search(line) and not r in self.matches:
191 self.matches[r] = line
192 if len(self.matches) == len(self.regex):
193 self.state = "passed"
197 if self.fail_on_fault:
198 if self.FAULT in line:
199 self.fault = True
201 if self.GCOV_START in line:
202 self.capture_coverage = True
203 elif self.GCOV_END in line:
204 self.capture_coverage = False
207 if self.record:
208 pattern = re.compile(self.record.get("regex", ""))
212 if not self.fieldnames:
214 self.fieldnames.append(k)
218 self.recording.append(csv)
220 self.process_test(line)
229 if self.state == "passed" and self.ordered and self.next_pattern < self.patterns_expected:
230 …logger.error(f"HARNESS:{self.__class__.__name__}: failed with only {self.next_pattern} matched pat…
231 self.state = "failed"
233 tc = self.instance.get_case_or_create(self.id)
234 if self.state == "passed":
246 def configure(self, instance: TestInstance): argument
247 super(Pytest, self).configure(instance)
248 self.running_dir = instance.build_dir
249 self.source_dir = instance.testsuite.source_dir
250 self.report_file = os.path.join(self.running_dir, 'report.xml')
251 self.pytest_log_file_path = os.path.join(self.running_dir, 'twister_harness.log')
252 self.reserved_serial = None
254 def pytest_run(self, timeout): argument
256 cmd = self.generate_command()
257 self.run_command(cmd, timeout)
260 self.state = 'failed'
261 self.instance.reason = str(pytest_exception)
263 if self.reserved_serial:
264 self.instance.handler.make_device_available(self.reserved_serial)
265 self._update_test_status()
267 def generate_command(self): argument
268 config = self.instance.testsuite.harness_config
275 f'--build-dir={self.running_dir}',
276 f'--junit-xml={self.report_file}',
279 f'--log-file={self.pytest_log_file_path}'
282 self.source_dir, os.path.expanduser(os.path.expandvars(src)))) for src in pytest_root])
285 handler: Handler = self.instance.handler
295 self._generate_parameters_for_hardware(handler)
305 def _generate_parameters_for_hardware(self, handler: Handler): argument
311 self.reserved_serial = hardware.serial_pty or hardware.serial
344 def run_command(self, cmd, timeout): argument
345 cmd, env = self._update_command_with_env_dependencies(cmd)
353 reader_t = threading.Thread(target=self._output_reader, args=(proc,), daemon=True)
360 self.instance.reason = 'Pytest timeout'
361 self.state = 'failed'
364 self.state = 'failed'
398 def _update_test_status(self): argument
399 if not self.state:
400 self.instance.testcases = []
402 self._parse_report_file(self.report_file)
404 logger.error(f'Error when parsing file {self.report_file}: {e}')
405 self.state = 'failed'
407 if not self.instance.testcases:
408 self.instance.init_cases()
410 self.instance.status = self.state or 'failed'
411 if self.instance.status in ['error', 'failed']:
412 self.instance.reason = self.instance.reason or 'Pytest failed'
413 self.instance.add_missing_case_status('blocked', self.instance.reason)
415 def _parse_report_file(self, report): argument
420 self.state = 'failed'
422 self.state = 'error'
424 self.state = 'skipped'
426 self.state = 'passed'
427 self.instance.execution_time = float(elem_ts.get('time'))
430 tc = self.instance.get_case_or_create(f"{self.id}.{elem_tc.get('name')}")
455 def handle(self, line): argument
457 non_ansi_line = self.ANSI_ESCAPE.sub('', line)
460 test_start_match = re.search(self.TEST_START_PATTERN, non_ansi_line)
464 if suite_name not in self.detected_suite_names:
465 self.detected_suite_names.append(suite_name)
468 name = "{}.{}.{}".format(self.id, suite_name, test_start_match.group("test_name"))
472 self.tc is None
473 ), "gTest error, {} didn't finish".format(self.tc)
476 tc = self.instance.get_case_by_name(name)
480 tc = self.instance.get_case_or_create(name)
481 self.tc = tc
482 self.tc.status = "started"
483 self.testcase_output += line + "\n"
484 self._match = True
487 finished_match = re.search(self.FINISHED_PATTERN, non_ansi_line)
489 tc = self.instance.get_case_or_create(self.id)
490 if self.has_failures or self.tc is not None:
491 self.state = "failed"
494 self.state = "passed"
499 state, name = self._check_result(non_ansi_line)
505 tc = self.instance.get_case_by_name(name)
507 tc is not None and tc == self.tc
508 ), "gTest error, mismatched tests. Expected {} but got {}".format(self.tc, tc)
511 self.tc = None
516 self.has_failures = True
517 tc.output = self.testcase_output
518 self.testcase_output = ""
519 self._match = False
521 def _check_result(self, line): argument
522 test_pass_match = re.search(self.TEST_PASS_PATTERN, line)
524 …return "passed", "{}.{}.{}".format(self.id, test_pass_match.group("suite_name"), test_pass_match.g…
525 test_fail_match = re.search(self.TEST_FAIL_PATTERN, line)
527 …return "failed", "{}.{}.{}".format(self.id, test_fail_match.group("suite_name"), test_fail_match.g…
537 def handle(self, line): argument
538 test_suite_match = re.search(self.test_suite_start_pattern, line)
541 self.detected_suite_names.append(suite_name)
543 testcase_match = re.search(self.ZTEST_START_PATTERN, line)
545 name = "{}.{}".format(self.id, testcase_match.group(2))
546 tc = self.instance.get_case_or_create(name)
552 if testcase_match or self._match:
553 self.testcase_output += line + "\n"
554 self._match = True
560 name = "{}.{}".format(self.id, result_match.group(3))
561 tc = self.instance.get_case_or_create(name)
562 tc.status = self.ztest_to_status[matched_status]
567 tc.output = self.testcase_output
568 self.testcase_output = ""
569 self._match = False
570 self.ztest = True
572 self.process_test(line)
574 if not self.ztest and self.state:
575 logger.debug(f"not a ztest and no state for {self.id}")
576 tc = self.instance.get_case_or_create(self.id)
577 if self.state == "passed":