Lines Matching refs:self
42 def __init__(self): argument
43 self._status = TwisterStatus.NONE
44 self.reason = None
45 self.type = None
46 self.regex = []
47 self.matches = OrderedDict()
48 self.ordered = True
49 self.id = None
50 self.fail_on_fault = True
51 self.fault = False
52 self.capture_coverage = False
53 self.next_pattern = 0
54 self.record = None
55 self.record_pattern = None
56 self.record_as_json = None
57 self.recording = []
58 self.ztest = False
59 self.detected_suite_names = []
60 self.run_id = None
61 self.started_suites = {}
62 self.started_cases = {}
63 self.matched_run_id = False
64 self.run_id_exists = False
65 self.instance: TestInstance | None = None
66 self.testcase_output = ""
67 self._match = False
71 def trace(self) -> bool: argument
72 return self.instance.handler.options.verbose > 2
75 def status(self) -> TwisterStatus: argument
76 return self._status
79 def status(self, value : TwisterStatus) -> None: argument
83 self._status = TwisterStatus[key]
85 raise StatusAttributeError(self.__class__, value) from err
87 def configure(self, instance): argument
88 self.instance = instance
90 self.id = instance.testsuite.id
91 self.run_id = instance.run_id
93 self.fail_on_fault = False
96 self.type = config.get('type', None)
97 self.regex = config.get('regex', [])
98 self.ordered = config.get('ordered', True)
99 self.record = config.get('record', {})
100 if self.record:
101 self.record_pattern = re.compile(self.record.get("regex", ""))
102 self.record_as_json = self.record.get("as_json")
104 def build(self): argument
107 def get_testcase_name(self): argument
111 return self.id
113 def translate_record(self, record: dict) -> dict: argument
114 if self.record_as_json:
115 for k in self.record_as_json:
127 def parse_record(self, line) -> re.Match: argument
129 if self.record_pattern:
130 match = self.record_pattern.search(line)
132 rec = self.translate_record(
135 self.recording.append(rec)
139 def process_test(self, line): argument
141 self.parse_record(line)
143 runid_match = re.search(self.run_id_pattern, line)
146 self.run_id_exists = True
147 if run_id == str(self.run_id):
148 self.matched_run_id = True
150 if self.RUN_PASSED in line:
151 if self.fault:
152 self.status = TwisterStatus.FAIL
153 self.reason = "Fault detected while running test"
155 self.status = TwisterStatus.PASS
157 if self.RUN_FAILED in line:
158 self.status = TwisterStatus.FAIL
159 self.reason = "Testsuite failed"
161 if self.fail_on_fault and line == self.FAULT:
162 self.fault = True
164 if self.GCOV_START in line:
165 self.capture_coverage = True
166 elif self.GCOV_END in line:
167 self.capture_coverage = False
173 def configure(self, instance): argument
175 self.instance = instance
179 self.path = config.get('robot_testsuite', None)
180 self.option = config.get('robot_option', None)
182 def handle(self, line): argument
188 self.instance.status = TwisterStatus.PASS
189 tc = self.instance.get_case_or_create(self.id)
192 def run_robot_test(self, command, handler): argument
196 if self.option:
197 if isinstance(self.option, list):
198 for option in self.option:
202 for v in str(self.option).split():
205 if self.path is None:
208 if isinstance(self.path, list):
209 for suite in self.path:
212 command.append(os.path.join(handler.sourcedir, self.path))
215 stderr=subprocess.STDOUT, cwd=self.instance.build_dir, env=env) as renode_test_proc:
218 self.instance.execution_time = time.time() - start_time
221 self.instance.status = TwisterStatus.PASS
225 self.instance.testcases[0].status = TwisterStatus.PASS
230 self.instance.status = TwisterStatus.FAIL
231 self.instance.testcases[0].status = TwisterStatus.FAIL
234 with open(os.path.join(self.instance.build_dir, handler.log), 'w') as log:
240 def get_testcase_name(self): argument
249 if self.instance and len(self.instance.testcases) == 1:
250 return self.instance.testcases[0].name
253 def configure(self, instance): argument
255 if self.regex is None or len(self.regex) == 0:
256 self.status = TwisterStatus.FAIL
257 tc = self.instance.set_case_status_by_name(
258 self.get_testcase_name(),
262 raise ConfigurationError(self.instance.name, tc.reason)
263 if self.type == "one_line":
264 self.pattern = re.compile(self.regex[0])
265 self.patterns_expected = 1
266 elif self.type == "multi_line":
267 self.patterns = []
268 for r in self.regex:
269 self.patterns.append(re.compile(r))
270 self.patterns_expected = len(self.patterns)
272 self.status = TwisterStatus.FAIL
273 tc = self.instance.set_case_status_by_name(
274 self.get_testcase_name(),
278 raise ConfigurationError(self.instance.name, tc.reason)
281 def handle(self, line): argument
282 if self.type == "one_line":
283 if self.pattern.search(line):
286 self.next_pattern += 1
287 self.status = TwisterStatus.PASS
288 elif self.type == "multi_line" and self.ordered:
289 if (self.next_pattern < len(self.patterns) and
290 self.patterns[self.next_pattern].search(line)):
294 self.next_pattern += 1
295 if self.next_pattern >= len(self.patterns):
296 self.status = TwisterStatus.PASS
297 elif self.type == "multi_line" and not self.ordered:
298 for i, pattern in enumerate(self.patterns):
299 r = self.regex[i]
300 if pattern.search(line) and r not in self.matches:
301 self.matches[r] = line
305 if len(self.matches) == len(self.regex):
306 self.status = TwisterStatus.PASS
310 if self.fail_on_fault and self.FAULT in line:
311 self.fault = True
313 if self.GCOV_START in line:
314 self.capture_coverage = True
315 elif self.GCOV_END in line:
316 self.capture_coverage = False
318 self.process_test(line)
326 if self.status == TwisterStatus.PASS and \
327 self.ordered and \
328 self.next_pattern < self.patterns_expected:
332 self.status = TwisterStatus.FAIL
333 self.reason = "patterns did not match (ordered)"
334 if self.status == TwisterStatus.PASS and \
335 not self.ordered and \
336 len(self.matches) < self.patterns_expected:
340 self.status = TwisterStatus.FAIL
341 self.reason = "patterns did not match (unordered)"
343 tc = self.instance.get_case_or_create(self.get_testcase_name())
344 if self.status == TwisterStatus.PASS:
356 def configure(self, instance: TestInstance): argument
358 self.running_dir = instance.build_dir
359 self.source_dir = instance.testsuite.source_dir
360 self.report_file = os.path.join(self.running_dir, 'report.xml')
361 self.pytest_log_file_path = os.path.join(self.running_dir, 'twister_harness.log')
362 self.reserved_dut = None
363 self._output = []
365 def pytest_run(self, timeout): argument
367 cmd = self.generate_command()
368 self.run_command(cmd, timeout)
371 self.status = TwisterStatus.FAIL
372 self.instance.reason = str(pytest_exception)
374 self.instance.record(self.recording)
375 self._update_test_status()
376 if self.reserved_dut:
377 self.instance.handler.make_dut_available(self.reserved_dut)
379 def generate_command(self): argument
380 config = self.instance.testsuite.harness_config
381 handler: Handler = self.instance.handler
397 self.source_dir, os.path.expanduser(os.path.expandvars(src)))) for src in pytest_root])
414 self._generate_parameters_for_hardware(handler)
439 def _generate_parameters_for_hardware(self, handler: Handler): argument
445 self.instance.dut = hardware.id
447 self.reserved_dut = hardware
493 def run_command(self, cmd, timeout): argument
494 cmd, env = self._update_command_with_env_dependencies(cmd)
502 reader_t = threading.Thread(target=self._output_reader, args=(proc,), daemon=True)
509 self.instance.reason = 'Pytest timeout'
510 self.status = TwisterStatus.FAIL
513 self.status = TwisterStatus.FAIL
517 self.status = TwisterStatus.ERROR
518 self.instance.reason = f'Pytest error - return code {proc.returncode}'
519 with open(self.pytest_log_file_path, 'w') as log_file:
521 log_file.write('\n'.join(self._output))
553 def _output_reader(self, proc): argument
554 self._output = []
559 self._output.append(line)
561 self.parse_record(line)
564 def _update_test_status(self): argument
565 if self.status == TwisterStatus.NONE:
566 self.instance.testcases = []
568 self._parse_report_file(self.report_file)
571 self.status = TwisterStatus.FAIL
573 if not self.instance.testcases:
574 self.instance.init_cases()
576 self.instance.status = self.status if self.status != TwisterStatus.NONE else \
578 if self.instance.status in [TwisterStatus.ERROR, TwisterStatus.FAIL]:
579 self.instance.reason = self.instance.reason or 'Pytest failed'
580 self.instance.add_missing_case_status(TwisterStatus.BLOCK, self.instance.reason)
582 def _parse_report_file(self, report): argument
588 self.status = TwisterStatus.FAIL
589 self.instance.reason = (
593 self.status = TwisterStatus.ERROR
594 self.instance.reason = 'Error during pytest execution'
596 self.status = TwisterStatus.SKIP
598 self.status = TwisterStatus.PASS
599 self.instance.execution_time = float(elem_ts.get('time'))
602 tc = self.instance.add_testcase(f"{self.id}.{elem_tc.get('name')}")
617 self.status = TwisterStatus.SKIP
618 self.instance.reason = 'No tests collected'
634 def __init__(self): argument
636 self.tc = None
637 self.has_failures = False
639 def handle(self, line): argument
641 non_ansi_line = self.ANSI_ESCAPE.sub('', line)
643 if self.status != TwisterStatus.NONE:
647 test_start_match = re.search(self.TEST_START_PATTERN, non_ansi_line)
651 if suite_name not in self.detected_suite_names:
652 self.detected_suite_names.append(suite_name)
655 name = "{}.{}.{}".format(self.id, suite_name, test_start_match.group("test_name"))
659 self.tc is None
663 tc = self.instance.get_case_by_name(name)
667 tc = self.instance.get_case_or_create(name)
668 self.tc = tc
669 self.tc.status = TwisterStatus.STARTED
670 self.testcase_output += line + "\n"
671 self._match = True
674 finished_match = re.search(self.FINISHED_PATTERN, non_ansi_line)
676 tc = self.instance.get_case_or_create(self.id)
677 if self.has_failures or self.tc is not None:
678 self.status = TwisterStatus.FAIL
681 self.status = TwisterStatus.PASS
686 state, name = self._check_result(non_ansi_line)
692 tc = self.instance.get_case_by_name(name)
694 tc is not None and tc == self.tc
698 self.tc = None
703 self.has_failures = True
704 tc.output = self.testcase_output
705 self.testcase_output = ""
706 self._match = False
708 def _check_result(self, line): argument
709 test_pass_match = re.search(self.TEST_PASS_PATTERN, line)
713 self.id, test_pass_match.group("suite_name"),
716 test_skip_match = re.search(self.TEST_SKIP_PATTERN, line)
720 self.id, test_skip_match.group("suite_name"),
723 test_fail_match = re.search(self.TEST_FAIL_PATTERN, line)
727 self.id, test_fail_match.group("suite_name"),
753 def get_testcase(self, tc_name, phase, ts_name=None): argument
758 ts_names = self.started_suites.keys()
760 if ts_name not in self.instance.testsuite.ztest_suite_names:
763 if ts_name not in self.detected_suite_names:
764 if self.trace:
766 self.detected_suite_names.append(ts_name)
771 if self.started_suites[ts_name_]['count'] < (0 if phase == 'TS_SUM' else 1):
773 tc_fq_id = self.instance.compose_case_name(f"{ts_name_}.{tc_name}")
774 if tc := self.instance.get_case_by_name(tc_fq_id):
775 if self.trace:
782 tc_id = self.instance.compose_case_name(tc_name)
783 return self.instance.get_case_or_create(tc_id)
785 def start_suite(self, suite_name): argument
786 if suite_name not in self.detected_suite_names:
787 self.detected_suite_names.append(suite_name)
788 if suite_name not in self.instance.testsuite.ztest_suite_names:
790 if suite_name in self.started_suites:
791 if self.started_suites[suite_name]['count'] > 0:
793 elif self.trace:
795 self.started_suites[suite_name]['count'] += 1
796 self.started_suites[suite_name]['repeat'] += 1
798 self.started_suites[suite_name] = { 'count': 1, 'repeat': 0 }
800 def end_suite(self, suite_name, phase='', suite_status=None): argument
801 if suite_name in self.started_suites:
802 if phase == 'TS_SUM' and self.started_suites[suite_name]['count'] == 0:
804 if self.started_suites[suite_name]['count'] < 1:
808 elif self.trace:
810 self.started_suites[suite_name]['count'] -= 1
812 self.start_suite(suite_name) # register skipped suites at their summary end
813 self.started_suites[suite_name]['count'] -= 1
817 def start_case(self, tc_name): argument
818 if tc_name in self.started_cases:
819 if self.started_cases[tc_name]['count'] > 0:
821 self.started_cases[tc_name]['count'] += 1
823 self.started_cases[tc_name] = { 'count': 1 }
825 def end_case(self, tc_name, phase=''): argument
826 if tc_name in self.started_cases:
827 if phase == 'TS_SUM' and self.started_cases[tc_name]['count'] == 0:
829 if self.started_cases[tc_name]['count'] < 1:
833 elif self.trace:
835 self.started_cases[tc_name]['count'] -= 1
840 def handle(self, line): argument
842 if self._match:
843 self.testcase_output += line + "\n"
845 if test_suite_start_match := re.search(self.test_suite_start_pattern, line):
846 self.start_suite(test_suite_start_match.group("suite_name"))
847 elif test_suite_end_match := re.search(self.test_suite_end_pattern, line):
849 self.end_suite(suite_name, 'TS_END')
850 elif testcase_match := re.search(self.test_case_start_pattern, line):
852 tc = self.get_testcase(tc_name, 'TC_START')
853 self.start_case(tc.name)
858 if not self._match:
859 self.testcase_output += line + "\n"
860 self._match = True
864 elif result_match := self.test_case_end_pattern.match(line):
867 tc = self.get_testcase(tc_name, 'TC_END')
868 self.end_case(tc.name)
874 tc.output = self.testcase_output
875 self.testcase_output = ""
876 self._match = False
877 self.ztest = True
878 elif test_suite_summary_match := self.test_suite_summary_pattern.match(line):
881 self._match = False
882 self.ztest = True
883 self.end_suite(suite_name, 'TS_SUM', suite_status=suite_status)
884 elif test_case_summary_match := self.test_case_summary_pattern.match(line):
888 tc = self.get_testcase(tc_name, 'TS_SUM', suite_name)
889 self.end_case(tc.name, 'TS_SUM')
895 tc.output = self.testcase_output
896 self.testcase_output = ""
897 self._match = False
898 self.ztest = True
900 self.process_test(line)
902 if not self.ztest and self.status != TwisterStatus.NONE:
904 tc = self.instance.get_case_or_create(self.id)
905 if self.status == TwisterStatus.PASS:
918 def build(self): argument
924 if self.instance is None:
927 original_exe_path: str = os.path.join(self.instance.build_dir, 'zephyr', 'zephyr.exe')
937 new_exe_name: str = self.instance.testsuite.harness_config.get('bsim_exe_name', '')
941 new_exe_name = self.instance.name