tools/testing/kunit/kunit_parser.py - linux - Git at Google

 # SPDX-License-Identifier: GPL-2.0
 #
 # Parses test results from a kernel dmesg log.
 #
 # Copyright (C) 2019, Google LLC.
 # Author: Felix Guo <felixguoxiuping@gmail.com>
 # Author: Brendan Higgins <brendanhiggins@google.com>

 import re

 from collections import namedtuple
 from datetime import datetime
 from enum import Enum, auto
 from functools import reduce
 from typing import List

 TestResult = namedtuple('TestResult', ['status','modules','log'])

 class TestModule(object):
 	def __init__(self):
 		self.status = None
 		self.name = None
 		self.cases = []

 	def __str__(self):
 		return 'TestModule(' + self.status + ',' + self.name + ',' + str(self.cases) + ')'

 	def __repr__(self):
 		return str(self)

 class TestCase(object):
 	def __init__(self):
 		self.status = None
 		self.name = ''
 		self.log = []

 	def __str__(self):
 		return 'TestCase(' + self.status + ',' + self.name + ',' + str(self.log) + ')'

 	def __repr__(self):
 		return str(self)

 class TestStatus(Enum):
 	SUCCESS = auto()
 	FAILURE = auto()
 	TEST_CRASHED = auto()
 	NO_TESTS = auto()

 kunit_start_re = re.compile(r'^TAP version [0-9]+$')
 kunit_end_re = re.compile('List of all partitions:')

 def isolate_kunit_output(kernel_output):
 	started = False
 	for line in kernel_output:
 		if kunit_start_re.match(line):
 			started = True
 			yield line
 		elif kunit_end_re.match(line):
 			break
 		elif started:
 			yield line

 def raw_output(kernel_output):
 	for line in kernel_output:
 		print(line)

 DIVIDER = '=' * 60

 RESET = '\033[0;0m'

 def red(text):
 	return '\033[1;31m' + text + RESET

 def yellow(text):
 	return '\033[1;33m' + text + RESET

 def green(text):
 	return '\033[1;32m' + text + RESET

 def print_with_timestamp(message):
 	print('[%s] %s' % (datetime.now().strftime('%H:%M:%S'), message))

 def format_module_divider(message):
 	return '======== ' + message + ' ========'

 def print_module_divider(message):
 	print_with_timestamp(DIVIDER)
 	print_with_timestamp(format_module_divider(message))

 def print_log(log):
 	for m in log:
 		print_with_timestamp(m)

 TAP_ENTRIES = re.compile(r'^(TAP|\t?ok|\t?not ok|\t?[0-9]+\.\.[0-9]+|\t?#).*$')

 def consume_non_diagnositic(lines: List[str]) -> None:
 	while not TAP_ENTRIES.match(lines[0]):
 		lines.pop(0)

 def save_non_diagnositic(lines: List[str], test_case: TestCase) -> None:
 	while not TAP_ENTRIES.match(lines[0]):
 		test_case.log.append(lines[0])
 		lines.pop(0)

 OkNotOkResult = namedtuple('OkNotOkResult', ['is_ok','description', 'text'])

 OK_NOT_OK_SUBTEST = re.compile(r'^\t(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)
 	line = lines[0]
 	match = OK_NOT_OK_SUBTEST.match(line)
 	if match:
 		test_case.log.append(lines.pop(0))
 		test_case.name = match.group(2)
 		if test_case.status == TestStatus.TEST_CRASHED:
 			return True
 		if match.group(1) == 'ok':
 			test_case.status = TestStatus.SUCCESS
 		else:
 			test_case.status = TestStatus.FAILURE
 		return True
 	else:
 		return False

 SUBTEST_DIAGNOSTIC = re.compile(r'^\t# .*?: (.*)$')
 DIAGNOSTIC_CRASH_MESSAGE = 'kunit test case crashed!'

 def parse_diagnostic(lines: List[str], test_case: TestCase) -> bool:
 	save_non_diagnositic(lines, test_case)
 	line = lines[0]
 	match = SUBTEST_DIAGNOSTIC.match(line)
 	if match:
 		test_case.log.append(lines.pop(0))
 		if match.group(1) == DIAGNOSTIC_CRASH_MESSAGE:
 			test_case.status = TestStatus.TEST_CRASHED
 		return True
 	else:
 		return False

 def parse_test_case(lines: List[str]) -> TestCase:
 	test_case = TestCase()
 	save_non_diagnositic(lines, test_case)
 	while parse_diagnostic(lines, test_case):
 		pass
 	if parse_ok_not_ok_test_case(lines, test_case):
 		return test_case
 	else:
 		return None

 SUBTEST_HEADER = re.compile(r'^\t# Subtest: (.*)$')

 def parse_subtest_header(lines: List[str]) -> str:
 	consume_non_diagnositic(lines)
 	match = SUBTEST_HEADER.match(lines[0])
 	if match:
 		lines.pop(0)
 		return match.group(1)
 	else:
 		return None

 SUBTEST_PLAN = re.compile(r'\t[0-9]+\.\.([0-9]+)')

 def parse_subtest_plan(lines: List[str]) -> int:
 	consume_non_diagnositic(lines)
 	match = SUBTEST_PLAN.match(lines[0])
 	if match:
 		lines.pop(0)
 		return match.group(1)
 	else:
 		return None

 def max_status(left: TestStatus, right: TestStatus) -> TestStatus:
 	if left == TestStatus.TEST_CRASHED or right == TestStatus.TEST_CRASHED:
 		return TestStatus.TEST_CRASHED
 	elif left == TestStatus.FAILURE or right == TestStatus.FAILURE:
 		return TestStatus.FAILURE
 	elif left != TestStatus.SUCCESS:
 		return left
 	elif right != TestStatus.SUCCESS:
 		return right
 	else:
 		return TestStatus.SUCCESS

 def parse_ok_not_ok_test_module(lines: List[str], test_module: TestModule) -> bool:
 	consume_non_diagnositic(lines)
 	line = lines[0]
 	match = OK_NOT_OK_MODULE.match(line)
 	if match:
 		lines.pop(0)
 		if match.group(1) == 'ok':
 			test_module.status = TestStatus.SUCCESS
 		else:
 			test_module.status = TestStatus.FAILURE
 		return True
 	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_test_case_errors(test_module: TestModule) -> TestStatus:
 	max_test_case_status = bubble_up_errors(lambda x: x.status, test_module.cases)
 	return max_status(max_test_case_status, test_module.status)

 def parse_test_module(lines: List[str]) -> TestModule:
 	if not lines:
 		return None
 	consume_non_diagnositic(lines)
 	test_module = TestModule()
 	test_module.status = TestStatus.SUCCESS
 	name = parse_subtest_header(lines)
 	if not name:
 		return None
 	test_module.name = name
 	test_case_num = parse_subtest_plan(lines)
 	if not test_case_num:
 		return None
 	test_case = parse_test_case(lines)
 	while test_case:
 		test_module.cases.append(test_case)
 		test_case = parse_test_case(lines)
 	if parse_ok_not_ok_test_module(lines, test_module):
 		test_module.status = bubble_up_test_case_errors(test_module)
 		return test_module
 	else:
 		print('failed to parse end of module' + lines[0])
 		return None

 TAP_HEADER = re.compile(r'^TAP version 14$')

 def parse_tap_header(lines: List[str]) -> bool:
 	consume_non_diagnositic(lines)
 	if TAP_HEADER.match(lines[0]):
 		lines.pop(0)
 		return True
 	else:
 		return False

 def bubble_up_module_errors(test_module_list: List[TestModule]) -> TestStatus:
 	return bubble_up_errors(lambda x: x.status, test_module_list)

 def parse_test_result(lines: List[str]) -> TestResult:
 	if not lines:
 		return TestResult(TestStatus.NO_TESTS, [], lines)
 	consume_non_diagnositic(lines)
 	if not parse_tap_header(lines):
 		return None
 	test_modules = []
 	test_module = parse_test_module(lines)
 	while test_module:
 		test_modules.append(test_module)
 		test_module = parse_test_module(lines)
 	return TestResult(bubble_up_module_errors(test_modules), test_modules, lines)

 def parse_run_tests(kernel_output) -> TestResult:
 	total_tests = 0
 	failed_tests = 0
 	crashed_tests = 0
 	test_result = parse_test_result(list(isolate_kunit_output(kernel_output)))
 	for test_module in test_result.modules:
 		if test_module.status == TestStatus.SUCCESS:
 			print_module_divider(green('[PASSED] ') + test_module.name)
 		elif test_module.status == TestStatus.TEST_CRASHED:
 			print_module_divider(red('[CRASHED] ' + test_module.name))
 		else:
 			print_module_divider(red('[FAILED] ') + test_module.name)
 		for test_case in test_module.cases:
 			total_tests += 1
 			if test_case.status == TestStatus.SUCCESS:
 				print_with_timestamp(green('[PASSED] ') + test_case.name)
 			elif test_case.status == TestStatus.TEST_CRASHED:
 				crashed_tests += 1
 				print_with_timestamp(red('[CRASHED] ' + test_case.name))
 				print_log(map(yellow, test_case.log))
 				print_with_timestamp('')
 			else:
 				failed_tests += 1
 				print_with_timestamp(red('[FAILED] ') + test_case.name)
 				print_log(map(yellow, test_case.log))
 				print_with_timestamp('')
 	print_with_timestamp(DIVIDER)
 	fmt = green if test_result.status == TestStatus.SUCCESS else red
 	print_with_timestamp(
 		fmt('Testing complete. %d tests run. %d failed. %d crashed.' %
 		    (total_tests, failed_tests, crashed_tests)))
 	return test_result
	# SPDX-License-Identifier: GPL-2.0
	#
	# Parses test results from a kernel dmesg log.
	#
	# Copyright (C) 2019, Google LLC.
	# Author: Felix Guo <felixguoxiuping@gmail.com>
	# Author: Brendan Higgins <brendanhiggins@google.com>

	import re

	from collections import namedtuple
	from datetime import datetime
	from enum import Enum, auto
	from functools import reduce
	from typing import List

	TestResult = namedtuple('TestResult', ['status','modules','log'])

	class TestModule(object):
	def __init__(self):
	self.status = None
	self.name = None
	self.cases = []

	def __str__(self):
	return 'TestModule(' + self.status + ',' + self.name + ',' + str(self.cases) + ')'

	def __repr__(self):
	return str(self)

	class TestCase(object):
	def __init__(self):
	self.status = None
	self.name = ''
	self.log = []

	def __str__(self):
	return 'TestCase(' + self.status + ',' + self.name + ',' + str(self.log) + ')'

	def __repr__(self):
	return str(self)

	class TestStatus(Enum):
	SUCCESS = auto()
	FAILURE = auto()
	TEST_CRASHED = auto()
	NO_TESTS = auto()

	kunit_start_re = re.compile(r'^TAP version [0-9]+$')
	kunit_end_re = re.compile('List of all partitions:')

	def isolate_kunit_output(kernel_output):
	started = False
	for line in kernel_output:
	if kunit_start_re.match(line):
	started = True
	yield line
	elif kunit_end_re.match(line):
	break
	elif started:
	yield line

	def raw_output(kernel_output):
	for line in kernel_output:
	print(line)

	DIVIDER = '=' * 60

	RESET = '\033[0;0m'

	def red(text):
	return '\033[1;31m' + text + RESET

	def yellow(text):
	return '\033[1;33m' + text + RESET

	def green(text):
	return '\033[1;32m' + text + RESET

	def print_with_timestamp(message):
	print('[%s] %s' % (datetime.now().strftime('%H:%M:%S'), message))

	def format_module_divider(message):
	return '======== ' + message + ' ========'

	def print_module_divider(message):
	print_with_timestamp(DIVIDER)
	print_with_timestamp(format_module_divider(message))

	def print_log(log):
	for m in log:
	print_with_timestamp(m)

	TAP_ENTRIES = re.compile(r'^(TAP\|\t?ok\|\t?not ok\|\t?[0-9]+\.\.[0-9]+\|\t?#).*$')

	def consume_non_diagnositic(lines: List[str]) -> None:
	while not TAP_ENTRIES.match(lines[0]):
	lines.pop(0)

	def save_non_diagnositic(lines: List[str], test_case: TestCase) -> None:
	while not TAP_ENTRIES.match(lines[0]):
	test_case.log.append(lines[0])
	lines.pop(0)

	OkNotOkResult = namedtuple('OkNotOkResult', ['is_ok','description', 'text'])

	OK_NOT_OK_SUBTEST = re.compile(r'^\t(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)
	line = lines[0]
	match = OK_NOT_OK_SUBTEST.match(line)
	if match:
	test_case.log.append(lines.pop(0))
	test_case.name = match.group(2)
	if test_case.status == TestStatus.TEST_CRASHED:
	return True
	if match.group(1) == 'ok':
	test_case.status = TestStatus.SUCCESS
	else:
	test_case.status = TestStatus.FAILURE
	return True
	else:
	return False

	SUBTEST_DIAGNOSTIC = re.compile(r'^\t# .?: (.)$')
	DIAGNOSTIC_CRASH_MESSAGE = 'kunit test case crashed!'

	def parse_diagnostic(lines: List[str], test_case: TestCase) -> bool:
	save_non_diagnositic(lines, test_case)
	line = lines[0]
	match = SUBTEST_DIAGNOSTIC.match(line)
	if match:
	test_case.log.append(lines.pop(0))
	if match.group(1) == DIAGNOSTIC_CRASH_MESSAGE:
	test_case.status = TestStatus.TEST_CRASHED
	return True
	else:
	return False

	def parse_test_case(lines: List[str]) -> TestCase:
	test_case = TestCase()
	save_non_diagnositic(lines, test_case)
	while parse_diagnostic(lines, test_case):
	pass
	if parse_ok_not_ok_test_case(lines, test_case):
	return test_case
	else:
	return None

	SUBTEST_HEADER = re.compile(r'^\t# Subtest: (.*)$')

	def parse_subtest_header(lines: List[str]) -> str:
	consume_non_diagnositic(lines)
	match = SUBTEST_HEADER.match(lines[0])
	if match:
	lines.pop(0)
	return match.group(1)
	else:
	return None

	SUBTEST_PLAN = re.compile(r'\t[0-9]+\.\.([0-9]+)')

	def parse_subtest_plan(lines: List[str]) -> int:
	consume_non_diagnositic(lines)
	match = SUBTEST_PLAN.match(lines[0])
	if match:
	lines.pop(0)
	return match.group(1)
	else:
	return None

	def max_status(left: TestStatus, right: TestStatus) -> TestStatus:
	if left == TestStatus.TEST_CRASHED or right == TestStatus.TEST_CRASHED:
	return TestStatus.TEST_CRASHED
	elif left == TestStatus.FAILURE or right == TestStatus.FAILURE:
	return TestStatus.FAILURE
	elif left != TestStatus.SUCCESS:
	return left
	elif right != TestStatus.SUCCESS:
	return right
	else:
	return TestStatus.SUCCESS

	def parse_ok_not_ok_test_module(lines: List[str], test_module: TestModule) -> bool:
	consume_non_diagnositic(lines)
	line = lines[0]
	match = OK_NOT_OK_MODULE.match(line)
	if match:
	lines.pop(0)
	if match.group(1) == 'ok':
	test_module.status = TestStatus.SUCCESS
	else:
	test_module.status = TestStatus.FAILURE
	return True
	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_test_case_errors(test_module: TestModule) -> TestStatus:
	max_test_case_status = bubble_up_errors(lambda x: x.status, test_module.cases)
	return max_status(max_test_case_status, test_module.status)

	def parse_test_module(lines: List[str]) -> TestModule:
	if not lines:
	return None
	consume_non_diagnositic(lines)
	test_module = TestModule()
	test_module.status = TestStatus.SUCCESS
	name = parse_subtest_header(lines)
	if not name:
	return None
	test_module.name = name
	test_case_num = parse_subtest_plan(lines)
	if not test_case_num:
	return None
	test_case = parse_test_case(lines)
	while test_case:
	test_module.cases.append(test_case)
	test_case = parse_test_case(lines)
	if parse_ok_not_ok_test_module(lines, test_module):
	test_module.status = bubble_up_test_case_errors(test_module)
	return test_module
	else:
	print('failed to parse end of module' + lines[0])
	return None

	TAP_HEADER = re.compile(r'^TAP version 14$')

	def parse_tap_header(lines: List[str]) -> bool:
	consume_non_diagnositic(lines)
	if TAP_HEADER.match(lines[0]):
	lines.pop(0)
	return True
	else:
	return False

	def bubble_up_module_errors(test_module_list: List[TestModule]) -> TestStatus:
	return bubble_up_errors(lambda x: x.status, test_module_list)

	def parse_test_result(lines: List[str]) -> TestResult:
	if not lines:
	return TestResult(TestStatus.NO_TESTS, [], lines)
	consume_non_diagnositic(lines)
	if not parse_tap_header(lines):
	return None
	test_modules = []
	test_module = parse_test_module(lines)
	while test_module:
	test_modules.append(test_module)
	test_module = parse_test_module(lines)
	return TestResult(bubble_up_module_errors(test_modules), test_modules, lines)

	def parse_run_tests(kernel_output) -> TestResult:
	total_tests = 0
	failed_tests = 0
	crashed_tests = 0
	test_result = parse_test_result(list(isolate_kunit_output(kernel_output)))
	for test_module in test_result.modules:
	if test_module.status == TestStatus.SUCCESS:
	print_module_divider(green('[PASSED] ') + test_module.name)
	elif test_module.status == TestStatus.TEST_CRASHED:
	print_module_divider(red('[CRASHED] ' + test_module.name))
	else:
	print_module_divider(red('[FAILED] ') + test_module.name)
	for test_case in test_module.cases:
	total_tests += 1
	if test_case.status == TestStatus.SUCCESS:
	print_with_timestamp(green('[PASSED] ') + test_case.name)
	elif test_case.status == TestStatus.TEST_CRASHED:
	crashed_tests += 1
	print_with_timestamp(red('[CRASHED] ' + test_case.name))
	print_log(map(yellow, test_case.log))
	print_with_timestamp('')
	else:
	failed_tests += 1
	print_with_timestamp(red('[FAILED] ') + test_case.name)
	print_log(map(yellow, test_case.log))
	print_with_timestamp('')
	print_with_timestamp(DIVIDER)
	fmt = green if test_result.status == TestStatus.SUCCESS else red
	print_with_timestamp(
	fmt('Testing complete. %d tests run. %d failed. %d crashed.' %
	(total_tests, failed_tests, crashed_tests)))
	return test_result