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 SKIPPED = auto() 47 TEST_CRASHED = auto() 48 NO_TESTS = auto() 49 FAILURE_TO_PARSE_TESTS = auto() 50 51class LineStream: 52 """Provides a peek()/pop() interface over an iterator of (line#, text).""" 53 _lines: Iterator[Tuple[int, str]] 54 _next: Tuple[int, str] 55 _done: bool 56 57 def __init__(self, lines: Iterator[Tuple[int, str]]): 58 self._lines = lines 59 self._done = False 60 self._next = (0, '') 61 self._get_next() 62 63 def _get_next(self) -> None: 64 try: 65 self._next = next(self._lines) 66 except StopIteration: 67 self._done = True 68 69 def peek(self) -> str: 70 return self._next[1] 71 72 def pop(self) -> str: 73 n = self._next 74 self._get_next() 75 return n[1] 76 77 def __bool__(self) -> bool: 78 return not self._done 79 80 # Only used by kunit_tool_test.py. 81 def __iter__(self) -> Iterator[str]: 82 while bool(self): 83 yield self.pop() 84 85 def line_number(self) -> int: 86 return self._next[0] 87 88kunit_start_re = re.compile(r'TAP version [0-9]+$') 89kunit_end_re = re.compile('(List of all partitions:|' 90 'Kernel panic - not syncing: VFS:|reboot: System halted)') 91 92def extract_tap_lines(kernel_output: Iterable[str]) -> LineStream: 93 def isolate_kunit_output(kernel_output: Iterable[str]) -> Iterator[Tuple[int, str]]: 94 line_num = 0 95 started = False 96 for line in kernel_output: 97 line_num += 1 98 line = line.rstrip() # line always has a trailing \n 99 if kunit_start_re.search(line): 100 prefix_len = len(line.split('TAP version')[0]) 101 started = True 102 yield line_num, line[prefix_len:] 103 elif kunit_end_re.search(line): 104 break 105 elif started: 106 yield line_num, line[prefix_len:] 107 return LineStream(lines=isolate_kunit_output(kernel_output)) 108 109def raw_output(kernel_output) -> None: 110 for line in kernel_output: 111 print(line.rstrip()) 112 113DIVIDER = '=' * 60 114 115RESET = '\033[0;0m' 116 117def red(text) -> str: 118 return '\033[1;31m' + text + RESET 119 120def yellow(text) -> str: 121 return '\033[1;33m' + text + RESET 122 123def green(text) -> str: 124 return '\033[1;32m' + text + RESET 125 126def print_with_timestamp(message) -> None: 127 print('[%s] %s' % (datetime.now().strftime('%H:%M:%S'), message)) 128 129def format_suite_divider(message) -> str: 130 return '======== ' + message + ' ========' 131 132def print_suite_divider(message) -> None: 133 print_with_timestamp(DIVIDER) 134 print_with_timestamp(format_suite_divider(message)) 135 136def print_log(log) -> None: 137 for m in log: 138 print_with_timestamp(m) 139 140TAP_ENTRIES = re.compile(r'^(TAP|[\s]*ok|[\s]*not ok|[\s]*[0-9]+\.\.[0-9]+|[\s]*#).*$') 141 142def consume_non_diagnostic(lines: LineStream) -> None: 143 while lines and not TAP_ENTRIES.match(lines.peek()): 144 lines.pop() 145 146def save_non_diagnostic(lines: LineStream, test_case: TestCase) -> None: 147 while lines and not TAP_ENTRIES.match(lines.peek()): 148 test_case.log.append(lines.peek()) 149 lines.pop() 150 151OkNotOkResult = namedtuple('OkNotOkResult', ['is_ok','description', 'text']) 152 153OK_NOT_OK_SKIP = re.compile(r'^[\s]*(ok|not ok) [0-9]+ - (.*) # SKIP(.*)$') 154 155OK_NOT_OK_SUBTEST = re.compile(r'^[\s]+(ok|not ok) [0-9]+ - (.*)$') 156 157OK_NOT_OK_MODULE = re.compile(r'^(ok|not ok) ([0-9]+) - (.*)$') 158 159def parse_ok_not_ok_test_case(lines: LineStream, test_case: TestCase) -> bool: 160 save_non_diagnostic(lines, test_case) 161 if not lines: 162 test_case.status = TestStatus.TEST_CRASHED 163 return True 164 line = lines.peek() 165 match = OK_NOT_OK_SUBTEST.match(line) 166 while not match and lines: 167 line = lines.pop() 168 match = OK_NOT_OK_SUBTEST.match(line) 169 if match: 170 test_case.log.append(lines.pop()) 171 test_case.name = match.group(2) 172 skip_match = OK_NOT_OK_SKIP.match(line) 173 if skip_match: 174 test_case.status = TestStatus.SKIPPED 175 return True 176 if test_case.status == TestStatus.TEST_CRASHED: 177 return True 178 if match.group(1) == 'ok': 179 test_case.status = TestStatus.SUCCESS 180 else: 181 test_case.status = TestStatus.FAILURE 182 return True 183 else: 184 return False 185 186SUBTEST_DIAGNOSTIC = re.compile(r'^[\s]+# (.*)$') 187DIAGNOSTIC_CRASH_MESSAGE = re.compile(r'^[\s]+# .*?: kunit test case crashed!$') 188 189def parse_diagnostic(lines: LineStream, test_case: TestCase) -> bool: 190 save_non_diagnostic(lines, test_case) 191 if not lines: 192 return False 193 line = lines.peek() 194 match = SUBTEST_DIAGNOSTIC.match(line) 195 if match: 196 test_case.log.append(lines.pop()) 197 crash_match = DIAGNOSTIC_CRASH_MESSAGE.match(line) 198 if crash_match: 199 test_case.status = TestStatus.TEST_CRASHED 200 return True 201 else: 202 return False 203 204def parse_test_case(lines: LineStream) -> Optional[TestCase]: 205 test_case = TestCase() 206 save_non_diagnostic(lines, test_case) 207 while parse_diagnostic(lines, test_case): 208 pass 209 if parse_ok_not_ok_test_case(lines, test_case): 210 return test_case 211 else: 212 return None 213 214SUBTEST_HEADER = re.compile(r'^[\s]+# Subtest: (.*)$') 215 216def parse_subtest_header(lines: LineStream) -> Optional[str]: 217 consume_non_diagnostic(lines) 218 if not lines: 219 return None 220 match = SUBTEST_HEADER.match(lines.peek()) 221 if match: 222 lines.pop() 223 return match.group(1) 224 else: 225 return None 226 227SUBTEST_PLAN = re.compile(r'[\s]+[0-9]+\.\.([0-9]+)') 228 229def parse_subtest_plan(lines: LineStream) -> Optional[int]: 230 consume_non_diagnostic(lines) 231 match = SUBTEST_PLAN.match(lines.peek()) 232 if match: 233 lines.pop() 234 return int(match.group(1)) 235 else: 236 return None 237 238def max_status(left: TestStatus, right: TestStatus) -> TestStatus: 239 if left == right: 240 return left 241 elif left == TestStatus.TEST_CRASHED or right == TestStatus.TEST_CRASHED: 242 return TestStatus.TEST_CRASHED 243 elif left == TestStatus.FAILURE or right == TestStatus.FAILURE: 244 return TestStatus.FAILURE 245 elif left == TestStatus.SKIPPED: 246 return right 247 else: 248 return left 249 250def parse_ok_not_ok_test_suite(lines: LineStream, 251 test_suite: TestSuite, 252 expected_suite_index: int) -> bool: 253 consume_non_diagnostic(lines) 254 if not lines: 255 test_suite.status = TestStatus.TEST_CRASHED 256 return False 257 line = lines.peek() 258 match = OK_NOT_OK_MODULE.match(line) 259 if match: 260 lines.pop() 261 if match.group(1) == 'ok': 262 test_suite.status = TestStatus.SUCCESS 263 else: 264 test_suite.status = TestStatus.FAILURE 265 skip_match = OK_NOT_OK_SKIP.match(line) 266 if skip_match: 267 test_suite.status = TestStatus.SKIPPED 268 suite_index = int(match.group(2)) 269 if suite_index != expected_suite_index: 270 print_with_timestamp( 271 red('[ERROR] ') + 'expected_suite_index ' + 272 str(expected_suite_index) + ', but got ' + 273 str(suite_index)) 274 return True 275 else: 276 return False 277 278def bubble_up_errors(status_list: Iterable[TestStatus]) -> TestStatus: 279 return reduce(max_status, status_list, TestStatus.SKIPPED) 280 281def bubble_up_test_case_errors(test_suite: TestSuite) -> TestStatus: 282 max_test_case_status = bubble_up_errors(x.status for x in test_suite.cases) 283 return max_status(max_test_case_status, test_suite.status) 284 285def parse_test_suite(lines: LineStream, expected_suite_index: int) -> Optional[TestSuite]: 286 if not lines: 287 return None 288 consume_non_diagnostic(lines) 289 test_suite = TestSuite() 290 test_suite.status = TestStatus.SUCCESS 291 name = parse_subtest_header(lines) 292 if not name: 293 return None 294 test_suite.name = name 295 expected_test_case_num = parse_subtest_plan(lines) 296 if expected_test_case_num is None: 297 return None 298 while expected_test_case_num > 0: 299 test_case = parse_test_case(lines) 300 if not test_case: 301 break 302 test_suite.cases.append(test_case) 303 expected_test_case_num -= 1 304 if parse_ok_not_ok_test_suite(lines, test_suite, expected_suite_index): 305 test_suite.status = bubble_up_test_case_errors(test_suite) 306 return test_suite 307 elif not lines: 308 print_with_timestamp(red('[ERROR] ') + 'ran out of lines before end token') 309 return test_suite 310 else: 311 print(f'failed to parse end of suite "{name}", at line {lines.line_number()}: {lines.peek()}') 312 return None 313 314TAP_HEADER = re.compile(r'^TAP version 14$') 315 316def parse_tap_header(lines: LineStream) -> bool: 317 consume_non_diagnostic(lines) 318 if TAP_HEADER.match(lines.peek()): 319 lines.pop() 320 return True 321 else: 322 return False 323 324TEST_PLAN = re.compile(r'[0-9]+\.\.([0-9]+)') 325 326def parse_test_plan(lines: LineStream) -> Optional[int]: 327 consume_non_diagnostic(lines) 328 match = TEST_PLAN.match(lines.peek()) 329 if match: 330 lines.pop() 331 return int(match.group(1)) 332 else: 333 return None 334 335def bubble_up_suite_errors(test_suites: Iterable[TestSuite]) -> TestStatus: 336 return bubble_up_errors(x.status for x in test_suites) 337 338def parse_test_result(lines: LineStream) -> TestResult: 339 consume_non_diagnostic(lines) 340 if not lines or not parse_tap_header(lines): 341 return TestResult(TestStatus.FAILURE_TO_PARSE_TESTS, [], lines) 342 expected_test_suite_num = parse_test_plan(lines) 343 if expected_test_suite_num == 0: 344 return TestResult(TestStatus.NO_TESTS, [], lines) 345 elif expected_test_suite_num is None: 346 return TestResult(TestStatus.FAILURE_TO_PARSE_TESTS, [], lines) 347 test_suites = [] 348 for i in range(1, expected_test_suite_num + 1): 349 test_suite = parse_test_suite(lines, i) 350 if test_suite: 351 test_suites.append(test_suite) 352 else: 353 print_with_timestamp( 354 red('[ERROR] ') + ' expected ' + 355 str(expected_test_suite_num) + 356 ' test suites, but got ' + str(i - 2)) 357 break 358 test_suite = parse_test_suite(lines, -1) 359 if test_suite: 360 print_with_timestamp(red('[ERROR] ') + 361 'got unexpected test suite: ' + test_suite.name) 362 if test_suites: 363 return TestResult(bubble_up_suite_errors(test_suites), test_suites, lines) 364 else: 365 return TestResult(TestStatus.NO_TESTS, [], lines) 366 367class TestCounts: 368 passed: int 369 failed: int 370 crashed: int 371 skipped: int 372 373 def __init__(self): 374 self.passed = 0 375 self.failed = 0 376 self.crashed = 0 377 self.skipped = 0 378 379 def total(self) -> int: 380 return self.passed + self.failed + self.crashed + self.skipped 381 382def print_and_count_results(test_result: TestResult) -> TestCounts: 383 counts = TestCounts() 384 for test_suite in test_result.suites: 385 if test_suite.status == TestStatus.SUCCESS: 386 print_suite_divider(green('[PASSED] ') + test_suite.name) 387 elif test_suite.status == TestStatus.SKIPPED: 388 print_suite_divider(yellow('[SKIPPED] ') + test_suite.name) 389 elif test_suite.status == TestStatus.TEST_CRASHED: 390 print_suite_divider(red('[CRASHED] ' + test_suite.name)) 391 else: 392 print_suite_divider(red('[FAILED] ') + test_suite.name) 393 for test_case in test_suite.cases: 394 if test_case.status == TestStatus.SUCCESS: 395 counts.passed += 1 396 print_with_timestamp(green('[PASSED] ') + test_case.name) 397 elif test_case.status == TestStatus.SKIPPED: 398 counts.skipped += 1 399 print_with_timestamp(yellow('[SKIPPED] ') + test_case.name) 400 elif test_case.status == TestStatus.TEST_CRASHED: 401 counts.crashed += 1 402 print_with_timestamp(red('[CRASHED] ' + test_case.name)) 403 print_log(map(yellow, test_case.log)) 404 print_with_timestamp('') 405 else: 406 counts.failed += 1 407 print_with_timestamp(red('[FAILED] ') + test_case.name) 408 print_log(map(yellow, test_case.log)) 409 print_with_timestamp('') 410 return counts 411 412def parse_run_tests(kernel_output: Iterable[str]) -> TestResult: 413 counts = TestCounts() 414 lines = extract_tap_lines(kernel_output) 415 test_result = parse_test_result(lines) 416 if test_result.status == TestStatus.NO_TESTS: 417 print(red('[ERROR] ') + yellow('no tests run!')) 418 elif test_result.status == TestStatus.FAILURE_TO_PARSE_TESTS: 419 print(red('[ERROR] ') + yellow('could not parse test results!')) 420 else: 421 counts = print_and_count_results(test_result) 422 print_with_timestamp(DIVIDER) 423 if test_result.status == TestStatus.SUCCESS: 424 fmt = green 425 elif test_result.status == TestStatus.SKIPPED: 426 fmt = yellow 427 else: 428 fmt =red 429 print_with_timestamp( 430 fmt('Testing complete. %d tests run. %d failed. %d crashed. %d skipped.' % 431 (counts.total(), counts.failed, counts.crashed, counts.skipped))) 432 return test_result 433