| kunit_parser.py (5937e0c04afc7d4b7b737fda93316ba4b74183c0) | kunit_parser.py (c2bb92bc4ea13842fdd27819c0d5b48df2b86ea5) |
|---|---|
| 1# SPDX-License-Identifier: GPL-2.0 2# 3# Parses KTAP test results from a kernel dmesg log and incrementally prints 4# results with reader-friendly format. Stores and returns test results in a 5# Test object. 6# 7# Copyright (C) 2019, Google LLC. 8# Author: Felix Guo <felixguoxiuping@gmail.com> 9# Author: Brendan Higgins <brendanhiggins@google.com> 10# Author: Rae Moar <rmoar@google.com> 11 12from __future__ import annotations 13from dataclasses import dataclass 14import re 15import sys | 1# SPDX-License-Identifier: GPL-2.0 2# 3# Parses KTAP test results from a kernel dmesg log and incrementally prints 4# results with reader-friendly format. Stores and returns test results in a 5# Test object. 6# 7# Copyright (C) 2019, Google LLC. 8# Author: Felix Guo <felixguoxiuping@gmail.com> 9# Author: Brendan Higgins <brendanhiggins@google.com> 10# Author: Rae Moar <rmoar@google.com> 11 12from __future__ import annotations 13from dataclasses import dataclass 14import re 15import sys |
| 16import textwrap |
|
| 16 17from enum import Enum, auto 18from typing import Iterable, Iterator, List, Optional, Tuple 19 20from kunit_printer import stdout 21 22class Test: 23 """ --- 179 unchanged lines hidden (view full) --- 203 204 def line_number(self) -> int: 205 """Returns the line number of the current line.""" 206 self._get_next() 207 return self._next[0] 208 209# Parsing helper methods: 210 | 17 18from enum import Enum, auto 19from typing import Iterable, Iterator, List, Optional, Tuple 20 21from kunit_printer import stdout 22 23class Test: 24 """ --- 179 unchanged lines hidden (view full) --- 204 205 def line_number(self) -> int: 206 """Returns the line number of the current line.""" 207 self._get_next() 208 return self._next[0] 209 210# Parsing helper methods: 211 |
| 211KTAP_START = re.compile(r'KTAP version ([0-9]+)$') 212TAP_START = re.compile(r'TAP version ([0-9]+)$') 213KTAP_END = re.compile('(List of all partitions:|' | 212KTAP_START = re.compile(r'\s*KTAP version ([0-9]+)$') 213TAP_START = re.compile(r'\s*TAP version ([0-9]+)$') 214KTAP_END = re.compile(r'\s*(List of all partitions:|' |
| 214 'Kernel panic - not syncing: VFS:|reboot: System halted)') 215 | 215 'Kernel panic - not syncing: VFS:|reboot: System halted)') 216 |
| 216def extract_tap_lines(kernel_output: Iterable[str], lstrip=True) -> LineStream: | 217def extract_tap_lines(kernel_output: Iterable[str]) -> LineStream: |
| 217 """Extracts KTAP lines from the kernel output.""" 218 def isolate_ktap_output(kernel_output: Iterable[str]) \ 219 -> Iterator[Tuple[int, str]]: 220 line_num = 0 221 started = False 222 for line in kernel_output: 223 line_num += 1 224 line = line.rstrip() # remove trailing \n --- 9 unchanged lines hidden (view full) --- 234 # to number of characters before version line 235 prefix_len = len(line.split('TAP version')[0]) 236 started = True 237 yield line_num, line[prefix_len:] 238 elif started and KTAP_END.search(line): 239 # stop extracting KTAP lines 240 break 241 elif started: | 218 """Extracts KTAP lines from the kernel output.""" 219 def isolate_ktap_output(kernel_output: Iterable[str]) \ 220 -> Iterator[Tuple[int, str]]: 221 line_num = 0 222 started = False 223 for line in kernel_output: 224 line_num += 1 225 line = line.rstrip() # remove trailing \n --- 9 unchanged lines hidden (view full) --- 235 # to number of characters before version line 236 prefix_len = len(line.split('TAP version')[0]) 237 started = True 238 yield line_num, line[prefix_len:] 239 elif started and KTAP_END.search(line): 240 # stop extracting KTAP lines 241 break 242 elif started: |
| 242 # remove the prefix and optionally any leading 243 # whitespace. Our parsing logic relies on this. | 243 # remove the prefix, if any. |
| 244 line = line[prefix_len:] | 244 line = line[prefix_len:] |
| 245 if lstrip: 246 line = line.lstrip() | |
| 247 yield line_num, line 248 return LineStream(lines=isolate_ktap_output(kernel_output)) 249 250KTAP_VERSIONS = [1] 251TAP_VERSIONS = [13, 14] 252 253def check_version(version_num: int, accepted_versions: List[int], 254 version_type: str, test: Test) -> None: --- 38 unchanged lines hidden (view full) --- 293 elif tap_match: 294 version_num = int(tap_match.group(1)) 295 check_version(version_num, TAP_VERSIONS, 'TAP', test) 296 else: 297 return False 298 lines.pop() 299 return True 300 | 245 yield line_num, line 246 return LineStream(lines=isolate_ktap_output(kernel_output)) 247 248KTAP_VERSIONS = [1] 249TAP_VERSIONS = [13, 14] 250 251def check_version(version_num: int, accepted_versions: List[int], 252 version_type: str, test: Test) -> None: --- 38 unchanged lines hidden (view full) --- 291 elif tap_match: 292 version_num = int(tap_match.group(1)) 293 check_version(version_num, TAP_VERSIONS, 'TAP', test) 294 else: 295 return False 296 lines.pop() 297 return True 298 |
| 301TEST_HEADER = re.compile(r'^# Subtest: (.*)$') | 299TEST_HEADER = re.compile(r'^\s*# Subtest: (.*)$') |
| 302 303def parse_test_header(lines: LineStream, test: Test) -> bool: 304 """ 305 Parses test header and stores test name in test object. 306 Returns False if fails to parse test header line. 307 308 Accepted format: 309 - '# Subtest: [test name]' --- 7 unchanged lines hidden (view full) --- 317 """ 318 match = TEST_HEADER.match(lines.peek()) 319 if not match: 320 return False 321 test.name = match.group(1) 322 lines.pop() 323 return True 324 | 300 301def parse_test_header(lines: LineStream, test: Test) -> bool: 302 """ 303 Parses test header and stores test name in test object. 304 Returns False if fails to parse test header line. 305 306 Accepted format: 307 - '# Subtest: [test name]' --- 7 unchanged lines hidden (view full) --- 315 """ 316 match = TEST_HEADER.match(lines.peek()) 317 if not match: 318 return False 319 test.name = match.group(1) 320 lines.pop() 321 return True 322 |
| 325TEST_PLAN = re.compile(r'1\.\.([0-9]+)') | 323TEST_PLAN = re.compile(r'^\s*1\.\.([0-9]+)') |
| 326 327def parse_test_plan(lines: LineStream, test: Test) -> bool: 328 """ 329 Parses test plan line and stores the expected number of subtests in 330 test object. Reports an error if expected count is 0. 331 Returns False and sets expected_count to None if there is no valid test 332 plan. 333 --- 11 unchanged lines hidden (view full) --- 345 if not match: 346 test.expected_count = None 347 return False 348 expected_count = int(match.group(1)) 349 test.expected_count = expected_count 350 lines.pop() 351 return True 352 | 324 325def parse_test_plan(lines: LineStream, test: Test) -> bool: 326 """ 327 Parses test plan line and stores the expected number of subtests in 328 test object. Reports an error if expected count is 0. 329 Returns False and sets expected_count to None if there is no valid test 330 plan. 331 --- 11 unchanged lines hidden (view full) --- 343 if not match: 344 test.expected_count = None 345 return False 346 expected_count = int(match.group(1)) 347 test.expected_count = expected_count 348 lines.pop() 349 return True 350 |
| 353TEST_RESULT = re.compile(r'^(ok|not ok) ([0-9]+) (- )?([^#]*)( # .*)?$') | 351TEST_RESULT = re.compile(r'^\s*(ok|not ok) ([0-9]+) (- )?([^#]*)( # .*)?$') |
| 354 | 352 |
| 355TEST_RESULT_SKIP = re.compile(r'^(ok|not ok) ([0-9]+) (- )?(.*) # SKIP(.*)$') | 353TEST_RESULT_SKIP = re.compile(r'^\s*(ok|not ok) ([0-9]+) (- )?(.*) # SKIP(.*)$') |
| 356 357def peek_test_name_match(lines: LineStream, test: Test) -> bool: 358 """ 359 Matches current line with the format of a test result line and checks 360 if the name matches the name of the current test. 361 Returns False if fails to match format or name. 362 363 Accepted format: --- 142 unchanged lines hidden (view full) --- 506 if test.expected_count == 1: 507 message += '(1 subtest)' 508 else: 509 message += f'({test.expected_count} subtests)' 510 stdout.print_with_timestamp(format_test_divider(message, len(message))) 511 512def print_log(log: Iterable[str]) -> None: 513 """Prints all strings in saved log for test in yellow.""" | 354 355def peek_test_name_match(lines: LineStream, test: Test) -> bool: 356 """ 357 Matches current line with the format of a test result line and checks 358 if the name matches the name of the current test. 359 Returns False if fails to match format or name. 360 361 Accepted format: --- 142 unchanged lines hidden (view full) --- 504 if test.expected_count == 1: 505 message += '(1 subtest)' 506 else: 507 message += f'({test.expected_count} subtests)' 508 stdout.print_with_timestamp(format_test_divider(message, len(message))) 509 510def print_log(log: Iterable[str]) -> None: 511 """Prints all strings in saved log for test in yellow.""" |
| 514 for m in log: 515 stdout.print_with_timestamp(stdout.yellow(m)) | 512 formatted = textwrap.dedent('\n'.join(log)) 513 for line in formatted.splitlines(): 514 stdout.print_with_timestamp(stdout.yellow(line)) |
| 516 517def format_test_result(test: Test) -> str: 518 """ 519 Returns string with formatted test result with colored status and test 520 name. 521 522 Example: 523 '[PASSED] example' --- 293 unchanged lines hidden --- | 515 516def format_test_result(test: Test) -> str: 517 """ 518 Returns string with formatted test result with colored status and test 519 name. 520 521 Example: 522 '[PASSED] example' --- 293 unchanged lines hidden --- |