testing/kunit/kunit.py

0001 #!/usr/bin/env python3
0002 # SPDX-License-Identifier: GPL-2.0
0003 #
0004 # A thin wrapper on top of the KUnit Kernel
0005 #
0006 # Copyright (C) 2019, Google LLC.
0007 # Author: Felix Guo <felixguoxiuping@gmail.com>
0008 # Author: Brendan Higgins <brendanhiggins@google.com>
0009
0010 import argparse
0011 import os
0012 import re
0013 import shlex
0014 import sys
0015 import time
0016
0017 assert sys.version_info >= (3, 7), "Python version is too old"
0018
0019 from dataclasses import dataclass
0020 from enum import Enum, auto
0021 from typing import Iterable, List, Optional, Sequence, Tuple
0022
0023 import kunit_json
0024 import kunit_kernel
0025 import kunit_parser
0026 from kunit_printer import stdout
0027
0028 class KunitStatus(Enum):
0029     SUCCESS = auto()
0030     CONFIG_FAILURE = auto()
0031     BUILD_FAILURE = auto()
0032     TEST_FAILURE = auto()
0033
0034 @dataclass
0035 class KunitResult:
0036     status: KunitStatus
0037     elapsed_time: float
0038
0039 @dataclass
0040 class KunitConfigRequest:
0041     build_dir: str
0042     make_options: Optional[List[str]]
0043
0044 @dataclass
0045 class KunitBuildRequest(KunitConfigRequest):
0046     jobs: int
0047     alltests: bool
0048
0049 @dataclass
0050 class KunitParseRequest:
0051     raw_output: Optional[str]
0052     json: Optional[str]
0053
0054 @dataclass
0055 class KunitExecRequest(KunitParseRequest):
0056     build_dir: str
0057     timeout: int
0058     alltests: bool
0059     filter_glob: str
0060     kernel_args: Optional[List[str]]
0061     run_isolated: Optional[str]
0062
0063 @dataclass
0064 class KunitRequest(KunitExecRequest, KunitBuildRequest):
0065     pass
0066
0067
0068 def get_kernel_root_path() -> str:
0069     path = sys.argv[0] if not __file__ else __file__
0070     parts = os.path.realpath(path).split('tools/testing/kunit')
0071     if len(parts) != 2:
0072         sys.exit(1)
0073     return parts[0]
0074
0075 def config_tests(linux: kunit_kernel.LinuxSourceTree,
0076          request: KunitConfigRequest) -> KunitResult:
0077     stdout.print_with_timestamp('Configuring KUnit Kernel ...')
0078
0079     config_start = time.time()
0080     success = linux.build_reconfig(request.build_dir, request.make_options)
0081     config_end = time.time()
0082     if not success:
0083         return KunitResult(KunitStatus.CONFIG_FAILURE,
0084                    config_end - config_start)
0085     return KunitResult(KunitStatus.SUCCESS,
0086                config_end - config_start)
0087
0088 def build_tests(linux: kunit_kernel.LinuxSourceTree,
0089         request: KunitBuildRequest) -> KunitResult:
0090     stdout.print_with_timestamp('Building KUnit Kernel ...')
0091
0092     build_start = time.time()
0093     success = linux.build_kernel(request.alltests,
0094                      request.jobs,
0095                      request.build_dir,
0096                      request.make_options)
0097     build_end = time.time()
0098     if not success:
0099         return KunitResult(KunitStatus.BUILD_FAILURE,
0100                    build_end - build_start)
0101     if not success:
0102         return KunitResult(KunitStatus.BUILD_FAILURE,
0103                    build_end - build_start)
0104     return KunitResult(KunitStatus.SUCCESS,
0105                build_end - build_start)
0106
0107 def config_and_build_tests(linux: kunit_kernel.LinuxSourceTree,
0108                request: KunitBuildRequest) -> KunitResult:
0109     config_result = config_tests(linux, request)
0110     if config_result.status != KunitStatus.SUCCESS:
0111         return config_result
0112
0113     return build_tests(linux, request)
0114
0115 def _list_tests(linux: kunit_kernel.LinuxSourceTree, request: KunitExecRequest) -> List[str]:
0116     args = ['kunit.action=list']
0117     if request.kernel_args:
0118         args.extend(request.kernel_args)
0119
0120     output = linux.run_kernel(args=args,
0121                timeout=None if request.alltests else request.timeout,
0122                filter_glob=request.filter_glob,
0123                build_dir=request.build_dir)
0124     lines = kunit_parser.extract_tap_lines(output)
0125     # Hack! Drop the dummy TAP version header that the executor prints out.
0126     lines.pop()
0127
0128     # Filter out any extraneous non-test output that might have gotten mixed in.
0129     return [l for l in lines if re.match(r'^[^\s.]+\.[^\s.]+$', l)]
0130
0131 def _suites_from_test_list(tests: List[str]) -> List[str]:
0132     """Extracts all the suites from an ordered list of tests."""
0133     suites = []  # type: List[str]
0134     for t in tests:
0135         parts = t.split('.', maxsplit=2)
0136         if len(parts) != 2:
0137             raise ValueError(f'internal KUnit error, test name should be of the form "<suite>.<test>", got "{t}"')
0138         suite, case = parts
0139         if not suites or suites[-1] != suite:
0140             suites.append(suite)
0141     return suites
0142
0143
0144
0145 def exec_tests(linux: kunit_kernel.LinuxSourceTree, request: KunitExecRequest) -> KunitResult:
0146     filter_globs = [request.filter_glob]
0147     if request.run_isolated:
0148         tests = _list_tests(linux, request)
0149         if request.run_isolated == 'test':
0150             filter_globs = tests
0151         if request.run_isolated == 'suite':
0152             filter_globs = _suites_from_test_list(tests)
0153             # Apply the test-part of the user's glob, if present.
0154             if '.' in request.filter_glob:
0155                 test_glob = request.filter_glob.split('.', maxsplit=2)[1]
0156                 filter_globs = [g + '.'+ test_glob for g in filter_globs]
0157
0158     metadata = kunit_json.Metadata(arch=linux.arch(), build_dir=request.build_dir, def_config='kunit_defconfig')
0159
0160     test_counts = kunit_parser.TestCounts()
0161     exec_time = 0.0
0162     for i, filter_glob in enumerate(filter_globs):
0163         stdout.print_with_timestamp('Starting KUnit Kernel ({}/{})...'.format(i+1, len(filter_globs)))
0164
0165         test_start = time.time()
0166         run_result = linux.run_kernel(
0167             args=request.kernel_args,
0168             timeout=None if request.alltests else request.timeout,
0169             filter_glob=filter_glob,
0170             build_dir=request.build_dir)
0171
0172         _, test_result = parse_tests(request, metadata, run_result)
0173         # run_kernel() doesn't block on the kernel exiting.
0174         # That only happens after we get the last line of output from `run_result`.
0175         # So exec_time here actually contains parsing + execution time, which is fine.
0176         test_end = time.time()
0177         exec_time += test_end - test_start
0178
0179         test_counts.add_subtest_counts(test_result.counts)
0180
0181     if len(filter_globs) == 1 and test_counts.crashed > 0:
0182         bd = request.build_dir
0183         print('The kernel seems to have crashed; you can decode the stack traces with:')
0184         print('$ scripts/decode_stacktrace.sh {}/vmlinux {} < {} | tee {}/decoded.log | {} parse'.format(
0185                 bd, bd, kunit_kernel.get_outfile_path(bd), bd, sys.argv[0]))
0186
0187     kunit_status = _map_to_overall_status(test_counts.get_status())
0188     return KunitResult(status=kunit_status, elapsed_time=exec_time)
0189
0190 def _map_to_overall_status(test_status: kunit_parser.TestStatus) -> KunitStatus:
0191     if test_status in (kunit_parser.TestStatus.SUCCESS, kunit_parser.TestStatus.SKIPPED):
0192         return KunitStatus.SUCCESS
0193     return KunitStatus.TEST_FAILURE
0194
0195 def parse_tests(request: KunitParseRequest, metadata: kunit_json.Metadata, input_data: Iterable[str]) -> Tuple[KunitResult, kunit_parser.Test]:
0196     parse_start = time.time()
0197
0198     test_result = kunit_parser.Test()
0199
0200     if request.raw_output:
0201         # Treat unparsed results as one passing test.
0202         test_result.status = kunit_parser.TestStatus.SUCCESS
0203         test_result.counts.passed = 1
0204
0205         output: Iterable[str] = input_data
0206         if request.raw_output == 'all':
0207             pass
0208         elif request.raw_output == 'kunit':
0209             output = kunit_parser.extract_tap_lines(output)
0210         for line in output:
0211             print(line.rstrip())
0212
0213     else:
0214         test_result = kunit_parser.parse_run_tests(input_data)
0215     parse_end = time.time()
0216
0217     if request.json:
0218         json_str = kunit_json.get_json_result(
0219                     test=test_result,
0220                     metadata=metadata)
0221         if request.json == 'stdout':
0222             print(json_str)
0223         else:
0224             with open(request.json, 'w') as f:
0225                 f.write(json_str)
0226             stdout.print_with_timestamp("Test results stored in %s" %
0227                 os.path.abspath(request.json))
0228
0229     if test_result.status != kunit_parser.TestStatus.SUCCESS:
0230         return KunitResult(KunitStatus.TEST_FAILURE, parse_end - parse_start), test_result
0231
0232     return KunitResult(KunitStatus.SUCCESS, parse_end - parse_start), test_result
0233
0234 def run_tests(linux: kunit_kernel.LinuxSourceTree,
0235           request: KunitRequest) -> KunitResult:
0236     run_start = time.time()
0237
0238     config_result = config_tests(linux, request)
0239     if config_result.status != KunitStatus.SUCCESS:
0240         return config_result
0241
0242     build_result = build_tests(linux, request)
0243     if build_result.status != KunitStatus.SUCCESS:
0244         return build_result
0245
0246     exec_result = exec_tests(linux, request)
0247
0248     run_end = time.time()
0249
0250     stdout.print_with_timestamp((
0251         'Elapsed time: %.3fs total, %.3fs configuring, %.3fs ' +
0252         'building, %.3fs running\n') % (
0253                 run_end - run_start,
0254                 config_result.elapsed_time,
0255                 build_result.elapsed_time,
0256                 exec_result.elapsed_time))
0257     return exec_result
0258
0259 # Problem:
0260 # $ kunit.py run --json
0261 # works as one would expect and prints the parsed test results as JSON.
0262 # $ kunit.py run --json suite_name
0263 # would *not* pass suite_name as the filter_glob and print as json.
0264 # argparse will consider it to be another way of writing
0265 # $ kunit.py run --json=suite_name
0266 # i.e. it would run all tests, and dump the json to a `suite_name` file.
0267 # So we hackily automatically rewrite --json => --json=stdout
0268 pseudo_bool_flag_defaults = {
0269         '--json': 'stdout',
0270         '--raw_output': 'kunit',
0271 }
0272 def massage_argv(argv: Sequence[str]) -> Sequence[str]:
0273     def massage_arg(arg: str) -> str:
0274         if arg not in pseudo_bool_flag_defaults:
0275             return arg
0276         return  f'{arg}={pseudo_bool_flag_defaults[arg]}'
0277     return list(map(massage_arg, argv))
0278
0279 def get_default_jobs() -> int:
0280     return len(os.sched_getaffinity(0))
0281
0282 def add_common_opts(parser) -> None:
0283     parser.add_argument('--build_dir',
0284                 help='As in the make command, it specifies the build '
0285                 'directory.',
0286                 type=str, default='.kunit', metavar='DIR')
0287     parser.add_argument('--make_options',
0288                 help='X=Y make option, can be repeated.',
0289                 action='append', metavar='X=Y')
0290     parser.add_argument('--alltests',
0291                 help='Run all KUnit tests through allyesconfig',
0292                 action='store_true')
0293     parser.add_argument('--kunitconfig',
0294                  help='Path to Kconfig fragment that enables KUnit tests.'
0295                  ' If given a directory, (e.g. lib/kunit), "/.kunitconfig" '
0296                  'will get  automatically appended. If repeated, the files '
0297                  'blindly concatenated, which might not work in all cases.',
0298                  action='append', metavar='PATHS')
0299     parser.add_argument('--kconfig_add',
0300                  help='Additional Kconfig options to append to the '
0301                  '.kunitconfig, e.g. CONFIG_KASAN=y. Can be repeated.',
0302                 action='append', metavar='CONFIG_X=Y')
0303
0304     parser.add_argument('--arch',
0305                 help=('Specifies the architecture to run tests under. '
0306                   'The architecture specified here must match the '
0307                   'string passed to the ARCH make param, '
0308                   'e.g. i386, x86_64, arm, um, etc. Non-UML '
0309                   'architectures run on QEMU.'),
0310                 type=str, default='um', metavar='ARCH')
0311
0312     parser.add_argument('--cross_compile',
0313                 help=('Sets make\'s CROSS_COMPILE variable; it should '
0314                   'be set to a toolchain path prefix (the prefix '
0315                   'of gcc and other tools in your toolchain, for '
0316                   'example `sparc64-linux-gnu-` if you have the '
0317                   'sparc toolchain installed on your system, or '
0318                   '`$HOME/toolchains/microblaze/gcc-9.2.0-nolibc/microblaze-linux/bin/microblaze-linux-` '
0319                   'if you have downloaded the microblaze toolchain '
0320                   'from the 0-day website to a directory in your '
0321                   'home directory called `toolchains`).'),
0322                 metavar='PREFIX')
0323
0324     parser.add_argument('--qemu_config',
0325                 help=('Takes a path to a path to a file containing '
0326                   'a QemuArchParams object.'),
0327                 type=str, metavar='FILE')
0328
0329     parser.add_argument('--qemu_args',
0330                 help='Additional QEMU arguments, e.g. "-smp 8"',
0331                 action='append', metavar='')
0332
0333 def add_build_opts(parser) -> None:
0334     parser.add_argument('--jobs',
0335                 help='As in the make command, "Specifies  the number of '
0336                 'jobs (commands) to run simultaneously."',
0337                 type=int, default=get_default_jobs(), metavar='N')
0338
0339 def add_exec_opts(parser) -> None:
0340     parser.add_argument('--timeout',
0341                 help='maximum number of seconds to allow for all tests '
0342                 'to run. This does not include time taken to build the '
0343                 'tests.',
0344                 type=int,
0345                 default=300,
0346                 metavar='SECONDS')
0347     parser.add_argument('filter_glob',
0348                 help='Filter which KUnit test suites/tests run at '
0349                 'boot-time, e.g. list* or list*.*del_test',
0350                 type=str,
0351                 nargs='?',
0352                 default='',
0353                 metavar='filter_glob')
0354     parser.add_argument('--kernel_args',
0355                 help='Kernel command-line parameters. Maybe be repeated',
0356                  action='append', metavar='')
0357     parser.add_argument('--run_isolated', help='If set, boot the kernel for each '
0358                 'individual suite/test. This is can be useful for debugging '
0359                 'a non-hermetic test, one that might pass/fail based on '
0360                 'what ran before it.',
0361                 type=str,
0362                 choices=['suite', 'test'])
0363
0364 def add_parse_opts(parser) -> None:
0365     parser.add_argument('--raw_output', help='If set don\'t format output from kernel. '
0366                 'If set to --raw_output=kunit, filters to just KUnit output.',
0367                  type=str, nargs='?', const='all', default=None, choices=['all', 'kunit'])
0368     parser.add_argument('--json',
0369                 nargs='?',
0370                 help='Stores test results in a JSON, and either '
0371                 'prints to stdout or saves to file if a '
0372                 'filename is specified',
0373                 type=str, const='stdout', default=None, metavar='FILE')
0374
0375
0376 def tree_from_args(cli_args: argparse.Namespace) -> kunit_kernel.LinuxSourceTree:
0377     """Returns a LinuxSourceTree based on the user's arguments."""
0378     # Allow users to specify multiple arguments in one string, e.g. '-smp 8'
0379     qemu_args: List[str] = []
0380     if cli_args.qemu_args:
0381         for arg in cli_args.qemu_args:
0382             qemu_args.extend(shlex.split(arg))
0383
0384     return kunit_kernel.LinuxSourceTree(cli_args.build_dir,
0385             kunitconfig_paths=cli_args.kunitconfig,
0386             kconfig_add=cli_args.kconfig_add,
0387             arch=cli_args.arch,
0388             cross_compile=cli_args.cross_compile,
0389             qemu_config_path=cli_args.qemu_config,
0390             extra_qemu_args=qemu_args)
0391
0392
0393 def main(argv):
0394     parser = argparse.ArgumentParser(
0395             description='Helps writing and running KUnit tests.')
0396     subparser = parser.add_subparsers(dest='subcommand')
0397
0398     # The 'run' command will config, build, exec, and parse in one go.
0399     run_parser = subparser.add_parser('run', help='Runs KUnit tests.')
0400     add_common_opts(run_parser)
0401     add_build_opts(run_parser)
0402     add_exec_opts(run_parser)
0403     add_parse_opts(run_parser)
0404
0405     config_parser = subparser.add_parser('config',
0406                         help='Ensures that .config contains all of '
0407                         'the options in .kunitconfig')
0408     add_common_opts(config_parser)
0409
0410     build_parser = subparser.add_parser('build', help='Builds a kernel with KUnit tests')
0411     add_common_opts(build_parser)
0412     add_build_opts(build_parser)
0413
0414     exec_parser = subparser.add_parser('exec', help='Run a kernel with KUnit tests')
0415     add_common_opts(exec_parser)
0416     add_exec_opts(exec_parser)
0417     add_parse_opts(exec_parser)
0418
0419     # The 'parse' option is special, as it doesn't need the kernel source
0420     # (therefore there is no need for a build_dir, hence no add_common_opts)
0421     # and the '--file' argument is not relevant to 'run', so isn't in
0422     # add_parse_opts()
0423     parse_parser = subparser.add_parser('parse',
0424                         help='Parses KUnit results from a file, '
0425                         'and parses formatted results.')
0426     add_parse_opts(parse_parser)
0427     parse_parser.add_argument('file',
0428                   help='Specifies the file to read results from.',
0429                   type=str, nargs='?', metavar='input_file')
0430
0431     cli_args = parser.parse_args(massage_argv(argv))
0432
0433     if get_kernel_root_path():
0434         os.chdir(get_kernel_root_path())
0435
0436     if cli_args.subcommand == 'run':
0437         if not os.path.exists(cli_args.build_dir):
0438             os.mkdir(cli_args.build_dir)
0439
0440         linux = tree_from_args(cli_args)
0441         request = KunitRequest(build_dir=cli_args.build_dir,
0442                        make_options=cli_args.make_options,
0443                        jobs=cli_args.jobs,
0444                        alltests=cli_args.alltests,
0445                        raw_output=cli_args.raw_output,
0446                        json=cli_args.json,
0447                        timeout=cli_args.timeout,
0448                        filter_glob=cli_args.filter_glob,
0449                        kernel_args=cli_args.kernel_args,
0450                        run_isolated=cli_args.run_isolated)
0451         result = run_tests(linux, request)
0452         if result.status != KunitStatus.SUCCESS:
0453             sys.exit(1)
0454     elif cli_args.subcommand == 'config':
0455         if cli_args.build_dir and (
0456                 not os.path.exists(cli_args.build_dir)):
0457             os.mkdir(cli_args.build_dir)
0458
0459         linux = tree_from_args(cli_args)
0460         request = KunitConfigRequest(build_dir=cli_args.build_dir,
0461                          make_options=cli_args.make_options)
0462         result = config_tests(linux, request)
0463         stdout.print_with_timestamp((
0464             'Elapsed time: %.3fs\n') % (
0465                 result.elapsed_time))
0466         if result.status != KunitStatus.SUCCESS:
0467             sys.exit(1)
0468     elif cli_args.subcommand == 'build':
0469         linux = tree_from_args(cli_args)
0470         request = KunitBuildRequest(build_dir=cli_args.build_dir,
0471                         make_options=cli_args.make_options,
0472                         jobs=cli_args.jobs,
0473                         alltests=cli_args.alltests)
0474         result = config_and_build_tests(linux, request)
0475         stdout.print_with_timestamp((
0476             'Elapsed time: %.3fs\n') % (
0477                 result.elapsed_time))
0478         if result.status != KunitStatus.SUCCESS:
0479             sys.exit(1)
0480     elif cli_args.subcommand == 'exec':
0481         linux = tree_from_args(cli_args)
0482         exec_request = KunitExecRequest(raw_output=cli_args.raw_output,
0483                         build_dir=cli_args.build_dir,
0484                         json=cli_args.json,
0485                         timeout=cli_args.timeout,
0486                         alltests=cli_args.alltests,
0487                         filter_glob=cli_args.filter_glob,
0488                         kernel_args=cli_args.kernel_args,
0489                         run_isolated=cli_args.run_isolated)
0490         result = exec_tests(linux, exec_request)
0491         stdout.print_with_timestamp((
0492             'Elapsed time: %.3fs\n') % (result.elapsed_time))
0493         if result.status != KunitStatus.SUCCESS:
0494             sys.exit(1)
0495     elif cli_args.subcommand == 'parse':
0496         if cli_args.file is None:
0497             sys.stdin.reconfigure(errors='backslashreplace')  # pytype: disable=attribute-error
0498             kunit_output = sys.stdin
0499         else:
0500             with open(cli_args.file, 'r', errors='backslashreplace') as f:
0501                 kunit_output = f.read().splitlines()
0502         # We know nothing about how the result was created!
0503         metadata = kunit_json.Metadata()
0504         request = KunitParseRequest(raw_output=cli_args.raw_output,
0505                         json=cli_args.json)
0506         result, _ = parse_tests(request, metadata, kunit_output)
0507         if result.status != KunitStatus.SUCCESS:
0508             sys.exit(1)
0509     else:
0510         parser.print_help()
0511
0512 if __name__ == '__main__':
0513     main(sys.argv[1:])