@@ -17,6 +17,7 @@ from collections import namedtuple
from enum import Enum, auto
import kunit_config
+import kunit_json
import kunit_kernel
import kunit_parser
@@ -30,9 +31,9 @@ KunitBuildRequest = namedtuple('KunitBuildRequest',
KunitExecRequest = namedtuple('KunitExecRequest',
['timeout', 'build_dir', 'alltests'])
KunitParseRequest = namedtuple('KunitParseRequest',
- ['raw_output', 'input_data'])
+ ['raw_output', 'input_data', 'build_dir', 'json'])
KunitRequest = namedtuple('KunitRequest', ['raw_output','timeout', 'jobs',
- 'build_dir', 'alltests',
+ 'build_dir', 'alltests', 'json',
'make_options'])
KernelDirectoryPath = sys.argv[0].split('tools/testing/kunit/')[0]
@@ -113,12 +114,22 @@ def parse_tests(request: KunitParseRequest) -> KunitResult:
test_result = kunit_parser.TestResult(kunit_parser.TestStatus.SUCCESS,
[],
'Tests not Parsed.')
+
if request.raw_output:
kunit_parser.raw_output(request.input_data)
else:
test_result = kunit_parser.parse_run_tests(request.input_data)
parse_end = time.time()
+ if request.json:
+ json_obj = kunit_json.get_json_result(
+ test_result=test_result,
+ def_config='kunit_defconfig',
+ build_dir=request.build_dir,
+ json_path=request.json)
+ if request.json == 'stdout':
+ print(json_obj)
+
if test_result.status != kunit_parser.TestStatus.SUCCESS:
return KunitResult(KunitStatus.TEST_FAILURE, test_result,
parse_end - parse_start)
@@ -151,7 +162,9 @@ def run_tests(linux: kunit_kernel.LinuxSourceTree,
return exec_result
parse_request = KunitParseRequest(request.raw_output,
- exec_result.result)
+ exec_result.result,
+ request.build_dir,
+ request.json)
parse_result = parse_tests(parse_request)
run_end = time.time()
@@ -195,7 +208,12 @@ def add_exec_opts(parser):
def add_parse_opts(parser):
parser.add_argument('--raw_output', help='don\'t format output from kernel',
action='store_true')
-
+ parser.add_argument('--json',
+ nargs='?',
+ help='Stores test results in a JSON, and either '
+ 'prints to stdout or saves to file if a '
+ 'filename is specified',
+ type=str, const='stdout', default=None)
def main(argv, linux=None):
parser = argparse.ArgumentParser(
@@ -253,6 +271,7 @@ def main(argv, linux=None):
cli_args.jobs,
cli_args.build_dir,
cli_args.alltests,
+ cli_args.json,
cli_args.make_options)
result = run_tests(linux, request)
if result.status != KunitStatus.SUCCESS:
@@ -297,7 +316,9 @@ def main(argv, linux=None):
cli_args.alltests)
exec_result = exec_tests(linux, exec_request)
parse_request = KunitParseRequest(cli_args.raw_output,
- exec_result.result)
+ exec_result.result,
+ cli_args.build_dir,
+ cli_args.json)
result = parse_tests(parse_request)
kunit_parser.print_with_timestamp((
'Elapsed time: %.3fs\n') % (
@@ -311,7 +332,9 @@ def main(argv, linux=None):
with open(cli_args.file, 'r') as f:
kunit_output = f.read().splitlines()
request = KunitParseRequest(cli_args.raw_output,
- kunit_output)
+ kunit_output,
+ cli_args.build_dir,
+ cli_args.json)
result = parse_tests(request)
if result.status != KunitStatus.SUCCESS:
sys.exit(1)
new file mode 100644
@@ -0,0 +1,63 @@
+# SPDX-License-Identifier: GPL-2.0
+#
+# Generates JSON from KUnit results according to
+# KernelCI spec: https://github.com/kernelci/kernelci-doc/wiki/Test-API
+#
+# Copyright (C) 2020, Google LLC.
+# Author: Heidi Fahim <heidifahim@google.com>
+
+import json
+import os
+
+import kunit_parser
+
+from kunit_parser import TestStatus
+
+def get_json_result(test_result, def_config, build_dir, json_path):
+ sub_groups = []
+
+ # Each test suite is mapped to a KernelCI sub_group
+ for test_suite in test_result.suites:
+ sub_group = {
+ "name": test_suite.name,
+ "arch": "UM",
+ "defconfig": def_config,
+ "build_environment": build_dir,
+ "test_cases": [],
+ "lab_name": None,
+ "kernel": None,
+ "job": None,
+ "git_branch": "kselftest",
+ }
+ test_cases = []
+ # TODO: Add attachments attribute in test_case with detailed
+ # failure message, see https://api.kernelci.org/schema-test-case.html#get
+ for case in test_suite.cases:
+ test_case = {"name": case.name, "status": "FAIL"}
+ if case.status == TestStatus.SUCCESS:
+ test_case["status"] = "PASS"
+ elif case.status == TestStatus.TEST_CRASHED:
+ test_case["status"] = "ERROR"
+ test_cases.append(test_case)
+ sub_group["test_cases"] = test_cases
+ sub_groups.append(sub_group)
+ test_group = {
+ "name": "KUnit Test Group",
+ "arch": "UM",
+ "defconfig": def_config,
+ "build_environment": build_dir,
+ "sub_groups": sub_groups,
+ "lab_name": None,
+ "kernel": None,
+ "job": None,
+ "git_branch": "kselftest",
+ }
+ json_obj = json.dumps(test_group, indent=4)
+ if json_path != 'stdout':
+ with open(json_path, 'w') as result_path:
+ result_path.write(json_obj)
+ root = __file__.split('tools/testing/kunit/')[0]
+ kunit_parser.print_with_timestamp(
+ "Test results stored in %s" %
+ os.path.join(root, result_path.name))
+ return json_obj
@@ -11,11 +11,13 @@ from unittest import mock
import tempfile, shutil # Handling test_tmpdir
+import json
import os
import kunit_config
import kunit_parser
import kunit_kernel
+import kunit_json
import kunit
test_tmpdir = ''
@@ -230,6 +232,37 @@ class KUnitParserTest(unittest.TestCase):
result = kunit_parser.parse_run_tests(file.readlines())
self.assertEqual('kunit-resource-test', result.suites[0].name)
+class KUnitJsonTest(unittest.TestCase):
+
+ def _json_for(self, log_file):
+ with(open(get_absolute_path(log_file))) as file:
+ test_result = kunit_parser.parse_run_tests(file)
+ json_obj = kunit_json.get_json_result(
+ test_result=test_result,
+ def_config='kunit_defconfig',
+ build_dir=None,
+ json_path='stdout')
+ return json.loads(json_obj)
+
+ def test_failed_test_json(self):
+ result = self._json_for(
+ 'test_data/test_is_test_passed-failure.log')
+ self.assertEqual(
+ {'name': 'example_simple_test', 'status': 'FAIL'},
+ result["sub_groups"][1]["test_cases"][0])
+
+ def test_crashed_test_json(self):
+ result = self._json_for(
+ 'test_data/test_is_test_passed-crash.log')
+ self.assertEqual(
+ {'name': 'example_simple_test', 'status': 'ERROR'},
+ result["sub_groups"][1]["test_cases"][0])
+
+ def test_no_tests_json(self):
+ result = self._json_for(
+ 'test_data/test_is_test_passed-no_tests_run.log')
+ self.assertEqual(0, len(result['sub_groups']))
+
class StrContains(str):
def __eq__(self, other):
return self in other