path: root/tools/testing/kunit/kunit_parser.py

# 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 annotations
from dataclasses import dataclass
import re
import textwrap

from enum import Enum, auto
from 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()

@dataclass
class 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)')
EXECUTOR_ERROR = re.compile(r'\s*kunit executor: (.*)$')

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