Lines Matching refs:self
43 def __init__(self): argument
44 self._status = TwisterStatus.NONE
45 self.reason = None
46 self.type = None
47 self.regex = []
48 self.matches = OrderedDict()
49 self.ordered = True
50 self.id = None
51 self.fail_on_fault = True
52 self.fault = False
53 self.capture_coverage = False
54 self.next_pattern = 0
55 self.record = None
56 self.record_patterns = []
57 self.record_merge = False
58 self.record_as_json = None
59 self.recording = []
60 self.ztest = False
61 self.detected_suite_names = []
62 self.run_id = None
63 self.started_suites = {}
64 self.started_cases = {}
65 self.matched_run_id = False
66 self.run_id_exists = False
67 self.instance: TestInstance | None = None
68 self.testcase_output = ""
69 self._match = False
73 def trace(self) -> bool: argument
74 return self.instance.handler.options.verbose > 2
77 def status(self) -> TwisterStatus: argument
78 return self._status
81 def status(self, value : TwisterStatus) -> None: argument
85 self._status = TwisterStatus[key]
87 raise StatusAttributeError(self.__class__, value) from err
89 def configure(self, instance): argument
90 self.instance = instance
92 self.id = instance.testsuite.id
93 self.run_id = instance.run_id
94 self.expect_reboot = getattr(instance.testsuite, 'expect_reboot', False)
96 self.fail_on_fault = False
99 self.type = config.get('type', None)
100 self.regex = config.get('regex', [])
101 self.ordered = config.get('ordered', True)
102 self.record = config.get('record', {})
103 if self.record:
104 self.record_patterns = [re.compile(p) for p in self.record.get("regex", [])]
105 self.record_merge = self.record.get("merge", False)
106 self.record_as_json = self.record.get("as_json")
108 def build(self): argument
111 def get_testcase_name(self): argument
115 return self.id
117 def translate_record(self, record: dict) -> dict: argument
118 if self.record_as_json:
119 for k in self.record_as_json:
131 def parse_record(self, line) -> int: argument
133 for record_pattern in self.record_patterns:
137 rec = self.translate_record(
140 if self.record_merge and len(self.recording) > 0:
142 if k in self.recording[0]:
143 if isinstance(self.recording[0][k], list):
144 self.recording[0][k].append(v)
146 self.recording[0][k] = [self.recording[0][k], v]
148 self.recording[0][k] = v
150 self.recording.append(rec)
153 def process_test(self, line): argument
155 self.parse_record(line)
157 runid_match = re.search(self.run_id_pattern, line)
160 self.run_id_exists = True
161 if run_id == str(self.run_id):
162 self.matched_run_id = True
164 if self.RUN_PASSED in line:
165 if self.fault:
166 self.status = TwisterStatus.FAIL
167 self.reason = "Fault detected while running test"
169 self.status = TwisterStatus.PASS
171 if self.RUN_FAILED in line:
172 self.status = TwisterStatus.FAIL
173 self.reason = "Testsuite failed"
175 if self.fail_on_fault and line == self.FAULT:
176 self.fault = True
178 if self.GCOV_START in line:
179 self.capture_coverage = True
180 elif self.GCOV_END in line:
181 self.capture_coverage = False
187 def configure(self, instance): argument
189 self.instance = instance
193 self.path = config.get('robot_testsuite', None)
194 self.option = config.get('robot_option', None)
196 def handle(self, line): argument
202 self.instance.status = TwisterStatus.PASS
203 tc = self.instance.get_case_or_create(self.id)
206 def run_robot_test(self, command, handler): argument
210 if self.option:
211 if isinstance(self.option, list):
212 for option in self.option:
216 for v in str(self.option).split():
219 if self.path is None:
222 if isinstance(self.path, list):
223 for suite in self.path:
226 command.append(os.path.join(handler.sourcedir, self.path))
229 stderr=subprocess.STDOUT, cwd=self.instance.build_dir, env=env) as renode_test_proc:
232 self.instance.execution_time = time.time() - start_time
235 self.instance.status = TwisterStatus.PASS
239 self.instance.testcases[0].status = TwisterStatus.PASS
244 self.instance.status = TwisterStatus.FAIL
245 self.instance.testcases[0].status = TwisterStatus.FAIL
248 with open(os.path.join(self.instance.build_dir, handler.log), 'w') as log:
254 def get_testcase_name(self): argument
263 if self.instance and len(self.instance.testcases) == 1:
264 return self.instance.testcases[0].name
267 def configure(self, instance): argument
269 if self.regex is None or len(self.regex) == 0:
270 self.status = TwisterStatus.FAIL
271 tc = self.instance.set_case_status_by_name(
272 self.get_testcase_name(),
276 raise ConfigurationError(self.instance.name, tc.reason)
277 if self.type == "one_line":
278 self.pattern = re.compile(self.regex[0])
279 self.patterns_expected = 1
280 elif self.type == "multi_line":
281 self.patterns = []
282 for r in self.regex:
283 self.patterns.append(re.compile(r))
284 self.patterns_expected = len(self.patterns)
286 self.status = TwisterStatus.FAIL
287 tc = self.instance.set_case_status_by_name(
288 self.get_testcase_name(),
292 raise ConfigurationError(self.instance.name, tc.reason)
295 def handle(self, line): argument
296 if self.type == "one_line":
297 if self.pattern.search(line):
300 self.next_pattern += 1
301 self.status = TwisterStatus.PASS
302 elif self.type == "multi_line" and self.ordered:
303 if (self.next_pattern < len(self.patterns) and
304 self.patterns[self.next_pattern].search(line)):
308 self.next_pattern += 1
309 if self.next_pattern >= len(self.patterns):
310 self.status = TwisterStatus.PASS
311 elif self.type == "multi_line" and not self.ordered:
312 for i, pattern in enumerate(self.patterns):
313 r = self.regex[i]
314 if pattern.search(line) and r not in self.matches:
315 self.matches[r] = line
319 if len(self.matches) == len(self.regex):
320 self.status = TwisterStatus.PASS
324 if self.fail_on_fault and self.FAULT in line:
325 self.fault = True
327 if self.GCOV_START in line:
328 self.capture_coverage = True
329 elif self.GCOV_END in line:
330 self.capture_coverage = False
332 self.process_test(line)
340 if self.status == TwisterStatus.PASS and \
341 self.ordered and \
342 self.next_pattern < self.patterns_expected:
346 self.status = TwisterStatus.FAIL
347 self.reason = "patterns did not match (ordered)"
348 if self.status == TwisterStatus.PASS and \
349 not self.ordered and \
350 len(self.matches) < self.patterns_expected:
354 self.status = TwisterStatus.FAIL
355 self.reason = "patterns did not match (unordered)"
357 tc = self.instance.get_case_or_create(self.get_testcase_name())
358 if self.status == TwisterStatus.PASS:
370 def configure(self, instance: TestInstance): argument
372 self.running_dir = instance.build_dir
373 self.source_dir = instance.testsuite.source_dir
374 self.report_file = os.path.join(self.running_dir, 'report.xml')
375 self.pytest_log_file_path = os.path.join(self.running_dir, 'twister_harness.log')
376 self.reserved_dut = None
377 self._output = []
379 def pytest_run(self, timeout): argument
381 cmd = self.generate_command()
382 self.run_command(cmd, timeout)
385 self.status = TwisterStatus.FAIL
386 self.instance.reason = str(pytest_exception)
388 self.instance.record(self.recording)
389 self._update_test_status()
390 if self.reserved_dut:
391 self.instance.handler.make_dut_available(self.reserved_dut)
393 def generate_command(self): argument
394 config = self.instance.testsuite.harness_config
395 handler: Handler = self.instance.handler
409 self.source_dir, os.path.expanduser(os.path.expandvars(src)))) for src in pytest_root])
426 self._generate_parameters_for_hardware(handler)
451 def _generate_parameters_for_hardware(self, handler: Handler): argument
457 self.instance.dut = hardware.id
459 self.reserved_dut = hardware
505 def run_command(self, cmd, timeout): argument
506 cmd, env = self._update_command_with_env_dependencies(cmd)
514 reader_t = threading.Thread(target=self._output_reader, args=(proc,), daemon=True)
521 self.instance.reason = 'Pytest timeout'
522 self.status = TwisterStatus.FAIL
525 self.status = TwisterStatus.FAIL
529 self.status = TwisterStatus.ERROR
530 self.instance.reason = f'Pytest error - return code {proc.returncode}'
531 with open(self.pytest_log_file_path, 'w') as log_file:
533 log_file.write('\n'.join(self._output))
565 def _output_reader(self, proc): argument
566 self._output = []
571 self._output.append(line)
573 self.parse_record(line)
576 def _update_test_status(self): argument
577 if self.status == TwisterStatus.NONE:
578 self.instance.testcases = []
580 self._parse_report_file(self.report_file)
583 self.status = TwisterStatus.FAIL
585 if not self.instance.testcases:
586 self.instance.init_cases()
588 self.instance.status = self.status if self.status != TwisterStatus.NONE else \
590 if self.instance.status in [TwisterStatus.ERROR, TwisterStatus.FAIL]:
591 self.instance.reason = self.instance.reason or 'Pytest failed'
592 self.instance.add_missing_case_status(TwisterStatus.BLOCK, self.instance.reason)
594 def _parse_report_file(self, report): argument
600 self.status = TwisterStatus.FAIL
601 self.instance.reason = (
605 self.status = TwisterStatus.ERROR
606 self.instance.reason = 'Error during pytest execution'
608 self.status = TwisterStatus.SKIP
610 self.status = TwisterStatus.PASS
611 self.instance.execution_time = float(elem_ts.get('time'))
614 tc = self.instance.add_testcase(f"{self.id}.{elem_tc.get('name')}")
629 self.status = TwisterStatus.SKIP
630 self.instance.reason = 'No tests collected'
633 def generate_command(self): argument
634 config = self.instance.testsuite.harness_config
639 if test_config_file := self._get_display_config_file(config):
645 def _get_display_config_file(self, harness_config): argument
647 test_config_path = os.path.join(self.source_dir, test_config_file)
655 def generate_command(self): argument
656 config = self.instance.testsuite.harness_config
661 if test_shell_file := self._get_shell_commands_file(config):
667 def _get_shell_commands_file(self, harness_config): argument
669 test_shell_file = os.path.join(self.running_dir, 'test_shell.yml')
676 self.source_dir, os.path.expanduser(os.path.expandvars(test_shell_file))
683 def generate_command(self): argument
684 config = self.instance.testsuite.harness_config
690 if self.instance.testsuite.harness == 'power':
708 def __init__(self): argument
710 self.tc = None
711 self.has_failures = False
713 def handle(self, line): argument
715 non_ansi_line = self.ANSI_ESCAPE.sub('', line)
717 if self.status != TwisterStatus.NONE:
721 test_start_match = re.search(self.TEST_START_PATTERN, non_ansi_line)
725 if suite_name not in self.detected_suite_names:
726 self.detected_suite_names.append(suite_name)
729 name = "{}.{}.{}".format(self.id, suite_name, test_start_match.group("test_name"))
733 self.tc is None
737 tc = self.instance.get_case_by_name(name)
741 tc = self.instance.get_case_or_create(name)
742 self.tc = tc
743 self.tc.status = TwisterStatus.STARTED
744 self.testcase_output += line + "\n"
745 self._match = True
748 finished_match = re.search(self.FINISHED_PATTERN, non_ansi_line)
750 tc = self.instance.get_case_or_create(self.id)
751 if self.has_failures or self.tc is not None:
752 self.status = TwisterStatus.FAIL
755 self.status = TwisterStatus.PASS
760 state, name = self._check_result(non_ansi_line)
766 tc = self.instance.get_case_by_name(name)
768 tc is not None and tc == self.tc
772 self.tc = None
777 self.has_failures = True
778 tc.output = self.testcase_output
779 self.testcase_output = ""
780 self._match = False
782 def _check_result(self, line): argument
783 test_pass_match = re.search(self.TEST_PASS_PATTERN, line)
787 self.id, test_pass_match.group("suite_name"),
790 test_skip_match = re.search(self.TEST_SKIP_PATTERN, line)
794 self.id, test_skip_match.group("suite_name"),
797 test_fail_match = re.search(self.TEST_FAIL_PATTERN, line)
801 self.id, test_fail_match.group("suite_name"),
830 def get_testcase(self, tc_name, phase, ts_name=None): argument
835 ts_names = self.started_suites.keys()
837 if self.trace and ts_name not in self.instance.testsuite.ztest_suite_names:
842 if ts_name not in self.detected_suite_names:
843 if self.trace:
845 self.detected_suite_names.append(ts_name)
850 if self.started_suites[ts_name_]['count'] < (0 if phase == 'TS_SUM' else 1):
852 tc_fq_id = self.instance.compose_case_name(f"{ts_name_}.{tc_name}")
853 if tc := self.instance.get_case_by_name(tc_fq_id):
854 if self.trace:
861 tc_id = self.instance.compose_case_name(tc_name)
862 return self.instance.get_case_or_create(tc_id)
864 def start_suite(self, suite_name, phase='TS_START'): argument
865 if suite_name not in self.detected_suite_names:
866 self.detected_suite_names.append(suite_name)
867 if self.trace and suite_name not in self.instance.testsuite.ztest_suite_names:
872 if suite_name in self.started_suites:
873 if self.started_suites[suite_name]['count'] > 0 and not self.expect_reboot:
877 elif self.trace:
879 self.started_suites[suite_name]['count'] += 1
880 self.started_suites[suite_name]['repeat'] += 1
882 self.started_suites[suite_name] = { 'count': 1, 'repeat': 0 }
884 def end_suite(self, suite_name, phase='TS_END', suite_status=None): argument
885 if suite_name in self.started_suites:
886 if phase == 'TS_SUM' and self.started_suites[suite_name]['count'] == 0:
888 if self.started_suites[suite_name]['count'] < 1:
892 elif self.trace:
894 self.started_suites[suite_name]['count'] -= 1
896 self.start_suite(suite_name, phase) # register skipped suites at their summary end
897 self.started_suites[suite_name]['count'] -= 1
901 def start_case(self, tc_name, phase='TC_START'): argument
902 if tc_name in self.started_cases:
903 if self.started_cases[tc_name]['count'] > 0 and not self.expect_reboot:
906 self.started_cases[tc_name]['count'] += 1
908 self.started_cases[tc_name] = { 'count': 1 }
910 def end_case(self, tc_name, phase='TC_END'): argument
911 if tc_name in self.started_cases:
912 if phase == 'TS_SUM' and self.started_cases[tc_name]['count'] == 0:
914 if self.started_cases[tc_name]['count'] < 1:
918 elif self.trace:
920 self.started_cases[tc_name]['count'] -= 1
924 def handle(self, line): argument
926 if self._match:
927 self.testcase_output += line + "\n"
928 if test_suite_start_match := re.search(self.test_suite_start_pattern, line):
929 self.start_suite(test_suite_start_match.group("suite_name"))
930 elif test_suite_end_match := re.search(self.test_suite_end_pattern, line):
932 self.end_suite(suite_name)
933 self.ztest = True
934 elif testcase_match := re.search(self.test_case_start_pattern, line):
936 tc = self.get_testcase(tc_name, 'TC_START')
937 self.start_case(tc.name)
942 if not self._match:
943 self.testcase_output += line + "\n"
944 self._match = True
948 elif result_match := self.test_case_end_pattern.match(line):
951 tc = self.get_testcase(tc_name, 'TC_END')
952 self.end_case(tc.name)
958 tc.output = self.testcase_output
959 self.testcase_output = ""
960 self._match = False
961 self.ztest = True
962 elif test_suite_summary_match := self.test_suite_summary_pattern.match(line):
965 self._match = False
966 self.ztest = True
967 self.end_suite(suite_name, 'TS_SUM', suite_status=suite_status)
968 elif test_case_summary_match := self.test_case_summary_pattern.match(line):
972 tc = self.get_testcase(tc_name, 'TS_SUM', suite_name)
973 self.end_case(tc.name, 'TS_SUM')
984 tc.output = self.testcase_output
985 self.testcase_output = ""
986 self._match = False
987 self.ztest = True
989 self.process_test(line)
991 if not self.ztest and self.status != TwisterStatus.NONE:
993 tc = self.instance.get_case_or_create(self.id)
994 if self.status == TwisterStatus.PASS:
1007 def build(self): argument
1013 if self.instance is None:
1016 original_exe_path: str = os.path.join(self.instance.build_dir, 'zephyr', 'zephyr.exe')
1026 new_exe_name: str = self.instance.testsuite.harness_config.get('bsim_exe_name', '')
1030 new_exe_name = self.instance.name
1040 def configure(self, instance: TestInstance): argument
1042 self.running_dir = instance.build_dir
1043 self.report_file = os.path.join(self.running_dir, 'report.xml')
1044 self.ctest_log_file_path = os.path.join(self.running_dir, 'twister_harness.log')
1045 self._output = []
1047 def ctest_run(self, timeout): argument
1048 assert self.instance is not None
1050 cmd = self.generate_command()
1051 self.run_command(cmd, timeout)
1054 self.status = TwisterStatus.FAIL
1055 self.instance.reason = str(err)
1057 self.instance.record(self.recording)
1058 self._update_test_status()
1060 def generate_command(self): argument
1061 config = self.instance.testsuite.harness_config
1062 handler: Handler = self.instance.handler
1068 self.running_dir,
1070 self.report_file,
1072 self.ctest_log_file_path,
1084 def run_command(self, cmd, timeout): argument
1091 reader_t = threading.Thread(target=self._output_reader, args=(proc,), daemon=True)
1098 self.instance.reason = 'Ctest timeout'
1099 self.status = TwisterStatus.FAIL
1102 self.status = TwisterStatus.FAIL
1106 self.status = TwisterStatus.ERROR
1107 self.instance.reason = f'Ctest error - return code {proc.returncode}'
1108 with open(self.ctest_log_file_path, 'w') as log_file:
1110 log_file.write('\n'.join(self._output))
1112 def _output_reader(self, proc): argument
1113 self._output = []
1118 self._output.append(line)
1120 self.parse_record(line)
1123 def _update_test_status(self): argument
1124 if self.status == TwisterStatus.NONE:
1125 self.instance.testcases = []
1127 self._parse_report_file(self.report_file)
1130 self.status = TwisterStatus.FAIL
1132 if not self.instance.testcases:
1133 self.instance.init_cases()
1135 self.instance.status = self.status if self.status != TwisterStatus.NONE else \
1137 if self.instance.status in [TwisterStatus.ERROR, TwisterStatus.FAIL]:
1138 self.instance.reason = self.instance.reason or 'Ctest failed'
1139 self.instance.add_missing_case_status(TwisterStatus.BLOCK, self.instance.reason)
1141 def _parse_report_file(self, report): argument
1144 self.status = TwisterStatus.SKIP
1145 self.instance.reason = 'No tests collected'
1152 self.status = TwisterStatus.FAIL
1153 self.instance.reason = f"{suite.failures}/{suite.tests} ctest scenario(s) failed"
1155 self.status = TwisterStatus.ERROR
1156 self.instance.reason = 'Error during ctest execution'
1158 self.status = TwisterStatus.SKIP
1160 self.status = TwisterStatus.PASS
1161 self.instance.execution_time = suite.time
1164 tc = self.instance.add_testcase(f"{self.id}.{case.name}")