Lines Matching full:self
47 def __init__(self): argument
48 self.state = None
49 self.reason = None
50 self.type = None
51 self.regex = []
52 self.matches = OrderedDict()
53 self.ordered = True
54 self.id = None
55 self.fail_on_fault = True
56 self.fault = False
57 self.capture_coverage = False
58 self.next_pattern = 0
59 self.record = None
60 self.record_pattern = None
61 self.record_as_json = None
62 self.recording = []
63 self.ztest = False
64 self.detected_suite_names = []
65 self.run_id = None
66 self.matched_run_id = False
67 self.run_id_exists = False
68 self.instance: TestInstance | None = None
69 self.testcase_output = ""
70 self._match = False
72 def configure(self, instance): argument
73 self.instance = instance
75 self.id = instance.testsuite.id
76 self.run_id = instance.run_id
78 self.fail_on_fault = False
81 self.type = config.get('type', None)
82 self.regex = config.get('regex', [])
83 self.ordered = config.get('ordered', True)
84 self.record = config.get('record', {})
85 if self.record:
86 self.record_pattern = re.compile(self.record.get("regex", ""))
87 self.record_as_json = self.record.get("as_json")
89 def build(self): argument
92 def get_testcase_name(self): argument
96 return self.id
98 def translate_record(self, record: dict) -> dict: argument
99 if self.record_as_json:
100 for k in self.record_as_json:
106 logger.warning(f"HARNESS:{self.__class__.__name__}: recording JSON failed:"
112 def parse_record(self, line) -> re.Match: argument
114 if self.record_pattern:
115 match = self.record_pattern.search(line)
117 … rec = self.translate_record({ k:v.strip() for k,v in match.groupdict(default="").items() })
118 self.recording.append(rec)
122 def process_test(self, line): argument
124 self.parse_record(line)
126 runid_match = re.search(self.run_id_pattern, line)
129 self.run_id_exists = True
130 if run_id == str(self.run_id):
131 self.matched_run_id = True
133 if self.RUN_PASSED in line:
134 if self.fault:
135 self.state = "failed"
136 self.reason = "Fault detected while running test"
138 self.state = "passed"
140 if self.RUN_FAILED in line:
141 self.state = "failed"
142 self.reason = "Testsuite failed"
144 if self.fail_on_fault:
145 if self.FAULT == line:
146 self.fault = True
148 if self.GCOV_START in line:
149 self.capture_coverage = True
150 elif self.GCOV_END in line:
151 self.capture_coverage = False
157 def configure(self, instance): argument
158 super(Robot, self).configure(instance)
159 self.instance = instance
163 self.path = config.get('robot_testsuite', None)
164 self.option = config.get('robot_option', None)
166 def handle(self, line): argument
172 self.instance.state = "passed"
173 tc = self.instance.get_case_or_create(self.id)
176 def run_robot_test(self, command, handler): argument
180 if self.option:
181 if isinstance(self.option, list):
182 for option in self.option:
186 for v in str(self.option).split():
189 if self.path is None:
192 if isinstance(self.path, list):
193 for suite in self.path:
196 command.append(os.path.join(handler.sourcedir, self.path))
199 stderr=subprocess.STDOUT, cwd=self.instance.build_dir, env=env) as renode_test_proc:
202 self.instance.execution_time = time.time() - start_time
205 self.instance.status = "passed"
209 self.instance.testcases[0].status = "passed"
212 (handler.sourcedir, self.instance.platform.name))
213 self.instance.status = "failed"
214 self.instance.testcases[0].status = "failed"
217 with open(os.path.join(self.instance.build_dir, handler.log), "wt") as log:
223 def get_testcase_name(self): argument
232 if self.instance and len(self.instance.testcases) == 1:
233 return self.instance.testcases[0].name
234 return super(Console, self).get_testcase_name()
236 def configure(self, instance): argument
237 super(Console, self).configure(instance)
238 if self.regex is None or len(self.regex) == 0:
239 self.state = "failed"
240 tc = self.instance.set_case_status_by_name(
241 self.get_testcase_name(),
243 f"HARNESS:{self.__class__.__name__}:no regex patterns configured."
245 raise ConfigurationError(self.instance.name, tc.reason)
246 if self.type == "one_line":
247 self.pattern = re.compile(self.regex[0])
248 self.patterns_expected = 1
249 elif self.type == "multi_line":
250 self.patterns = []
251 for r in self.regex:
252 self.patterns.append(re.compile(r))
253 self.patterns_expected = len(self.patterns)
255 self.state = "failed"
256 tc = self.instance.set_case_status_by_name(
257 self.get_testcase_name(),
259 f"HARNESS:{self.__class__.__name__}:incorrect type={self.type}"
261 raise ConfigurationError(self.instance.name, tc.reason)
264 def handle(self, line): argument
265 if self.type == "one_line":
266 if self.pattern.search(line):
267 logger.debug(f"HARNESS:{self.__class__.__name__}:EXPECTED:"
268 f"'{self.pattern.pattern}'")
269 self.next_pattern += 1
270 self.state = "passed"
271 elif self.type == "multi_line" and self.ordered:
272 if (self.next_pattern < len(self.patterns) and
273 self.patterns[self.next_pattern].search(line)):
274 logger.debug(f"HARNESS:{self.__class__.__name__}:EXPECTED("
275 f"{self.next_pattern + 1}/{self.patterns_expected}):"
276 f"'{self.patterns[self.next_pattern].pattern}'")
277 self.next_pattern += 1
278 if self.next_pattern >= len(self.patterns):
279 self.state = "passed"
280 elif self.type == "multi_line" and not self.ordered:
281 for i, pattern in enumerate(self.patterns):
282 r = self.regex[i]
283 if pattern.search(line) and not r in self.matches:
284 self.matches[r] = line
285 logger.debug(f"HARNESS:{self.__class__.__name__}:EXPECTED("
286 f"{len(self.matches)}/{self.patterns_expected}):"
288 if len(self.matches) == len(self.regex):
289 self.state = "passed"
293 if self.fail_on_fault:
294 if self.FAULT in line:
295 self.fault = True
297 if self.GCOV_START in line:
298 self.capture_coverage = True
299 elif self.GCOV_END in line:
300 self.capture_coverage = False
302 self.process_test(line)
310 if self.state == "passed" and self.ordered and self.next_pattern < self.patterns_expected:
311 logger.error(f"HARNESS:{self.__class__.__name__}: failed with"
312 f" {self.next_pattern} of {self.patterns_expected}"
314 self.state = "failed"
315 self.reason = "patterns did not match (ordered)"
316 … if self.state == "passed" and not self.ordered and len(self.matches) < self.patterns_expected:
317 logger.error(f"HARNESS:{self.__class__.__name__}: failed with"
318 f" {len(self.matches)} of {self.patterns_expected}"
320 self.state = "failed"
321 self.reason = "patterns did not match (unordered)"
323 tc = self.instance.get_case_or_create(self.get_testcase_name())
324 if self.state == "passed":
336 def configure(self, instance: TestInstance): argument
337 super(Pytest, self).configure(instance)
338 self.running_dir = instance.build_dir
339 self.source_dir = instance.testsuite.source_dir
340 self.report_file = os.path.join(self.running_dir, 'report.xml')
341 self.pytest_log_file_path = os.path.join(self.running_dir, 'twister_harness.log')
342 self.reserved_serial = None
344 def pytest_run(self, timeout): argument
346 cmd = self.generate_command()
347 self.run_command(cmd, timeout)
350 self.state = 'failed'
351 self.instance.reason = str(pytest_exception)
353 if self.reserved_serial:
354 self.instance.handler.make_device_available(self.reserved_serial)
355 self.instance.record(self.recording)
356 self._update_test_status()
358 def generate_command(self): argument
359 config = self.instance.testsuite.harness_config
360 handler: Handler = self.instance.handler
368 f'--build-dir={self.running_dir}',
369 f'--junit-xml={self.report_file}',
372 f'--log-file={self.pytest_log_file_path}'
375 self.source_dir, os.path.expanduser(os.path.expandvars(src)))) for src in pytest_root])
392 self._generate_parameters_for_hardware(handler)
416 def _generate_parameters_for_hardware(self, handler: Handler): argument
422 self.instance.dut = hardware.id
424 self.reserved_serial = hardware.serial_pty or hardware.serial
467 def run_command(self, cmd, timeout): argument
468 cmd, env = self._update_command_with_env_dependencies(cmd)
476 … reader_t = threading.Thread(target=self._output_reader, args=(proc, self), daemon=True)
483 self.instance.reason = 'Pytest timeout'
484 self.state = 'failed'
487 self.state = 'failed'
522 def _update_test_status(self): argument
523 if not self.state:
524 self.instance.testcases = []
526 self._parse_report_file(self.report_file)
528 logger.error(f'Error when parsing file {self.report_file}: {e}')
529 self.state = 'failed'
531 if not self.instance.testcases:
532 self.instance.init_cases()
534 self.instance.status = self.state or 'failed'
535 if self.instance.status in ['error', 'failed']:
536 self.instance.reason = self.instance.reason or 'Pytest failed'
537 self.instance.add_missing_case_status('blocked', self.instance.reason)
539 def _parse_report_file(self, report): argument
544 self.state = 'failed'
545 …self.instance.reason = f"{elem_ts.get('failures')}/{elem_ts.get('tests')} pytest scenario(s) faile…
547 self.state = 'error'
548 self.instance.reason = 'Error during pytest execution'
550 self.state = 'skipped'
552 self.state = 'passed'
553 self.instance.execution_time = float(elem_ts.get('time'))
556 tc = self.instance.add_testcase(f"{self.id}.{elem_tc.get('name')}")
571 self.state = 'skipped'
572 self.instance.reason = 'No tests collected'
583 def __init__(self): argument
585 self.tc = None
586 self.has_failures = False
588 def handle(self, line): argument
590 non_ansi_line = self.ANSI_ESCAPE.sub('', line)
592 if self.state:
596 test_start_match = re.search(self.TEST_START_PATTERN, non_ansi_line)
600 if suite_name not in self.detected_suite_names:
601 self.detected_suite_names.append(suite_name)
604 name = "{}.{}.{}".format(self.id, suite_name, test_start_match.group("test_name"))
608 self.tc is None
609 ), "gTest error, {} didn't finish".format(self.tc)
612 tc = self.instance.get_case_by_name(name)
616 tc = self.instance.get_case_or_create(name)
617 self.tc = tc
618 self.tc.status = "started"
619 self.testcase_output += line + "\n"
620 self._match = True
623 finished_match = re.search(self.FINISHED_PATTERN, non_ansi_line)
625 tc = self.instance.get_case_or_create(self.id)
626 if self.has_failures or self.tc is not None:
627 self.state = "failed"
630 self.state = "passed"
635 state, name = self._check_result(non_ansi_line)
641 tc = self.instance.get_case_by_name(name)
643 tc is not None and tc == self.tc
644 ), "gTest error, mismatched tests. Expected {} but got {}".format(self.tc, tc)
647 self.tc = None
652 self.has_failures = True
653 tc.output = self.testcase_output
654 self.testcase_output = ""
655 self._match = False
657 def _check_result(self, line): argument
658 test_pass_match = re.search(self.TEST_PASS_PATTERN, line)
660 …return "passed", "{}.{}.{}".format(self.id, test_pass_match.group("suite_name"), test_pass_match.g…
661 test_skip_match = re.search(self.TEST_SKIP_PATTERN, line)
663 …return "skipped", "{}.{}.{}".format(self.id, test_skip_match.group("suite_name"), test_skip_match.…
664 test_fail_match = re.search(self.TEST_FAIL_PATTERN, line)
666 …return "failed", "{}.{}.{}".format(self.id, test_fail_match.group("suite_name"), test_fail_match.g…
676 def handle(self, line): argument
677 test_suite_match = re.search(self.test_suite_start_pattern, line)
680 self.detected_suite_names.append(suite_name)
682 testcase_match = re.search(self.ZTEST_START_PATTERN, line)
684 name = "{}.{}".format(self.id, testcase_match.group(2))
685 tc = self.instance.get_case_or_create(name)
691 if testcase_match or self._match:
692 self.testcase_output += line + "\n"
693 self._match = True
705 name = "{}.{}".format(self.id, result_match.group(3))
706 tc = self.instance.get_case_or_create(name)
707 tc.status = self.ztest_to_status[matched_status]
712 tc.output = self.testcase_output
713 self.testcase_output = ""
714 self._match = False
715 self.ztest = True
718 self.detected_suite_names.append(summary_match.group(2))
719 name = "{}.{}".format(self.id, summary_match.group(4))
720 tc = self.instance.get_case_or_create(name)
721 tc.status = self.ztest_to_status[matched_status]
726 tc.output = self.testcase_output
727 self.testcase_output = ""
728 self._match = False
729 self.ztest = True
731 self.process_test(line)
733 if not self.ztest and self.state:
734 logger.debug(f"not a ztest and no state for {self.id}")
735 tc = self.instance.get_case_or_create(self.id)
736 if self.state == "passed":
749 def build(self): argument
755 if self.instance is None:
758 original_exe_path: str = os.path.join(self.instance.build_dir, 'zephyr', 'zephyr.exe')
768 new_exe_name: str = self.instance.testsuite.harness_config.get('bsim_exe_name', '')
770 new_exe_name = f'bs_{self.instance.platform.name}_{new_exe_name}'
772 new_exe_name = self.instance.name