# SPDX-License-Identifier: GPL-2.0## Parses KTAP test results from a kernel dmesg log and incrementally prints# results with reader-friendly format. Stores and returns test results in a# Test object.## Copyright (C) 2019, Google LLC.# Author: Felix Guo <felixguoxiuping@gmail.com># Author: Brendan Higgins <brendanhiggins@google.com># Author: Rae Moar <rmoar@google.com> from __future__ import annotationsfrom dataclasses import dataclassimport reimport textwrap from enum import Enum, autofrom typing import Iterable, Iterator, List, Optional, Tuple from kunit_printer import stdout class Test:	"""	A class to represent a test parsed from KTAP results. All KTAP	results within a test log are stored in a main Test object as	subtests. 	Attributes:	status : TestStatus - status of the test	name : str - name of the test	expected_count : int - expected number of subtests (0 if single		test case and None if unknown expected number of subtests)	subtests : List[Test] - list of subtests	log : List[str] - log of KTAP lines that correspond to the test	counts : TestCounts - counts of the test statuses and errors of		subtests or of the test itself if the test is a single		test case.	"""	def __init__(self) -> None:		"""Creates Test object with default attributes."""		self.status = TestStatus.TEST_CRASHED		self.name = ''		self.expected_count = 0  # type: Optional[int]		self.subtests = []  # type: List[Test]		self.log = []  # type: List[str]		self.counts = TestCounts() 	def __str__(self) -> str:		"""Returns string representation of a Test class object."""		return (f'Test({self.status}, {self.name}, {self.expected_count}, '			f'{self.subtests}, {self.log}, {self.counts})') 	def __repr__(self) -> str:		"""Returns string representation of a Test class object."""		return str(self) 	def add_error(self, error_message: str) -> None:		"""Records an error that occurred while parsing this test."""		self.counts.errors += 1		stdout.print_with_timestamp(stdout.red('[ERROR]') + f' Test: {self.name}: {error_message}') 	def ok_status(self) -> bool:		"""Returns true if the status was ok, i.e. passed or skipped."""		return self.status in (TestStatus.SUCCESS, TestStatus.SKIPPED) class TestStatus(Enum):	"""An enumeration class to represent the status of a test."""	SUCCESS = auto()	FAILURE = auto()	SKIPPED = auto()	TEST_CRASHED = auto()	NO_TESTS = auto()	FAILURE_TO_PARSE_TESTS = auto() @dataclassclass TestCounts:	"""	Tracks the counts of statuses of all test cases and any errors within	a Test.	"""	passed: int = 0	failed: int = 0	crashed: int = 0	skipped: int = 0	errors: int = 0 	def __str__(self) -> str:		"""Returns the string representation of a TestCounts object."""		statuses = [('passed', self.passed), ('failed', self.failed),			('crashed', self.crashed), ('skipped', self.skipped),			('errors', self.errors)]		return f'Ran {self.total()} tests: ' + \			', '.join(f'{s}: {n}' for s, n in statuses if n > 0) 	def total(self) -> int:		"""Returns the total number of test cases within a test		object, where a test case is a test with no subtests.		"""		return (self.passed + self.failed + self.crashed +			self.skipped) 	def add_subtest_counts(self, counts: TestCounts) -> None:		"""		Adds the counts of another TestCounts object to the current		TestCounts object. Used to add the counts of a subtest to the		parent test. 		Parameters:		counts - a different TestCounts object whose counts			will be added to the counts of the TestCounts object		"""		self.passed += counts.passed		self.failed += counts.failed		self.crashed += counts.crashed		self.skipped += counts.skipped		self.errors += counts.errors 	def get_status(self) -> TestStatus:		"""Returns the aggregated status of a Test using test		counts.		"""		if self.total() == 0:			return TestStatus.NO_TESTS		if self.crashed:			# Crashes should take priority.			return TestStatus.TEST_CRASHED		if self.failed:			return TestStatus.FAILURE		if self.passed:			# No failures or crashes, looks good!			return TestStatus.SUCCESS		# We have only skipped tests.		return TestStatus.SKIPPED 	def add_status(self, status: TestStatus) -> None:		"""Increments the count for `status`."""		if status == TestStatus.SUCCESS:			self.passed += 1		elif status == TestStatus.FAILURE:			self.failed += 1		elif status == TestStatus.SKIPPED:			self.skipped += 1		elif status != TestStatus.NO_TESTS:			self.crashed += 1 class LineStream:	"""	A class to represent the lines of kernel output.	Provides a lazy peek()/pop() interface over an iterator of	(line#, text).	"""	_lines: Iterator[Tuple[int, str]]	_next: Tuple[int, str]	_need_next: bool	_done: bool 	def __init__(self, lines: Iterator[Tuple[int, str]]):		"""Creates a new LineStream that wraps the given iterator."""		self._lines = lines		self._done = False		self._need_next = True		self._next = (0, '') 	def _get_next(self) -> None:		"""Advances the LineSteam to the next line, if necessary."""		if not self._need_next:			return		try:			self._next = next(self._lines)		except StopIteration:			self._done = True		finally:			self._need_next = False 	def peek(self) -> str:		"""Returns the current line, without advancing the LineStream.		"""		self._get_next()		return self._next[1] 	def pop(self) -> str:		"""Returns the current line and advances the LineStream to		the next line.		"""		s = self.peek()		if self._done:			raise ValueError(f'LineStream: going past EOF, last line was {s}')		self._need_next = True		return s 	def __bool__(self) -> bool:		"""Returns True if stream has more lines."""		self._get_next()		return not self._done 	# Only used by kunit_tool_test.py.	def __iter__(self) -> Iterator[str]:		"""Empties all lines stored in LineStream object into		Iterator object and returns the Iterator object.		"""		while bool(self):			yield self.pop() 	def line_number(self) -> int:		"""Returns the line number of the current line."""		self._get_next()		return self._next[0] # Parsing helper methods: KTAP_START = re.compile(r'\s*KTAP version ([0-9]+)$')TAP_START = re.compile(r'\s*TAP version ([0-9]+)$')KTAP_END = re.compile(r'\s*(List of all partitions:|'	'Kernel panic - not syncing: VFS:|reboot: System halted)') def extract_tap_lines(kernel_output: Iterable[str]) -> LineStream:	"""Extracts KTAP lines from the kernel output."""	def isolate_ktap_output(kernel_output: Iterable[str]) \			-> Iterator[Tuple[int, str]]:		line_num = 0		started = False		for line in kernel_output:			line_num += 1			line = line.rstrip()  # remove trailing \n			if not started and KTAP_START.search(line):				# start extracting KTAP lines and set prefix				# to number of characters before version line				prefix_len = len(					line.split('KTAP version')[0])				started = True				yield line_num, line[prefix_len:]			elif not started and TAP_START.search(line):				# start extracting KTAP lines and set prefix				# to number of characters before version line				prefix_len = len(line.split('TAP version')[0])				started = True				yield line_num, line[prefix_len:]			elif started and KTAP_END.search(line):				# stop extracting KTAP lines				break			elif started:				# remove the prefix, if any.				line = line[prefix_len:]				yield line_num, line	return LineStream(lines=isolate_ktap_output(kernel_output)) KTAP_VERSIONS = [1]TAP_VERSIONS = [13, 14] def check_version(version_num: int, accepted_versions: List[int],			version_type: str, test: Test) -> None:	"""	Adds error to test object if version number is too high or too	low. 	Parameters:	version_num - The inputted version number from the parsed KTAP or TAP		header line	accepted_version - List of accepted KTAP or TAP versions	version_type - 'KTAP' or 'TAP' depending on the type of		version line.	test - Test object for current test being parsed	"""	if version_num < min(accepted_versions):		test.add_error(f'{version_type} version lower than expected!')	elif version_num > max(accepted_versions):		test.add_error(f'{version_type} version higer than expected!') def parse_ktap_header(lines: LineStream, test: Test) -> bool:	"""	Parses KTAP/TAP header line and checks version number.	Returns False if fails to parse KTAP/TAP header line. 	Accepted formats:	- 'KTAP version [version number]'	- 'TAP version [version number]' 	Parameters:	lines - LineStream of KTAP output to parse	test - Test object for current test being parsed 	Return:	True if successfully parsed KTAP/TAP header line	"""	ktap_match = KTAP_START.match(lines.peek())	tap_match = TAP_START.match(lines.peek())	if ktap_match:		version_num = int(ktap_match.group(1))		check_version(version_num, KTAP_VERSIONS, 'KTAP', test)	elif tap_match:		version_num = int(tap_match.group(1))		check_version(version_num, TAP_VERSIONS, 'TAP', test)	else:		return False	lines.pop()	return True TEST_HEADER = re.compile(r'^\s*# Subtest: (.*)$') def parse_test_header(lines: LineStream, test: Test) -> bool:	"""	Parses test header and stores test name in test object.	Returns False if fails to parse test header line. 	Accepted format:	- '# Subtest: [test name]' 	Parameters:	lines - LineStream of KTAP output to parse	test - Test object for current test being parsed 	Return:	True if successfully parsed test header line	"""	match = TEST_HEADER.match(lines.peek())	if not match:		return False	test.name = match.group(1)	lines.pop()	return True TEST_PLAN = re.compile(r'^\s*1\.\.([0-9]+)') def parse_test_plan(lines: LineStream, test: Test) -> bool:	"""	Parses test plan line and stores the expected number of subtests in	test object. Reports an error if expected count is 0.	Returns False and sets expected_count to None if there is no valid test	plan. 	Accepted format:	- '1..[number of subtests]' 	Parameters:	lines - LineStream of KTAP output to parse	test - Test object for current test being parsed 	Return:	True if successfully parsed test plan line	"""	match = TEST_PLAN.match(lines.peek())	if not match:		test.expected_count = None		return False	expected_count = int(match.group(1))	test.expected_count = expected_count	lines.pop()	return True TEST_RESULT = re.compile(r'^\s*(ok|not ok) ([0-9]+) (- )?([^#]*)( # .*)?$') TEST_RESULT_SKIP = re.compile(r'^\s*(ok|not ok) ([0-9]+) (- )?(.*) # SKIP(.*)$') def peek_test_name_match(lines: LineStream, test: Test) -> bool:	"""	Matches current line with the format of a test result line and checks	if the name matches the name of the current test.	Returns False if fails to match format or name. 	Accepted format:	- '[ok|not ok] [test number] [-] [test name] [optional skip		directive]' 	Parameters:	lines - LineStream of KTAP output to parse	test - Test object for current test being parsed 	Return:	True if matched a test result line and the name matching the		expected test name	"""	line = lines.peek()	match = TEST_RESULT.match(line)	if not match:		return False	name = match.group(4)	return name == test.name def parse_test_result(lines: LineStream, test: Test,			expected_num: int) -> bool:	"""	Parses test result line and stores the status and name in the test	object. Reports an error if the test number does not match expected	test number.	Returns False if fails to parse test result line. 	Note that the SKIP directive is the only direction that causes a	change in status. 	Accepted format:	- '[ok|not ok] [test number] [-] [test name] [optional skip		directive]' 	Parameters:	lines - LineStream of KTAP output to parse	test - Test object for current test being parsed	expected_num - expected test number for current test 	Return:	True if successfully parsed a test result line.	"""	line = lines.peek()	match = TEST_RESULT.match(line)	skip_match = TEST_RESULT_SKIP.match(line) 	# Check if line matches test result line format	if not match:		return False	lines.pop() 	# Set name of test object	if skip_match:		test.name = skip_match.group(4)	else:		test.name = match.group(4) 	# Check test num	num = int(match.group(2))	if num != expected_num:		test.add_error(f'Expected test number {expected_num} but found {num}') 	# Set status of test object	status = match.group(1)	if skip_match:		test.status = TestStatus.SKIPPED	elif status == 'ok':		test.status = TestStatus.SUCCESS	else:		test.status = TestStatus.FAILURE	return True def parse_diagnostic(lines: LineStream) -> List[str]:	"""	Parse lines that do not match the format of a test result line or	test header line and returns them in list. 	Line formats that are not parsed:	- '# Subtest: [test name]'	- '[ok|not ok] [test number] [-] [test name] [optional skip		directive]'	- 'KTAP version [version number]' 	Parameters:	lines - LineStream of KTAP output to parse 	Return:	Log of diagnostic lines	"""	log = []  # type: List[str]	non_diagnostic_lines = [TEST_RESULT, TEST_HEADER, KTAP_START]	while lines and not any(re.match(lines.peek())			for re in non_diagnostic_lines):		log.append(lines.pop())	return log  # Printing helper methods: DIVIDER = '=' * 60 def format_test_divider(message: str, len_message: int) -> str:	"""	Returns string with message centered in fixed width divider. 	Example:	'===================== message example =====================' 	Parameters:	message - message to be centered in divider line	len_message - length of the message to be printed such that		any characters of the color codes are not counted 	Return:	String containing message centered in fixed width divider	"""	default_count = 3  # default number of dashes	len_1 = default_count	len_2 = default_count	difference = len(DIVIDER) - len_message - 2  # 2 spaces added	if difference > 0:		# calculate number of dashes for each side of the divider		len_1 = int(difference / 2)		len_2 = difference - len_1	return ('=' * len_1) + f' {message} ' + ('=' * len_2) def print_test_header(test: Test) -> None:	"""	Prints test header with test name and optionally the expected number	of subtests. 	Example:	'=================== example (2 subtests) ===================' 	Parameters:	test - Test object representing current test being printed	"""	message = test.name	if message != "":		# Add a leading space before the subtest counts only if a test name		# is provided using a "# Subtest" header line.

Lines 1–500 of 814. Load more