@@ -441,6 +441,7 @@ def parse_diagnostic(lines: LineStream) -> List[str]:
- '# Subtest: [test name]'
- '[ok|not ok] [test number] [-] [test name] [optional skip
directive]'
+ - 'KTAP version [version number]'
Parameters:
lines - LineStream of KTAP output to parse
@@ -449,8 +450,9 @@ def parse_diagnostic(lines: LineStream) -> List[str]:
Log of diagnostic lines
"""
log = [] # type: List[str]
- while lines and not TEST_RESULT.match(lines.peek()) and not \
- TEST_HEADER.match(lines.peek()):
+ non_diagnostic_lines = [TEST_RESULT, TEST_HEADER, KTAP_START]
+ while lines and not any(re.match(lines.peek())
+ for re in non_diagnostic_lines):
log.append(lines.pop())
return log
@@ -496,11 +498,15 @@ def print_test_header(test: Test) -> None:
test - Test object representing current test being printed
"""
message = test.name
+ if message != "":
+ # Add a leading space before the subtest counts only if a test name
+ # is provided using a "# Subtest" header line.
+ message += " "
if test.expected_count:
if test.expected_count == 1:
- message += ' (1 subtest)'
+ message += '(1 subtest)'
else:
- message += f' ({test.expected_count} subtests)'
+ message += f'({test.expected_count} subtests)'
stdout.print_with_timestamp(format_test_divider(message, len(message)))
def print_log(log: Iterable[str]) -> None:
@@ -647,7 +653,7 @@ def bubble_up_test_results(test: Test) -> None:
elif test.counts.get_status() == TestStatus.TEST_CRASHED:
test.status = TestStatus.TEST_CRASHED
-def parse_test(lines: LineStream, expected_num: int, log: List[str]) -> Test:
+def parse_test(lines: LineStream, expected_num: int, log: List[str], is_subtest: bool) -> Test:
"""
Finds next test to parse in LineStream, creates new Test object,
parses any subtests of the test, populates Test object with all
@@ -665,15 +671,32 @@ def parse_test(lines: LineStream, expected_num: int, log: List[str]) -> Test:
1..4
[subtests]
- - Subtest header line
+ - Subtest header (must include either the KTAP version line or
+ "# Subtest" header line)
- Example:
+ Example (preferred format with both KTAP version line and
+ "# Subtest" line):
+
+ KTAP version 1
+ # Subtest: name
+ 1..3
+ [subtests]
+ ok 1 name
+
+ Example (only "# Subtest" line):
# Subtest: name
1..3
[subtests]
ok 1 name
+ Example (only KTAP version line, compliant with KTAP v1 spec):
+
+ KTAP version 1
+ 1..3
+ [subtests]
+ ok 1 name
+
- Test result line
Example:
@@ -685,28 +708,29 @@ def parse_test(lines: LineStream, expected_num: int, log: List[str]) -> Test:
expected_num - expected test number for test to be parsed
log - list of strings containing any preceding diagnostic lines
corresponding to the current test
+ is_subtest - boolean indicating whether test is a subtest
Return:
Test object populated with characteristics and any subtests
"""
test = Test()
test.log.extend(log)
- parent_test = False
- main = parse_ktap_header(lines, test)
- if main:
- # If KTAP/TAP header is found, attempt to parse
+ if not is_subtest:
+ # If parsing the main/top-level test, parse KTAP version line and
# test plan
test.name = "main"
+ ktap_line = parse_ktap_header(lines, test)
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
- # subtest header
- parent_test = parse_test_header(lines, test)
+ # If not the main test, attempt to parse a test header contatining
+ # the KTAP version line and/or subtest header line
+ ktap_line = parse_ktap_header(lines, test)
+ subtest_line = parse_test_header(lines, test)
+ parent_test = (ktap_line or subtest_line)
if parent_test:
- # If subtest header is found, attempt to parse
- # test plan and print header
+ # If KTAP version line and/or subtest header is found, attempt
+ # to parse test plan and print test header
parse_test_plan(lines, test)
print_test_header(test)
expected_count = test.expected_count
@@ -721,7 +745,7 @@ def parse_test(lines: LineStream, expected_num: int, log: List[str]) -> Test:
sub_log = parse_diagnostic(lines)
sub_test = Test()
if not lines or (peek_test_name_match(lines, test) and
- not main):
+ is_subtest):
if expected_count and test_num <= expected_count:
# If parser reaches end of test before
# parsing expected number of subtests, print
@@ -735,20 +759,19 @@ def parse_test(lines: LineStream, expected_num: int, log: List[str]) -> Test:
test.log.extend(sub_log)
break
else:
- sub_test = parse_test(lines, test_num, sub_log)
+ sub_test = parse_test(lines, test_num, sub_log, True)
subtests.append(sub_test)
test_num += 1
test.subtests = subtests
- if not main:
+ if is_subtest:
# If not main test, look for test result line
test.log.extend(parse_diagnostic(lines))
- if (parent_test and peek_test_name_match(lines, test)) or \
- not parent_test:
- parse_test_result(lines, test, expected_num)
- else:
+ if test.name != "" and not peek_test_name_match(lines, test):
test.add_error('missing subtest result line!')
+ else:
+ parse_test_result(lines, test, expected_num)
- # Check for there being no tests
+ # Check for there being no subtests within parent test
if parent_test and len(subtests) == 0:
# Don't override a bad status if this test had one reported.
# Assumption: no subtests means CRASHED is from Test.__init__()
@@ -758,11 +781,11 @@ def parse_test(lines: LineStream, expected_num: int, log: List[str]) -> Test:
# Add statuses to TestCounts attribute in Test object
bubble_up_test_results(test)
- if parent_test and not main:
+ if parent_test and is_subtest:
# If test has subtests and is not the main test object, print
# footer.
print_test_footer(test)
- elif not main:
+ elif is_subtest:
print_test_result(test)
return test
@@ -785,7 +808,7 @@ def parse_run_tests(kernel_output: Iterable[str]) -> Test:
test.add_error('could not find any KTAP output!')
test.status = TestStatus.FAILURE_TO_PARSE_TESTS
else:
- test = parse_test(lines, 0, [])
+ test = parse_test(lines, 0, [], False)
if test.status != TestStatus.NO_TESTS:
test.status = test.counts.get_status()
stdout.print_with_timestamp(DIVIDER)
@@ -312,6 +312,14 @@ class KUnitParserTest(unittest.TestCase):
self.assertEqual(kunit_parser._summarize_failed_tests(result),
'Failures: all_failed_suite, some_failed_suite.test2')
+ def test_ktap_format(self):
+ ktap_log = test_data_path('test_parse_ktap_output.log')
+ with open(ktap_log) as file:
+ result = kunit_parser.parse_run_tests(file.readlines())
+ self.assertEqual(result.counts, kunit_parser.TestCounts(passed=3))
+ self.assertEqual('suite', result.subtests[0].name)
+ self.assertEqual('case_1', result.subtests[0].subtests[0].name)
+ self.assertEqual('case_2', result.subtests[0].subtests[1].name)
def line_stream_from_strs(strs: Iterable[str]) -> kunit_parser.LineStream:
return kunit_parser.LineStream(enumerate(strs, start=1))
new file mode 100644
@@ -0,0 +1,8 @@
+KTAP version 1
+1..1
+ KTAP version 1
+ 1..3
+ ok 1 case_1
+ ok 2 case_2
+ ok 3 case_3
+ok 1 suite