| kunit_parser.py (50501936288d6a29d7ef78f25d00e33240fad45f) | kunit_parser.py (723c8258c8fe167191b53e274dea435c4522e4d7) |
|---|---|
| 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> --- 198 unchanged lines hidden (view full) --- 207 return self._next[0] 208 209# Parsing helper methods: 210 211KTAP_START = re.compile(r'\s*KTAP version ([0-9]+)$') 212TAP_START = re.compile(r'\s*TAP version ([0-9]+)$') 213KTAP_END = re.compile(r'\s*(List of all partitions:|' 214 'Kernel panic - not syncing: VFS:|reboot: System halted)') | 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> --- 198 unchanged lines hidden (view full) --- 207 return self._next[0] 208 209# Parsing helper methods: 210 211KTAP_START = re.compile(r'\s*KTAP version ([0-9]+)$') 212TAP_START = re.compile(r'\s*TAP version ([0-9]+)$') 213KTAP_END = re.compile(r'\s*(List of all partitions:|' 214 'Kernel panic - not syncing: VFS:|reboot: System halted)') |
| 215EXECUTOR_ERROR = re.compile(r'\s*kunit executor: (.*)$') |
|
| 215 216def 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: --- 14 unchanged lines hidden (view full) --- 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: 242 # remove the prefix, if any. 243 line = line[prefix_len:] 244 yield line_num, line | 216 217def extract_tap_lines(kernel_output: Iterable[str]) -> LineStream: 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: --- 14 unchanged lines hidden (view full) --- 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: 243 # remove the prefix, if any. 244 line = line[prefix_len:] 245 yield line_num, line |
| 246 elif EXECUTOR_ERROR.search(line): 247 yield line_num, line |
|
| 245 return LineStream(lines=isolate_ktap_output(kernel_output)) 246 247KTAP_VERSIONS = [1] 248TAP_VERSIONS = [13, 14] 249 250def check_version(version_num: int, accepted_versions: List[int], 251 version_type: str, test: Test) -> None: 252 """ --- 189 unchanged lines hidden (view full) --- 442 443 Parameters: 444 lines - LineStream of KTAP output to parse 445 446 Return: 447 Log of diagnostic lines 448 """ 449 log = [] # type: List[str] | 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: 255 """ --- 189 unchanged lines hidden (view full) --- 445 446 Parameters: 447 lines - LineStream of KTAP output to parse 448 449 Return: 450 Log of diagnostic lines 451 """ 452 log = [] # type: List[str] |
| 450 non_diagnostic_lines = [TEST_RESULT, TEST_HEADER, KTAP_START] | 453 non_diagnostic_lines = [TEST_RESULT, TEST_HEADER, KTAP_START, TAP_START] |
| 451 while lines and not any(re.match(lines.peek()) 452 for re in non_diagnostic_lines): 453 log.append(lines.pop()) 454 return log 455 456 457# Printing helper methods: 458 --- 249 unchanged lines hidden (view full) --- 708 corresponding to the current test 709 is_subtest - boolean indicating whether test is a subtest 710 711 Return: 712 Test object populated with characteristics and any subtests 713 """ 714 test = Test() 715 test.log.extend(log) | 454 while lines and not any(re.match(lines.peek()) 455 for re in non_diagnostic_lines): 456 log.append(lines.pop()) 457 return log 458 459 460# Printing helper methods: 461 --- 249 unchanged lines hidden (view full) --- 711 corresponding to the current test 712 is_subtest - boolean indicating whether test is a subtest 713 714 Return: 715 Test object populated with characteristics and any subtests 716 """ 717 test = Test() 718 test.log.extend(log) |
| 719 720 # Parse any errors prior to parsing tests 721 err_log = parse_diagnostic(lines) 722 test.log.extend(err_log) 723 |
|
| 716 if not is_subtest: 717 # If parsing the main/top-level test, parse KTAP version line and 718 # test plan 719 test.name = "main" 720 ktap_line = parse_ktap_header(lines, test) 721 parse_test_plan(lines, test) 722 parent_test = True 723 else: --- 45 unchanged lines hidden (view full) --- 769 else: 770 parse_test_result(lines, test, expected_num) 771 772 # Check for there being no subtests within parent test 773 if parent_test and len(subtests) == 0: 774 # Don't override a bad status if this test had one reported. 775 # Assumption: no subtests means CRASHED is from Test.__init__() 776 if test.status in (TestStatus.TEST_CRASHED, TestStatus.SUCCESS): | 724 if not is_subtest: 725 # If parsing the main/top-level test, parse KTAP version line and 726 # test plan 727 test.name = "main" 728 ktap_line = parse_ktap_header(lines, test) 729 parse_test_plan(lines, test) 730 parent_test = True 731 else: --- 45 unchanged lines hidden (view full) --- 777 else: 778 parse_test_result(lines, test, expected_num) 779 780 # Check for there being no subtests within parent test 781 if parent_test and len(subtests) == 0: 782 # Don't override a bad status if this test had one reported. 783 # Assumption: no subtests means CRASHED is from Test.__init__() 784 if test.status in (TestStatus.TEST_CRASHED, TestStatus.SUCCESS): |
| 785 print_log(test.log) |
|
| 777 test.status = TestStatus.NO_TESTS 778 test.add_error('0 tests run!') 779 780 # Add statuses to TestCounts attribute in Test object 781 bubble_up_test_results(test) 782 if parent_test and is_subtest: 783 # If test has subtests and is not the main test object, print 784 # footer. --- 30 unchanged lines hidden --- | 786 test.status = TestStatus.NO_TESTS 787 test.add_error('0 tests run!') 788 789 # Add statuses to TestCounts attribute in Test object 790 bubble_up_test_results(test) 791 if parent_test and is_subtest: 792 # If test has subtests and is not the main test object, print 793 # footer. --- 30 unchanged lines hidden --- |