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.
22 class Test:  class
24 	A class to represent a test parsed from KTAP results. All KTAP
25 	results within a test log are stored in a main Test object as
29 	status : TestStatus - status of the test
30 	name : str - name of the test
32 		test case and None if unknown expected number of subtests)
33 	subtests : List[Test] - list of subtests
34 	log : List[str] - log of KTAP lines that correspond to the test
35 	counts : TestCounts - counts of the test statuses and errors of
36 		subtests or of the test itself if the test is a single
37 		test case.
40 		"""Creates Test object with default attributes."""
44 		self.subtests = []  # type: List[Test]
49 		"""Returns string representation of a Test class object."""
50 		return (f'Test({self.status}, {self.name}, {self.expected_count}, '
54 		"""Returns string representation of a Test class object."""
58 		"""Records an error that occurred while parsing this test."""
60 		printer.print_with_timestamp(stdout.red('[ERROR]') + f' Test: {self.name}: {error_message}')
67 	"""An enumeration class to represent the status of a test."""
78 	Tracks the counts of statuses of all test cases and any errors within
79 	a Test.
96 		"""Returns the total number of test cases within a test
97 		object, where a test case is a test with no subtests.
106 		parent test.
119 		"""Returns the aggregated status of a Test using test
254 			version_type: str, test: Test, printer: Printer) -> None:  argument
256 	Adds error to test object if version number is too high or too
265 	test - Test object for current test being parsed
269 		test.add_error(printer, f'{version_type} version lower than expected!')
271 		test.add_error(printer, f'{version_type} version higer than expected!')
273 def parse_ktap_header(lines: LineStream, test: Test, printer: Printer) -> bool:  argument
284 	test - Test object for current test being parsed
294 		check_version(version_num, KTAP_VERSIONS, 'KTAP', test, printer)
297 		check_version(version_num, TAP_VERSIONS, 'TAP', test, printer)
305 def parse_test_header(lines: LineStream, test: Test) -> bool:  argument
307 	Parses test header and stores test name in test object.
308 	Returns False if fails to parse test header line.
311 	- '# Subtest: [test name]'
315 	test - Test object for current test being parsed
318 	True if successfully parsed test header line
323 	test.name = match.group(1)
329 def parse_test_plan(lines: LineStream, test: Test) -> bool:  argument
331 	Parses test plan line and stores the expected number of subtests in
332 	test object. Reports an error if expected count is 0.
333 	Returns False and sets expected_count to None if there is no valid test
341 	test - Test object for current test being parsed
344 	True if successfully parsed test plan line
348 		test.expected_count = None
351 	test.expected_count = expected_count
359 def peek_test_name_match(lines: LineStream, test: Test) -> bool:  argument
361 	Matches current line with the format of a test result line and checks
362 	if the name matches the name of the current test.
366 	- '[ok|not ok] [test number] [-] [test name] [optional skip
371 	test - Test object for current test being parsed
374 	True if matched a test result line and the name matching the
375 		expected test name
382 	return name == test.name
384 def parse_test_result(lines: LineStream, test: Test,  argument
387 	Parses test result line and stores the status and name in the test
388 	object. Reports an error if the test number does not match expected
389 	test number.
390 	Returns False if fails to parse test result line.
396 	- '[ok|not ok] [test number] [-] [test name] [optional skip
401 	test - Test object for current test being parsed
402 	expected_num - expected test number for current test
406 	True if successfully parsed a test result line.
412 	# Check if line matches test result line format
417 	# Set name of test object
419 		test.name = skip_match.group(4)
421 		test.name = match.group(4)
423 	# Check test num
426 		test.add_error(printer, f'Expected test number {expected_num} but found {num}')
428 	# Set status of test object
431 		test.status = TestStatus.SKIPPED
433 		test.status = TestStatus.SUCCESS
435 		test.status = TestStatus.FAILURE
440 	Parse lines that do not match the format of a test result line or
441 	test header line and returns them in list.
444 	- '# Subtest: [test name]'
445 	- '[ok|not ok] [test number] [-] [test name] [optional skip
492 def print_test_header(test: Test, printer: Printer) -> None:  argument
494 	Prints test header with test name and optionally the expected number
501 	test - Test object representing current test being printed
504 	message = test.name
506 		# Add a leading space before the subtest counts only if a test name
509 	if test.expected_count:
510 		if test.expected_count == 1:
513 			message += f'({test.expected_count} subtests)'
517 	"""Prints all strings in saved log for test in yellow."""
522 def format_test_result(test: Test, printer: Printer) -> str:  argument
524 	Returns string with formatted test result with colored status and test
531 	test - Test object representing current test being printed
535 	String containing formatted test result
537 	if test.status == TestStatus.SUCCESS:
538 		return printer.green('[PASSED] ') + test.name
539 	if test.status == TestStatus.SKIPPED:
540 		return printer.yellow('[SKIPPED] ') + test.name
541 	if test.status == TestStatus.NO_TESTS:
542 		return printer.yellow('[NO TESTS RUN] ') + test.name
543 	if test.status == TestStatus.TEST_CRASHED:
544 		print_log(test.log, printer)
545 		return stdout.red('[CRASHED] ') + test.name
546 	print_log(test.log, printer)
547 	return printer.red('[FAILED] ') + test.name
549 def print_test_result(test: Test, printer: Printer) -> None:  argument
551 	Prints result line with status of test.
557 	test - Test object representing current test being printed
560 	printer.print_with_timestamp(format_test_result(test, printer))
562 def print_test_footer(test: Test, printer: Printer) -> None:  argument
564 	Prints test footer with status of test.
570 	test - Test object representing current test being printed
573 	message = format_test_result(test, printer)
577 def print_test(test: Test, failed_only: bool, printer: Printer) -> None:  argument
579 	Prints Test object to given printer. For a child test, the result line is
580 	printed. For a parent test, the test header, all child test results, and
581 	the test footer are all printed. If failed_only is true, only failed/crashed
585 	test - Test object to print
589 	if test.name == "main":
591 		for subtest in test.subtests:
594 	elif test.subtests != []:
595 		if not failed_only or not test.ok_status():
596 			print_test_header(test, printer)
597 			for subtest in test.subtests:
599 			print_test_footer(test, printer)
601 		if not failed_only or not test.ok_status():
602 			print_test_result(test, printer)
604 def _summarize_failed_tests(test: Test) -> str:  argument
605 	"""Tries to summarize all the failing subtests in `test`."""
607 	def failed_names(test: Test, parent_name: str) -> List[str]:  argument
608 		# Note: we use 'main' internally for the top-level test.
610 			full_name = test.name
612 			full_name = parent_name + '.' + test.name
614 		if not test.subtests:  # this is a leaf node
618 		# Don't summarize it down "the top-level test failed", though.
619 		failed_subtests = [sub for sub in test.subtests if not sub.ok_status()]
620 		if parent_name and len(failed_subtests) ==  len(test.subtests):
628 	failures = failed_names(test, '')
636 def print_summary_line(test: Test, printer: Printer) -> None:  argument
638 	Prints summary line of test object. Color of line is dependent on
639 	status of test. Color is green if test passes, yellow if test is
640 	skipped, and red if the test fails or crashes. Summary line contains
641 	counts of the statuses of the tests subtests or the test itself if it
648 	test - Test object representing current test being printed
651 	if test.status == TestStatus.SUCCESS:
653 	elif test.status in (TestStatus.SKIPPED, TestStatus.NO_TESTS):
657 	printer.print_with_timestamp(color(f'Testing complete. {test.counts}'))
661 	if test.ok_status() or test.counts.total() < 100:
663 	summarized = _summarize_failed_tests(test)
670 def bubble_up_test_results(test: Test) -> None:  argument
672 	If the test has subtests, add the test counts of the subtests to the
673 	test and check if any of the tests crashed and if so set the test
674 	status to crashed. Otherwise if the test has no subtests add the
675 	status of the test to the test counts.
678 	test - Test object for current test being parsed
680 	subtests = test.subtests
681 	counts = test.counts
682 	status = test.status
687 	elif test.counts.get_status() == TestStatus.TEST_CRASHED:
688 		test.status = TestStatus.TEST_CRASHED
690 …(lines: LineStream, expected_num: int, log: List[str], is_subtest: bool, printer: Printer) -> Test:
692 	Finds next test to parse in LineStream, creates new Test object,
693 	parses any subtests of the test, populates Test object with all
694 	information (status, name) about the test and the Test objects for
695 	any subtests, and then returns the Test object. The method accepts
698 	Accepted test formats:
734 	- Test result line
738 	ok 1 - test
742 	expected_num - expected test number for test to be parsed
744 		corresponding to the current test
745 	is_subtest - boolean indicating whether test is a subtest
749 	Test object populated with characteristics and any subtests
751 	test = Test()
752 	test.log.extend(log)
756 	test.log.extend(err_log)
759 		# If parsing the main/top-level test, parse KTAP version line and
760 		# test plan
761 		test.name = "main"
762 		ktap_line = parse_ktap_header(lines, test, printer)
763 		test.log.extend(parse_diagnostic(lines))
764 		parse_test_plan(lines, test)
767 		# If not the main test, attempt to parse a test header containing
769 		ktap_line = parse_ktap_header(lines, test, printer)
770 		subtest_line = parse_test_header(lines, test)
774 			# to parse test plan and print test header
775 			test.log.extend(parse_diagnostic(lines))
776 			parse_test_plan(lines, test)
777 			print_test_header(test, printer)
778 	expected_count = test.expected_count
784 		# if expected number of tests is unknown break when test
788 		sub_test = Test()
789 		if not lines or (peek_test_name_match(lines, test) and
792 				# If parser reaches end of test before
795 				test.add_error(printer, 'missing expected subtest!')
797 				test.counts.add_status(
801 				test.log.extend(sub_log)
807 	test.subtests = subtests
809 		# If not main test, look for test result line
810 		test.log.extend(parse_diagnostic(lines))
811 		if test.name != "" and not peek_test_name_match(lines, test):
812 			test.add_error(printer, 'missing subtest result line!')
814 			parse_test_result(lines, test, expected_num, printer)
816 	# Check for there being no subtests within parent test
818 		# Don't override a bad status if this test had one reported.
819 		# Assumption: no subtests means CRASHED is from Test.__init__()
820 		if test.status in (TestStatus.TEST_CRASHED, TestStatus.SUCCESS):
821 			print_log(test.log, printer)
822 			test.status = TestStatus.NO_TESTS
823 			test.add_error(printer, '0 tests run!')
825 	# Add statuses to TestCounts attribute in Test object
826 	bubble_up_test_results(test)
828 		# If test has subtests and is not the main test object, print
830 		print_test_footer(test, printer)
832 		print_test_result(test, printer)
833 	return test
835 def parse_run_tests(kernel_output: Iterable[str], printer: Printer) -> Test:
837 	Using kernel output, extract KTAP lines, parse the lines for test
838 	results and print condensed test results and summary line.
845 	Test - the main test object with all subtests.
849 	test = Test()
851 		test.name = '<missing>'
852 		test.add_error(printer, 'Could not find any KTAP output. Did any KUnit tests run?')
853 		test.status = TestStatus.FAILURE_TO_PARSE_TESTS
855 		test = parse_test(lines, 0, [], False, printer)
856 		if test.status != TestStatus.NO_TESTS:
857 			test.status = test.counts.get_status()
859 	return test