diff options
Diffstat (limited to 'tools/testing/kunit/kunit_parser.py')
| -rw-r--r-- | tools/testing/kunit/kunit_parser.py | 81 |
1 files changed, 40 insertions, 41 deletions
diff --git a/tools/testing/kunit/kunit_parser.py b/tools/testing/kunit/kunit_parser.py index 6614ec4d0898..e8bcc139702e 100644 --- a/tools/testing/kunit/kunit_parser.py +++ b/tools/testing/kunit/kunit_parser.py @@ -12,32 +12,32 @@ from collections import namedtuple from datetime import datetime from enum import Enum, auto from functools import reduce -from typing import List, Optional, Tuple +from typing import Iterable, Iterator, List, Optional, Tuple TestResult = namedtuple('TestResult', ['status','suites','log']) class TestSuite(object): - def __init__(self): - self.status = None - self.name = None - self.cases = [] + def __init__(self) -> None: + self.status = TestStatus.SUCCESS + self.name = '' + self.cases = [] # type: List[TestCase] - def __str__(self): - return 'TestSuite(' + self.status + ',' + self.name + ',' + str(self.cases) + ')' + def __str__(self) -> str: + return 'TestSuite(' + str(self.status) + ',' + self.name + ',' + str(self.cases) + ')' - def __repr__(self): + def __repr__(self) -> str: return str(self) class TestCase(object): - def __init__(self): - self.status = None + def __init__(self) -> None: + self.status = TestStatus.SUCCESS self.name = '' - self.log = [] + self.log = [] # type: List[str] - def __str__(self): - return 'TestCase(' + self.status + ',' + self.name + ',' + str(self.log) + ')' + def __str__(self) -> str: + return 'TestCase(' + str(self.status) + ',' + self.name + ',' + str(self.log) + ')' - def __repr__(self): + def __repr__(self) -> str: return str(self) class TestStatus(Enum): @@ -51,7 +51,7 @@ kunit_start_re = re.compile(r'TAP version [0-9]+$') kunit_end_re = re.compile('(List of all partitions:|' 'Kernel panic - not syncing: VFS:)') -def isolate_kunit_output(kernel_output): +def isolate_kunit_output(kernel_output) -> Iterator[str]: started = False for line in kernel_output: line = line.rstrip() # line always has a trailing \n @@ -64,7 +64,7 @@ def isolate_kunit_output(kernel_output): elif started: yield line[prefix_len:] if prefix_len > 0 else line -def raw_output(kernel_output): +def raw_output(kernel_output) -> None: for line in kernel_output: print(line.rstrip()) @@ -72,36 +72,36 @@ DIVIDER = '=' * 60 RESET = '\033[0;0m' -def red(text): +def red(text) -> str: return '\033[1;31m' + text + RESET -def yellow(text): +def yellow(text) -> str: return '\033[1;33m' + text + RESET -def green(text): +def green(text) -> str: return '\033[1;32m' + text + RESET -def print_with_timestamp(message): +def print_with_timestamp(message) -> None: print('[%s] %s' % (datetime.now().strftime('%H:%M:%S'), message)) -def format_suite_divider(message): +def format_suite_divider(message) -> str: return '======== ' + message + ' ========' -def print_suite_divider(message): +def print_suite_divider(message) -> None: print_with_timestamp(DIVIDER) print_with_timestamp(format_suite_divider(message)) -def print_log(log): +def print_log(log) -> None: for m in log: print_with_timestamp(m) TAP_ENTRIES = re.compile(r'^(TAP|[\s]*ok|[\s]*not ok|[\s]*[0-9]+\.\.[0-9]+|[\s]*#).*$') -def consume_non_diagnositic(lines: List[str]) -> None: +def consume_non_diagnostic(lines: List[str]) -> None: while lines and not TAP_ENTRIES.match(lines[0]): lines.pop(0) -def save_non_diagnositic(lines: List[str], test_case: TestCase) -> None: +def save_non_diagnostic(lines: List[str], test_case: TestCase) -> None: while lines and not TAP_ENTRIES.match(lines[0]): test_case.log.append(lines[0]) lines.pop(0) @@ -113,7 +113,7 @@ OK_NOT_OK_SUBTEST = re.compile(r'^[\s]+(ok|not ok) [0-9]+ - (.*)$') OK_NOT_OK_MODULE = re.compile(r'^(ok|not ok) ([0-9]+) - (.*)$') def parse_ok_not_ok_test_case(lines: List[str], test_case: TestCase) -> bool: - save_non_diagnositic(lines, test_case) + save_non_diagnostic(lines, test_case) if not lines: test_case.status = TestStatus.TEST_CRASHED return True @@ -139,7 +139,7 @@ SUBTEST_DIAGNOSTIC = re.compile(r'^[\s]+# (.*)$') DIAGNOSTIC_CRASH_MESSAGE = re.compile(r'^[\s]+# .*?: kunit test case crashed!$') def parse_diagnostic(lines: List[str], test_case: TestCase) -> bool: - save_non_diagnositic(lines, test_case) + save_non_diagnostic(lines, test_case) if not lines: return False line = lines[0] @@ -155,7 +155,7 @@ def parse_diagnostic(lines: List[str], test_case: TestCase) -> bool: def parse_test_case(lines: List[str]) -> Optional[TestCase]: test_case = TestCase() - save_non_diagnositic(lines, test_case) + save_non_diagnostic(lines, test_case) while parse_diagnostic(lines, test_case): pass if parse_ok_not_ok_test_case(lines, test_case): @@ -166,7 +166,7 @@ def parse_test_case(lines: List[str]) -> Optional[TestCase]: SUBTEST_HEADER = re.compile(r'^[\s]+# Subtest: (.*)$') def parse_subtest_header(lines: List[str]) -> Optional[str]: - consume_non_diagnositic(lines) + consume_non_diagnostic(lines) if not lines: return None match = SUBTEST_HEADER.match(lines[0]) @@ -179,7 +179,7 @@ def parse_subtest_header(lines: List[str]) -> Optional[str]: SUBTEST_PLAN = re.compile(r'[\s]+[0-9]+\.\.([0-9]+)') def parse_subtest_plan(lines: List[str]) -> Optional[int]: - consume_non_diagnositic(lines) + consume_non_diagnostic(lines) match = SUBTEST_PLAN.match(lines[0]) if match: lines.pop(0) @@ -202,7 +202,7 @@ def max_status(left: TestStatus, right: TestStatus) -> TestStatus: def parse_ok_not_ok_test_suite(lines: List[str], test_suite: TestSuite, expected_suite_index: int) -> bool: - consume_non_diagnositic(lines) + consume_non_diagnostic(lines) if not lines: test_suite.status = TestStatus.TEST_CRASHED return False @@ -224,18 +224,17 @@ def parse_ok_not_ok_test_suite(lines: List[str], else: return False -def bubble_up_errors(to_status, status_container_list) -> TestStatus: - status_list = map(to_status, status_container_list) - return reduce(max_status, status_list, TestStatus.SUCCESS) +def bubble_up_errors(statuses: Iterable[TestStatus]) -> TestStatus: + return reduce(max_status, statuses, TestStatus.SUCCESS) def bubble_up_test_case_errors(test_suite: TestSuite) -> TestStatus: - max_test_case_status = bubble_up_errors(lambda x: x.status, test_suite.cases) + max_test_case_status = bubble_up_errors(x.status for x in test_suite.cases) return max_status(max_test_case_status, test_suite.status) def parse_test_suite(lines: List[str], expected_suite_index: int) -> Optional[TestSuite]: if not lines: return None - consume_non_diagnositic(lines) + consume_non_diagnostic(lines) test_suite = TestSuite() test_suite.status = TestStatus.SUCCESS name = parse_subtest_header(lines) @@ -264,7 +263,7 @@ def parse_test_suite(lines: List[str], expected_suite_index: int) -> Optional[Te TAP_HEADER = re.compile(r'^TAP version 14$') def parse_tap_header(lines: List[str]) -> bool: - consume_non_diagnositic(lines) + consume_non_diagnostic(lines) if TAP_HEADER.match(lines[0]): lines.pop(0) return True @@ -274,7 +273,7 @@ def parse_tap_header(lines: List[str]) -> bool: TEST_PLAN = re.compile(r'[0-9]+\.\.([0-9]+)') def parse_test_plan(lines: List[str]) -> Optional[int]: - consume_non_diagnositic(lines) + consume_non_diagnostic(lines) match = TEST_PLAN.match(lines[0]) if match: lines.pop(0) @@ -282,11 +281,11 @@ def parse_test_plan(lines: List[str]) -> Optional[int]: else: return None -def bubble_up_suite_errors(test_suite_list: List[TestSuite]) -> TestStatus: - return bubble_up_errors(lambda x: x.status, test_suite_list) +def bubble_up_suite_errors(test_suites: Iterable[TestSuite]) -> TestStatus: + return bubble_up_errors(x.status for x in test_suites) def parse_test_result(lines: List[str]) -> TestResult: - consume_non_diagnositic(lines) + consume_non_diagnostic(lines) if not lines or not parse_tap_header(lines): return TestResult(TestStatus.NO_TESTS, [], lines) expected_test_suite_num = parse_test_plan(lines) |