Lines Matching full:self
45 def __init__(self): argument
46 self.state = None
47 self.type = None
48 self.regex = []
49 self.matches = OrderedDict()
50 self.ordered = True
51 self.repeat = 1
52 self.testcases = []
53 self.id = None
54 self.fail_on_fault = True
55 self.fault = False
56 self.capture_coverage = False
57 self.next_pattern = 0
58 self.record = None
59 self.recording = []
60 self.fieldnames = []
61 self.ztest = False
62 self.detected_suite_names = []
63 self.run_id = None
64 self.matched_run_id = False
65 self.run_id_exists = False
66 self.instance = None
67 self.testcase_output = ""
68 self._match = False
70 def configure(self, instance): argument
71 self.instance = instance
73 self.id = instance.testsuite.id
74 self.run_id = instance.run_id
76 self.fail_on_fault = False
79 self.type = config.get('type', None)
80 self.regex = config.get('regex', [])
81 self.repeat = config.get('repeat', 1)
82 self.ordered = config.get('ordered', True)
83 self.record = config.get('record', {})
85 def process_test(self, line): argument
87 runid_match = re.search(self.run_id_pattern, line)
90 self.run_id_exists = True
91 if run_id == str(self.run_id):
92 self.matched_run_id = True
94 if self.RUN_PASSED in line:
95 if self.fault:
96 self.state = "failed"
98 self.state = "passed"
100 if self.RUN_FAILED in line:
101 self.state = "failed"
103 if self.fail_on_fault:
104 if self.FAULT == line:
105 self.fault = True
107 if self.GCOV_START in line:
108 self.capture_coverage = True
109 elif self.GCOV_END in line:
110 self.capture_coverage = False
116 def configure(self, instance): argument
117 super(Robot, self).configure(instance)
118 self.instance = instance
122 self.path = config.get('robot_test_path', None)
124 def handle(self, line): argument
130 self.instance.state = "passed"
131 tc = self.instance.get_case_or_create(self.id)
134 def run_robot_test(self, command, handler): argument
138 env["ROBOT_FILES"] = self.path
141 stderr=subprocess.STDOUT, cwd=self.instance.build_dir, env=env) as cmake_proc:
144 self.instance.execution_time = time.time() - start_time
147 self.instance.status = "passed"
150 (handler.sourcedir, self.instance.platform.name))
151 self.instance.status = "failed"
154 with open(os.path.join(self.instance.build_dir, handler.log), "wt") as log:
160 def configure(self, instance): argument
161 super(Console, self).configure(instance)
162 if self.type == "one_line":
163 self.pattern = re.compile(self.regex[0])
164 elif self.type == "multi_line":
165 self.patterns = []
166 for r in self.regex:
167 self.patterns.append(re.compile(r))
169 def handle(self, line): argument
170 if self.type == "one_line":
171 if self.pattern.search(line):
172 self.state = "passed"
173 elif self.type == "multi_line" and self.ordered:
174 if (self.next_pattern < len(self.patterns) and
175 self.patterns[self.next_pattern].search(line)):
176 self.next_pattern += 1
177 if self.next_pattern >= len(self.patterns):
178 self.state = "passed"
179 elif self.type == "multi_line" and not self.ordered:
180 for i, pattern in enumerate(self.patterns):
181 r = self.regex[i]
182 if pattern.search(line) and not r in self.matches:
183 self.matches[r] = line
184 if len(self.matches) == len(self.regex):
185 self.state = "passed"
189 if self.fail_on_fault:
190 if self.FAULT in line:
191 self.fault = True
193 if self.GCOV_START in line:
194 self.capture_coverage = True
195 elif self.GCOV_END in line:
196 self.capture_coverage = False
199 if self.record:
200 pattern = re.compile(self.record.get("regex", ""))
204 if not self.fieldnames:
206 self.fieldnames.append(k)
210 self.recording.append(csv)
212 self.process_test(line)
214 tc = self.instance.get_case_or_create(self.id)
215 if self.state == "passed":
227 def configure(self, instance): argument
228 super(Pytest, self).configure(instance)
229 self.running_dir = instance.build_dir
230 self.source_dir = instance.testsuite.source_dir
231 self.report_file = os.path.join(self.running_dir, 'report.xml')
232 self.reserved_serial = None
234 def pytest_run(self, timeout): argument
236 cmd = self.generate_command()
240 self.run_command(cmd, timeout)
244 if self.reserved_serial:
245 self.instance.handler.make_device_available(self.reserved_serial)
246 self._apply_instance_status()
248 def generate_command(self): argument
249 config = self.instance.testsuite.harness_config
257 os.path.join(self.source_dir, pytest_root),
258 f'--build-dir={self.running_dir}',
259 f'--junit-xml={self.report_file}'
263 handler = self.instance.handler
270 self._generate_parameters_for_hardware(handler)
280 def _generate_parameters_for_hardware(self, handler): argument
286 self.reserved_serial = hardware.serial_pty or hardware.serial
319 def run_command(self, cmd, timeout): argument
320 cmd, env = self._update_command_with_env_dependencies(cmd)
326 reader_t = threading.Thread(target=self._output_reader, args=(proc,), daemon=True)
332 self.state = 'failed'
335 if self.state != 'failed':
336 tree = ET.parse(self.report_file)
341 self.state = "failed"
343 self.state = "skipped"
345 self.state = "error"
347 self.state = "passed"
348 self.instance.execution_time = float(child.attrib['time'])
351 self.state = "failed"
353 self.state = "failed"
356 self.state = "failed"
358 tc = self.instance.get_case_or_create(self.id)
359 if self.state == "passed":
362 elif self.state == "skipped":
400 def _apply_instance_status(self): argument
401 if self.state:
402 self.instance.status = self.state
403 if self.state in ["error", "failed"]:
404 self.instance.reason = "Pytest failed"
406 self.instance.status = "failed"
407 self.instance.reason = "Pytest timeout"
408 if self.instance.status in ["error", "failed"]:
409 self.instance.add_missing_case_status("blocked", self.instance.reason)
421 def handle(self, line): argument
423 non_ansi_line = self.ANSI_ESCAPE.sub('', line)
426 test_start_match = re.search(self.TEST_START_PATTERN, non_ansi_line)
430 if suite_name not in self.detected_suite_names:
431 self.detected_suite_names.append(suite_name)
434 name = "{}.{}.{}".format(self.id, suite_name, test_start_match.group("test_name"))
438 self.tc is None
439 ), "gTest error, {} didn't finish".format(self.tc)
442 tc = self.instance.get_case_by_name(name)
446 tc = self.instance.get_case_or_create(name)
447 self.tc = tc
448 self.tc.status = "started"
449 self.testcase_output += line + "\n"
450 self._match = True
453 finished_match = re.search(self.FINISHED_PATTERN, non_ansi_line)
455 tc = self.instance.get_case_or_create(self.id)
456 if self.has_failures or self.tc is not None:
457 self.state = "failed"
460 self.state = "passed"
465 state, name = self._check_result(non_ansi_line)
471 tc = self.instance.get_case_by_name(name)
473 tc is not None and tc == self.tc
474 ), "gTest error, mismatched tests. Expected {} but got {}".format(self.tc, tc)
477 self.tc = None
482 self.has_failures = True
483 tc.output = self.testcase_output
484 self.testcase_output = ""
485 self._match = False
487 def _check_result(self, line): argument
488 test_pass_match = re.search(self.TEST_PASS_PATTERN, line)
490 …return "passed", "{}.{}.{}".format(self.id, test_pass_match.group("suite_name"), test_pass_match.g…
491 test_fail_match = re.search(self.TEST_FAIL_PATTERN, line)
493 …return "failed", "{}.{}.{}".format(self.id, test_fail_match.group("suite_name"), test_fail_match.g…
503 def handle(self, line): argument
504 test_suite_match = re.search(self.test_suite_start_pattern, line)
507 self.detected_suite_names.append(suite_name)
509 testcase_match = re.search(self.ZTEST_START_PATTERN, line)
511 name = "{}.{}".format(self.id, testcase_match.group(2))
512 tc = self.instance.get_case_or_create(name)
518 if testcase_match or self._match:
519 self.testcase_output += line + "\n"
520 self._match = True
526 name = "{}.{}".format(self.id, result_match.group(3))
527 tc = self.instance.get_case_or_create(name)
528 tc.status = self.ztest_to_status[matched_status]
533 tc.output = self.testcase_output
534 self.testcase_output = ""
535 self._match = False
536 self.ztest = True
538 self.process_test(line)
540 if not self.ztest and self.state:
541 logger.debug(f"not a ztest and no state for {self.id}")
542 tc = self.instance.get_case_or_create(self.id)
543 if self.state == "passed":