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