1# SPDX-License-Identifier: GPL-2.0 2# 3# Parses test results from a kernel dmesg log. 4# 5# Copyright (C) 2019, Google LLC. 6# Author: Felix Guo <felixguoxiuping@gmail.com> 7# Author: Brendan Higgins <brendanhiggins@google.com> 8 9import re 10 11from collections import namedtuple 12from datetime import datetime 13from enum import Enum, auto 14from functools import reduce 15from typing import Iterable, Iterator, List, Optional, Tuple 16 17TestResult = namedtuple('TestResult', ['status','suites','log']) 18 19class TestSuite(object): 20 def __init__(self) -> None: 21 self.status = TestStatus.SUCCESS 22 self.name = '' 23 self.cases = [] # type: List[TestCase] 24 25 def __str__(self) -> str: 26 return 'TestSuite(' + str(self.status) + ',' + self.name + ',' + str(self.cases) + ')' 27 28 def __repr__(self) -> str: 29 return str(self) 30 31class TestCase(object): 32 def __init__(self) -> None: 33 self.status = TestStatus.SUCCESS 34 self.name = '' 35 self.log = [] # type: List[str] 36 37 def __str__(self) -> str: 38 return 'TestCase(' + str(self.status) + ',' + self.name + ',' + str(self.log) + ')' 39 40 def __repr__(self) -> str: 41 return str(self) 42 43class TestStatus(Enum): 44 SUCCESS = auto() 45 FAILURE = auto() 46 TEST_CRASHED = auto() 47 NO_TESTS = auto() 48 FAILURE_TO_PARSE_TESTS = auto() 49 50kunit_start_re = re.compile(r'TAP version [0-9]+$') 51kunit_end_re = re.compile('(List of all partitions:|' 52 'Kernel panic - not syncing: VFS:)') 53 54def isolate_kunit_output(kernel_output) -> Iterator[str]: 55 started = False 56 for line in kernel_output: 57 line = line.rstrip() # line always has a trailing \n 58 if kunit_start_re.search(line): 59 prefix_len = len(line.split('TAP version')[0]) 60 started = True 61 yield line[prefix_len:] if prefix_len > 0 else line 62 elif kunit_end_re.search(line): 63 break 64 elif started: 65 yield line[prefix_len:] if prefix_len > 0 else line 66 67def raw_output(kernel_output) -> None: 68 for line in kernel_output: 69 print(line.rstrip()) 70 71DIVIDER = '=' * 60 72 73RESET = '\033[0;0m' 74 75def red(text) -> str: 76 return '\033[1;31m' + text + RESET 77 78def yellow(text) -> str: 79 return '\033[1;33m' + text + RESET 80 81def green(text) -> str: 82 return '\033[1;32m' + text + RESET 83 84def print_with_timestamp(message) -> None: 85 print('[%s] %s' % (datetime.now().strftime('%H:%M:%S'), message)) 86 87def format_suite_divider(message) -> str: 88 return '======== ' + message + ' ========' 89 90def print_suite_divider(message) -> None: 91 print_with_timestamp(DIVIDER) 92 print_with_timestamp(format_suite_divider(message)) 93 94def print_log(log) -> None: 95 for m in log: 96 print_with_timestamp(m) 97 98TAP_ENTRIES = re.compile(r'^(TAP|[\s]*ok|[\s]*not ok|[\s]*[0-9]+\.\.[0-9]+|[\s]*#).*$') 99 100def consume_non_diagnostic(lines: List[str]) -> None: 101 while lines and not TAP_ENTRIES.match(lines[0]): 102 lines.pop(0) 103 104def save_non_diagnostic(lines: List[str], test_case: TestCase) -> None: 105 while lines and not TAP_ENTRIES.match(lines[0]): 106 test_case.log.append(lines[0]) 107 lines.pop(0) 108 109OkNotOkResult = namedtuple('OkNotOkResult', ['is_ok','description', 'text']) 110 111OK_NOT_OK_SUBTEST = re.compile(r'^[\s]+(ok|not ok) [0-9]+ - (.*)$') 112 113OK_NOT_OK_MODULE = re.compile(r'^(ok|not ok) ([0-9]+) - (.*)$') 114 115def parse_ok_not_ok_test_case(lines: List[str], test_case: TestCase) -> bool: 116 save_non_diagnostic(lines, test_case) 117 if not lines: 118 test_case.status = TestStatus.TEST_CRASHED 119 return True 120 line = lines[0] 121 match = OK_NOT_OK_SUBTEST.match(line) 122 while not match and lines: 123 line = lines.pop(0) 124 match = OK_NOT_OK_SUBTEST.match(line) 125 if match: 126 test_case.log.append(lines.pop(0)) 127 test_case.name = match.group(2) 128 if test_case.status == TestStatus.TEST_CRASHED: 129 return True 130 if match.group(1) == 'ok': 131 test_case.status = TestStatus.SUCCESS 132 else: 133 test_case.status = TestStatus.FAILURE 134 return True 135 else: 136 return False 137 138SUBTEST_DIAGNOSTIC = re.compile(r'^[\s]+# (.*)$') 139DIAGNOSTIC_CRASH_MESSAGE = re.compile(r'^[\s]+# .*?: kunit test case crashed!$') 140 141def parse_diagnostic(lines: List[str], test_case: TestCase) -> bool: 142 save_non_diagnostic(lines, test_case) 143 if not lines: 144 return False 145 line = lines[0] 146 match = SUBTEST_DIAGNOSTIC.match(line) 147 if match: 148 test_case.log.append(lines.pop(0)) 149 crash_match = DIAGNOSTIC_CRASH_MESSAGE.match(line) 150 if crash_match: 151 test_case.status = TestStatus.TEST_CRASHED 152 return True 153 else: 154 return False 155 156def parse_test_case(lines: List[str]) -> Optional[TestCase]: 157 test_case = TestCase() 158 save_non_diagnostic(lines, test_case) 159 while parse_diagnostic(lines, test_case): 160 pass 161 if parse_ok_not_ok_test_case(lines, test_case): 162 return test_case 163 else: 164 return None 165 166SUBTEST_HEADER = re.compile(r'^[\s]+# Subtest: (.*)$') 167 168def parse_subtest_header(lines: List[str]) -> Optional[str]: 169 consume_non_diagnostic(lines) 170 if not lines: 171 return None 172 match = SUBTEST_HEADER.match(lines[0]) 173 if match: 174 lines.pop(0) 175 return match.group(1) 176 else: 177 return None 178 179SUBTEST_PLAN = re.compile(r'[\s]+[0-9]+\.\.([0-9]+)') 180 181def parse_subtest_plan(lines: List[str]) -> Optional[int]: 182 consume_non_diagnostic(lines) 183 match = SUBTEST_PLAN.match(lines[0]) 184 if match: 185 lines.pop(0) 186 return int(match.group(1)) 187 else: 188 return None 189 190def max_status(left: TestStatus, right: TestStatus) -> TestStatus: 191 if left == TestStatus.TEST_CRASHED or right == TestStatus.TEST_CRASHED: 192 return TestStatus.TEST_CRASHED 193 elif left == TestStatus.FAILURE or right == TestStatus.FAILURE: 194 return TestStatus.FAILURE 195 elif left != TestStatus.SUCCESS: 196 return left 197 elif right != TestStatus.SUCCESS: 198 return right 199 else: 200 return TestStatus.SUCCESS 201 202def parse_ok_not_ok_test_suite(lines: List[str], 203 test_suite: TestSuite, 204 expected_suite_index: int) -> bool: 205 consume_non_diagnostic(lines) 206 if not lines: 207 test_suite.status = TestStatus.TEST_CRASHED 208 return False 209 line = lines[0] 210 match = OK_NOT_OK_MODULE.match(line) 211 if match: 212 lines.pop(0) 213 if match.group(1) == 'ok': 214 test_suite.status = TestStatus.SUCCESS 215 else: 216 test_suite.status = TestStatus.FAILURE 217 suite_index = int(match.group(2)) 218 if suite_index != expected_suite_index: 219 print_with_timestamp( 220 red('[ERROR] ') + 'expected_suite_index ' + 221 str(expected_suite_index) + ', but got ' + 222 str(suite_index)) 223 return True 224 else: 225 return False 226 227def bubble_up_errors(statuses: Iterable[TestStatus]) -> TestStatus: 228 return reduce(max_status, statuses, TestStatus.SUCCESS) 229 230def bubble_up_test_case_errors(test_suite: TestSuite) -> TestStatus: 231 max_test_case_status = bubble_up_errors(x.status for x in test_suite.cases) 232 return max_status(max_test_case_status, test_suite.status) 233 234def parse_test_suite(lines: List[str], expected_suite_index: int) -> Optional[TestSuite]: 235 if not lines: 236 return None 237 consume_non_diagnostic(lines) 238 test_suite = TestSuite() 239 test_suite.status = TestStatus.SUCCESS 240 name = parse_subtest_header(lines) 241 if not name: 242 return None 243 test_suite.name = name 244 expected_test_case_num = parse_subtest_plan(lines) 245 if expected_test_case_num is None: 246 return None 247 while expected_test_case_num > 0: 248 test_case = parse_test_case(lines) 249 if not test_case: 250 break 251 test_suite.cases.append(test_case) 252 expected_test_case_num -= 1 253 if parse_ok_not_ok_test_suite(lines, test_suite, expected_suite_index): 254 test_suite.status = bubble_up_test_case_errors(test_suite) 255 return test_suite 256 elif not lines: 257 print_with_timestamp(red('[ERROR] ') + 'ran out of lines before end token') 258 return test_suite 259 else: 260 print('failed to parse end of suite' + lines[0]) 261 return None 262 263TAP_HEADER = re.compile(r'^TAP version 14$') 264 265def parse_tap_header(lines: List[str]) -> bool: 266 consume_non_diagnostic(lines) 267 if TAP_HEADER.match(lines[0]): 268 lines.pop(0) 269 return True 270 else: 271 return False 272 273TEST_PLAN = re.compile(r'[0-9]+\.\.([0-9]+)') 274 275def parse_test_plan(lines: List[str]) -> Optional[int]: 276 consume_non_diagnostic(lines) 277 match = TEST_PLAN.match(lines[0]) 278 if match: 279 lines.pop(0) 280 return int(match.group(1)) 281 else: 282 return None 283 284def bubble_up_suite_errors(test_suites: Iterable[TestSuite]) -> TestStatus: 285 return bubble_up_errors(x.status for x in test_suites) 286 287def parse_test_result(lines: List[str]) -> TestResult: 288 consume_non_diagnostic(lines) 289 if not lines or not parse_tap_header(lines): 290 return TestResult(TestStatus.NO_TESTS, [], lines) 291 expected_test_suite_num = parse_test_plan(lines) 292 if not expected_test_suite_num: 293 return TestResult(TestStatus.FAILURE_TO_PARSE_TESTS, [], lines) 294 test_suites = [] 295 for i in range(1, expected_test_suite_num + 1): 296 test_suite = parse_test_suite(lines, i) 297 if test_suite: 298 test_suites.append(test_suite) 299 else: 300 print_with_timestamp( 301 red('[ERROR] ') + ' expected ' + 302 str(expected_test_suite_num) + 303 ' test suites, but got ' + str(i - 2)) 304 break 305 test_suite = parse_test_suite(lines, -1) 306 if test_suite: 307 print_with_timestamp(red('[ERROR] ') + 308 'got unexpected test suite: ' + test_suite.name) 309 if test_suites: 310 return TestResult(bubble_up_suite_errors(test_suites), test_suites, lines) 311 else: 312 return TestResult(TestStatus.NO_TESTS, [], lines) 313 314def print_and_count_results(test_result: TestResult) -> Tuple[int, int, int]: 315 total_tests = 0 316 failed_tests = 0 317 crashed_tests = 0 318 for test_suite in test_result.suites: 319 if test_suite.status == TestStatus.SUCCESS: 320 print_suite_divider(green('[PASSED] ') + test_suite.name) 321 elif test_suite.status == TestStatus.TEST_CRASHED: 322 print_suite_divider(red('[CRASHED] ' + test_suite.name)) 323 else: 324 print_suite_divider(red('[FAILED] ') + test_suite.name) 325 for test_case in test_suite.cases: 326 total_tests += 1 327 if test_case.status == TestStatus.SUCCESS: 328 print_with_timestamp(green('[PASSED] ') + test_case.name) 329 elif test_case.status == TestStatus.TEST_CRASHED: 330 crashed_tests += 1 331 print_with_timestamp(red('[CRASHED] ' + test_case.name)) 332 print_log(map(yellow, test_case.log)) 333 print_with_timestamp('') 334 else: 335 failed_tests += 1 336 print_with_timestamp(red('[FAILED] ') + test_case.name) 337 print_log(map(yellow, test_case.log)) 338 print_with_timestamp('') 339 return total_tests, failed_tests, crashed_tests 340 341def parse_run_tests(kernel_output) -> TestResult: 342 total_tests = 0 343 failed_tests = 0 344 crashed_tests = 0 345 test_result = parse_test_result(list(isolate_kunit_output(kernel_output))) 346 if test_result.status == TestStatus.NO_TESTS: 347 print(red('[ERROR] ') + yellow('no tests run!')) 348 elif test_result.status == TestStatus.FAILURE_TO_PARSE_TESTS: 349 print(red('[ERROR] ') + yellow('could not parse test results!')) 350 else: 351 (total_tests, 352 failed_tests, 353 crashed_tests) = print_and_count_results(test_result) 354 print_with_timestamp(DIVIDER) 355 fmt = green if test_result.status == TestStatus.SUCCESS else red 356 print_with_timestamp( 357 fmt('Testing complete. %d tests run. %d failed. %d crashed.' % 358 (total_tests, failed_tests, crashed_tests))) 359 return test_result 360