| # SPDX-License-Identifier: Apache-2.0 |
| import re |
| from collections import OrderedDict |
| |
| result_re = re.compile(".*(PASS|FAIL|SKIP) - (test_)?(.*)") |
| |
| class Harness: |
| GCOV_START = "GCOV_COVERAGE_DUMP_START" |
| GCOV_END = "GCOV_COVERAGE_DUMP_END" |
| FAULT = "ZEPHYR FATAL ERROR" |
| RUN_PASSED = "PROJECT EXECUTION SUCCESSFUL" |
| RUN_FAILED = "PROJECT EXECUTION FAILED" |
| |
| def __init__(self): |
| self.state = None |
| self.type = None |
| self.regex = [] |
| self.matches = OrderedDict() |
| self.ordered = True |
| self.repeat = 1 |
| self.tests = {} |
| self.id = None |
| self.fail_on_fault = True |
| self.fault = False |
| self.capture_coverage = False |
| self.next_pattern = 0 |
| self.record = None |
| self.recording = [] |
| self.fieldnames = [] |
| self.ztest = False |
| |
| def configure(self, instance): |
| config = instance.testcase.harness_config |
| self.id = instance.testcase.id |
| if "ignore_faults" in instance.testcase.tags: |
| self.fail_on_fault = False |
| |
| if config: |
| self.type = config.get('type', None) |
| self.regex = config.get('regex', []) |
| self.repeat = config.get('repeat', 1) |
| self.ordered = config.get('ordered', True) |
| self.record = config.get('record', {}) |
| |
| def process_test(self, line): |
| |
| if self.RUN_PASSED in line: |
| if self.fault: |
| self.state = "failed" |
| else: |
| self.state = "passed" |
| |
| if self.RUN_FAILED in line: |
| self.state = "failed" |
| |
| if self.fail_on_fault: |
| if self.FAULT == line: |
| self.fault = True |
| |
| if self.GCOV_START in line: |
| self.capture_coverage = True |
| elif self.GCOV_END in line: |
| self.capture_coverage = False |
| |
| class Console(Harness): |
| |
| def configure(self, instance): |
| super(Console, self).configure(instance) |
| if self.type == "one_line": |
| self.pattern = re.compile(self.regex[0]) |
| elif self.type == "multi_line": |
| self.patterns = [] |
| for r in self.regex: |
| self.patterns.append(re.compile(r)) |
| |
| def handle(self, line): |
| if self.type == "one_line": |
| if self.pattern.search(line): |
| self.state = "passed" |
| elif self.type == "multi_line" and self.ordered: |
| if (self.next_pattern < len(self.patterns) and |
| self.patterns[self.next_pattern].search(line)): |
| self.next_pattern += 1 |
| if self.next_pattern >= len(self.patterns): |
| self.state = "passed" |
| elif self.type == "multi_line" and not self.ordered: |
| for i, pattern in enumerate(self.patterns): |
| r = self.regex[i] |
| if pattern.search(line) and not r in self.matches: |
| self.matches[r] = line |
| if len(self.matches) == len(self.regex): |
| self.state = "passed" |
| |
| if self.fail_on_fault: |
| if self.FAULT in line: |
| self.fault = True |
| |
| if self.GCOV_START in line: |
| self.capture_coverage = True |
| elif self.GCOV_END in line: |
| self.capture_coverage = False |
| |
| |
| if self.record: |
| pattern = re.compile(self.record.get("regex", "")) |
| match = pattern.search(line) |
| if match: |
| csv = [] |
| if not self.fieldnames: |
| for k,v in match.groupdict().items(): |
| self.fieldnames.append(k) |
| |
| for k,v in match.groupdict().items(): |
| csv.append(v.strip()) |
| self.recording.append(csv) |
| |
| if self.state == "passed": |
| self.tests[self.id] = "PASS" |
| else: |
| self.tests[self.id] = "FAIL" |
| |
| self.process_test(line) |
| |
| class Test(Harness): |
| RUN_PASSED = "PROJECT EXECUTION SUCCESSFUL" |
| RUN_FAILED = "PROJECT EXECUTION FAILED" |
| |
| def handle(self, line): |
| match = result_re.match(line) |
| if match and match.group(2): |
| name = "{}.{}".format(self.id, match.group(3)) |
| self.tests[name] = match.group(1) |
| self.ztest = True |
| |
| if self.RUN_PASSED in line: |
| if self.fault: |
| self.state = "failed" |
| else: |
| self.state = "passed" |
| |
| if self.RUN_FAILED in line: |
| self.state = "failed" |
| |
| if self.fail_on_fault: |
| if self.FAULT in line: |
| self.fault = True |
| |
| if not self.ztest and self.state: |
| if self.state == "passed": |
| self.tests[self.id] = "PASS" |
| else: |
| self.tests[self.id] = "FAIL" |
| |
| if self.GCOV_START in line: |
| self.capture_coverage = True |
| elif self.GCOV_END in line: |
| self.capture_coverage = False |
| |
| self.process_test(line) |
| |
| class Ztest(Test): |
| pass |
