diff options
Diffstat (limited to 'tools/testing/kunit/kunit_parser.py')
| -rw-r--r-- | tools/testing/kunit/kunit_parser.py | 57 |
1 files changed, 34 insertions, 23 deletions
diff --git a/tools/testing/kunit/kunit_parser.py b/tools/testing/kunit/kunit_parser.py index 3355196d0515..05ff334761dd 100644 --- a/tools/testing/kunit/kunit_parser.py +++ b/tools/testing/kunit/kunit_parser.py @@ -12,14 +12,11 @@ from __future__ import annotations import re -from collections import namedtuple -from datetime import datetime +import datetime from enum import Enum, auto from functools import reduce from typing import Iterable, Iterator, List, Optional, Tuple -TestResult = namedtuple('TestResult', ['status','test','log']) - class Test(object): """ A class to represent a test parsed from KTAP results. All KTAP @@ -168,42 +165,51 @@ class TestCounts: class LineStream: """ A class to represent the lines of kernel output. - Provides a peek()/pop() interface over an iterator of + 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, '') - self._get_next() def _get_next(self) -> None: - """Advances the LineSteam to the next line.""" + """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. """ - n = self._next - self._get_next() - return n[1] + 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. @@ -216,6 +222,7 @@ class LineStream: def line_number(self) -> int: """Returns the line number of the current line.""" + self._get_next() return self._next[0] # Parsing helper methods: @@ -340,8 +347,8 @@ 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 reports missing test plan error if fails to parse - test plan. + Returns False and sets expected_count to None if there is no valid test + plan. Accepted format: - '1..[number of subtests]' @@ -356,14 +363,10 @@ def parse_test_plan(lines: LineStream, test: Test) -> bool: match = TEST_PLAN.match(lines.peek()) if not match: test.expected_count = None - test.add_error('missing plan line!') return False test.log.append(lines.pop()) expected_count = int(match.group(1)) test.expected_count = expected_count - if expected_count == 0: - test.status = TestStatus.NO_TESTS - test.add_error('0 tests run!') return True TEST_RESULT = re.compile(r'^(ok|not ok) ([0-9]+) (- )?([^#]*)( # .*)?$') @@ -514,7 +517,7 @@ ANSI_LEN = len(red('')) def print_with_timestamp(message: str) -> None: """Prints message with timestamp at beginning.""" - print('[%s] %s' % (datetime.now().strftime('%H:%M:%S'), message)) + print('[%s] %s' % (datetime.datetime.now().strftime('%H:%M:%S'), message)) def format_test_divider(message: str, len_message: int) -> str: """ @@ -590,6 +593,8 @@ def format_test_result(test: Test) -> str: return (green('[PASSED] ') + test.name) elif test.status == TestStatus.SKIPPED: return (yellow('[SKIPPED] ') + test.name) + elif test.status == TestStatus.NO_TESTS: + return (yellow('[NO TESTS RUN] ') + test.name) elif test.status == TestStatus.TEST_CRASHED: print_log(test.log) return (red('[CRASHED] ') + test.name) @@ -732,6 +737,7 @@ def parse_test(lines: LineStream, expected_num: int, log: List[str]) -> Test: # test plan test.name = "main" parse_test_plan(lines, test) + parent_test = True else: # If KTAP/TAP header is not found, test must be subtest # header or test result line so parse attempt to parser @@ -745,7 +751,7 @@ def parse_test(lines: LineStream, expected_num: int, log: List[str]) -> Test: expected_count = test.expected_count subtests = [] test_num = 1 - while expected_count is None or test_num <= expected_count: + while parent_test and (expected_count is None or test_num <= expected_count): # Loop to parse any subtests. # Break after parsing expected number of tests or # if expected number of tests is unknown break when test @@ -780,9 +786,15 @@ def parse_test(lines: LineStream, expected_num: int, log: List[str]) -> Test: parse_test_result(lines, test, expected_num) else: test.add_error('missing subtest result line!') + + # Check for there being no tests + if parent_test and len(subtests) == 0: + test.status = TestStatus.NO_TESTS + test.add_error('0 tests run!') + # Add statuses to TestCounts attribute in Test object bubble_up_test_results(test) - if parent_test: + if parent_test and not main: # If test has subtests and is not the main test object, print # footer. print_test_footer(test) @@ -790,7 +802,7 @@ def parse_test(lines: LineStream, expected_num: int, log: List[str]) -> Test: print_test_result(test) return test -def parse_run_tests(kernel_output: Iterable[str]) -> TestResult: +def parse_run_tests(kernel_output: Iterable[str]) -> Test: """ Using kernel output, extract KTAP lines, parse the lines for test results and print condensed test results and summary line . @@ -799,8 +811,7 @@ def parse_run_tests(kernel_output: Iterable[str]) -> TestResult: kernel_output - Iterable object contains lines of kernel output Return: - TestResult - Tuple containg status of main test object, main test - object with all subtests, and log of all KTAP lines. + Test - the main test object with all subtests. """ print_with_timestamp(DIVIDER) lines = extract_tap_lines(kernel_output) @@ -814,4 +825,4 @@ def parse_run_tests(kernel_output: Iterable[str]) -> TestResult: test.status = test.counts.get_status() print_with_timestamp(DIVIDER) print_summary_line(test) - return TestResult(test.status, test, lines) + return test |