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 List 16 17TestResult = namedtuple('TestResult', ['status','suites','log']) 18 19class TestSuite(object): 20 def __init__(self): 21 self.status = None 22 self.name = None 23 self.cases = [] 24 25 def __str__(self): 26 return 'TestSuite(' + self.status + ',' + self.name + ',' + str(self.cases) + ')' 27 28 def __repr__(self): 29 return str(self) 30 31class TestCase(object): 32 def __init__(self): 33 self.status = None 34 self.name = '' 35 self.log = [] 36 37 def __str__(self): 38 return 'TestCase(' + self.status + ',' + self.name + ',' + str(self.log) + ')' 39 40 def __repr__(self): 41 return str(self) 42 43class TestStatus(Enum): 44 SUCCESS = auto() 45 FAILURE = auto() 46 TEST_CRASHED = auto() 47 NO_TESTS = auto() 48 49kunit_start_re = re.compile(r'TAP version [0-9]+$') 50kunit_end_re = re.compile('(List of all partitions:|' 51 'Kernel panic - not syncing: VFS:|reboot: System halted)') 52 53def isolate_kunit_output(kernel_output): 54 started = False 55 for line in kernel_output: 56 if kunit_start_re.search(line): 57 prefix_len = len(line.split('TAP version')[0]) 58 started = True 59 yield line[prefix_len:] if prefix_len > 0 else line 60 elif kunit_end_re.search(line): 61 break 62 elif started: 63 yield line[prefix_len:] if prefix_len > 0 else line 64 65def raw_output(kernel_output): 66 for line in kernel_output: 67 print(line) 68 yield line 69 70DIVIDER = '=' * 60 71 72RESET = '\033[0;0m' 73 74def red(text): 75 return '\033[1;31m' + text + RESET 76 77def yellow(text): 78 return '\033[1;33m' + text + RESET 79 80def green(text): 81 return '\033[1;32m' + text + RESET 82 83def print_with_timestamp(message): 84 print('[%s] %s' % (datetime.now().strftime('%H:%M:%S'), message)) 85 86def format_suite_divider(message): 87 return '======== ' + message + ' ========' 88 89def print_suite_divider(message): 90 print_with_timestamp(DIVIDER) 91 print_with_timestamp(format_suite_divider(message)) 92 93def print_log(log): 94 for m in log: 95 print_with_timestamp(m) 96 97TAP_ENTRIES = re.compile(r'^(TAP|[\s]*ok|[\s]*not ok|[\s]*[0-9]+\.\.[0-9]+|[\s]*#).*$') 98 99def consume_non_diagnositic(lines: List[str]) -> None: 100 while lines and not TAP_ENTRIES.match(lines[0]): 101 lines.pop(0) 102 103def save_non_diagnositic(lines: List[str], test_case: TestCase) -> None: 104 while lines and not TAP_ENTRIES.match(lines[0]): 105 test_case.log.append(lines[0]) 106 lines.pop(0) 107 108OkNotOkResult = namedtuple('OkNotOkResult', ['is_ok','description', 'text']) 109 110OK_NOT_OK_SUBTEST = re.compile(r'^[\s]+(ok|not ok) [0-9]+ - (.*)$') 111 112OK_NOT_OK_MODULE = re.compile(r'^(ok|not ok) [0-9]+ - (.*)$') 113 114def parse_ok_not_ok_test_case(lines: List[str], test_case: TestCase) -> bool: 115 save_non_diagnositic(lines, test_case) 116 if not lines: 117 test_case.status = TestStatus.TEST_CRASHED 118 return True 119 line = lines[0] 120 match = OK_NOT_OK_SUBTEST.match(line) 121 while not match and lines: 122 line = lines.pop(0) 123 match = OK_NOT_OK_SUBTEST.match(line) 124 if match: 125 test_case.log.append(lines.pop(0)) 126 test_case.name = match.group(2) 127 if test_case.status == TestStatus.TEST_CRASHED: 128 return True 129 if match.group(1) == 'ok': 130 test_case.status = TestStatus.SUCCESS 131 else: 132 test_case.status = TestStatus.FAILURE 133 return True 134 else: 135 return False 136 137SUBTEST_DIAGNOSTIC = re.compile(r'^[\s]+# .*?: (.*)$') 138DIAGNOSTIC_CRASH_MESSAGE = 'kunit test case crashed!' 139 140def parse_diagnostic(lines: List[str], test_case: TestCase) -> bool: 141 save_non_diagnositic(lines, test_case) 142 if not lines: 143 return False 144 line = lines[0] 145 match = SUBTEST_DIAGNOSTIC.match(line) 146 if match: 147 test_case.log.append(lines.pop(0)) 148 if match.group(1) == DIAGNOSTIC_CRASH_MESSAGE: 149 test_case.status = TestStatus.TEST_CRASHED 150 return True 151 else: 152 return False 153 154def parse_test_case(lines: List[str]) -> TestCase: 155 test_case = TestCase() 156 save_non_diagnositic(lines, test_case) 157 while parse_diagnostic(lines, test_case): 158 pass 159 if parse_ok_not_ok_test_case(lines, test_case): 160 return test_case 161 else: 162 return None 163 164SUBTEST_HEADER = re.compile(r'^[\s]+# Subtest: (.*)$') 165 166def parse_subtest_header(lines: List[str]) -> str: 167 consume_non_diagnositic(lines) 168 if not lines: 169 return None 170 match = SUBTEST_HEADER.match(lines[0]) 171 if match: 172 lines.pop(0) 173 return match.group(1) 174 else: 175 return None 176 177SUBTEST_PLAN = re.compile(r'[\s]+[0-9]+\.\.([0-9]+)') 178 179def parse_subtest_plan(lines: List[str]) -> int: 180 consume_non_diagnositic(lines) 181 match = SUBTEST_PLAN.match(lines[0]) 182 if match: 183 lines.pop(0) 184 return int(match.group(1)) 185 else: 186 return None 187 188def max_status(left: TestStatus, right: TestStatus) -> TestStatus: 189 if left == TestStatus.TEST_CRASHED or right == TestStatus.TEST_CRASHED: 190 return TestStatus.TEST_CRASHED 191 elif left == TestStatus.FAILURE or right == TestStatus.FAILURE: 192 return TestStatus.FAILURE 193 elif left != TestStatus.SUCCESS: 194 return left 195 elif right != TestStatus.SUCCESS: 196 return right 197 else: 198 return TestStatus.SUCCESS 199 200def parse_ok_not_ok_test_suite(lines: List[str], test_suite: TestSuite) -> bool: 201 consume_non_diagnositic(lines) 202 if not lines: 203 test_suite.status = TestStatus.TEST_CRASHED 204 return False 205 line = lines[0] 206 match = OK_NOT_OK_MODULE.match(line) 207 if match: 208 lines.pop(0) 209 if match.group(1) == 'ok': 210 test_suite.status = TestStatus.SUCCESS 211 else: 212 test_suite.status = TestStatus.FAILURE 213 return True 214 else: 215 return False 216 217def bubble_up_errors(to_status, status_container_list) -> TestStatus: 218 status_list = map(to_status, status_container_list) 219 return reduce(max_status, status_list, TestStatus.SUCCESS) 220 221def bubble_up_test_case_errors(test_suite: TestSuite) -> TestStatus: 222 max_test_case_status = bubble_up_errors(lambda x: x.status, test_suite.cases) 223 return max_status(max_test_case_status, test_suite.status) 224 225def parse_test_suite(lines: List[str]) -> TestSuite: 226 if not lines: 227 return None 228 consume_non_diagnositic(lines) 229 test_suite = TestSuite() 230 test_suite.status = TestStatus.SUCCESS 231 name = parse_subtest_header(lines) 232 if not name: 233 return None 234 test_suite.name = name 235 expected_test_case_num = parse_subtest_plan(lines) 236 if not expected_test_case_num: 237 return None 238 while expected_test_case_num > 0: 239 test_case = parse_test_case(lines) 240 if not test_case: 241 break 242 test_suite.cases.append(test_case) 243 expected_test_case_num -= 1 244 if parse_ok_not_ok_test_suite(lines, test_suite): 245 test_suite.status = bubble_up_test_case_errors(test_suite) 246 return test_suite 247 elif not lines: 248 print_with_timestamp(red('[ERROR] ') + 'ran out of lines before end token') 249 return test_suite 250 else: 251 print('failed to parse end of suite' + lines[0]) 252 return None 253 254TAP_HEADER = re.compile(r'^TAP version 14$') 255 256def parse_tap_header(lines: List[str]) -> bool: 257 consume_non_diagnositic(lines) 258 if TAP_HEADER.match(lines[0]): 259 lines.pop(0) 260 return True 261 else: 262 return False 263 264def bubble_up_suite_errors(test_suite_list: List[TestSuite]) -> TestStatus: 265 return bubble_up_errors(lambda x: x.status, test_suite_list) 266 267def parse_test_result(lines: List[str]) -> TestResult: 268 consume_non_diagnositic(lines) 269 if not lines or not parse_tap_header(lines): 270 return TestResult(TestStatus.NO_TESTS, [], lines) 271 test_suites = [] 272 test_suite = parse_test_suite(lines) 273 while test_suite: 274 test_suites.append(test_suite) 275 test_suite = parse_test_suite(lines) 276 return TestResult(bubble_up_suite_errors(test_suites), test_suites, lines) 277 278def parse_run_tests(kernel_output) -> TestResult: 279 total_tests = 0 280 failed_tests = 0 281 crashed_tests = 0 282 test_result = parse_test_result(list(isolate_kunit_output(kernel_output))) 283 if test_result.status == TestStatus.NO_TESTS: 284 print_with_timestamp(red('[ERROR] ') + 'no kunit output detected') 285 for test_suite in test_result.suites: 286 if test_suite.status == TestStatus.SUCCESS: 287 print_suite_divider(green('[PASSED] ') + test_suite.name) 288 elif test_suite.status == TestStatus.TEST_CRASHED: 289 print_suite_divider(red('[CRASHED] ' + test_suite.name)) 290 else: 291 print_suite_divider(red('[FAILED] ') + test_suite.name) 292 for test_case in test_suite.cases: 293 total_tests += 1 294 if test_case.status == TestStatus.SUCCESS: 295 print_with_timestamp(green('[PASSED] ') + test_case.name) 296 elif test_case.status == TestStatus.TEST_CRASHED: 297 crashed_tests += 1 298 print_with_timestamp(red('[CRASHED] ' + test_case.name)) 299 print_log(map(yellow, test_case.log)) 300 print_with_timestamp('') 301 else: 302 failed_tests += 1 303 print_with_timestamp(red('[FAILED] ') + test_case.name) 304 print_log(map(yellow, test_case.log)) 305 print_with_timestamp('') 306 print_with_timestamp(DIVIDER) 307 fmt = green if test_result.status == TestStatus.SUCCESS else red 308 print_with_timestamp( 309 fmt('Testing complete. %d tests run. %d failed. %d crashed.' % 310 (total_tests, failed_tests, crashed_tests))) 311 return test_result 312