1#!/usr/bin/env python3 2# SPDX-License-Identifier: GPL-2.0 3# 4# A thin wrapper on top of the KUnit Kernel 5# 6# Copyright (C) 2019, Google LLC. 7# Author: Felix Guo <felixguoxiuping@gmail.com> 8# Author: Brendan Higgins <brendanhiggins@google.com> 9 10import argparse 11import os 12import re 13import sys 14import time 15 16assert sys.version_info >= (3, 7), "Python version is too old" 17 18from collections import namedtuple 19from enum import Enum, auto 20from typing import Iterable, Sequence, List 21 22import kunit_json 23import kunit_kernel 24import kunit_parser 25 26KunitResult = namedtuple('KunitResult', ['status','result','elapsed_time']) 27 28KunitConfigRequest = namedtuple('KunitConfigRequest', 29 ['build_dir', 'make_options']) 30KunitBuildRequest = namedtuple('KunitBuildRequest', 31 ['jobs', 'build_dir', 'alltests', 32 'make_options']) 33KunitExecRequest = namedtuple('KunitExecRequest', 34 ['timeout', 'build_dir', 'alltests', 35 'filter_glob', 'kernel_args', 'run_isolated']) 36KunitParseRequest = namedtuple('KunitParseRequest', 37 ['raw_output', 'build_dir', 'json']) 38KunitRequest = namedtuple('KunitRequest', ['raw_output','timeout', 'jobs', 39 'build_dir', 'alltests', 'filter_glob', 40 'kernel_args', 'run_isolated', 'json', 'make_options']) 41 42KernelDirectoryPath = sys.argv[0].split('tools/testing/kunit/')[0] 43 44class KunitStatus(Enum): 45 SUCCESS = auto() 46 CONFIG_FAILURE = auto() 47 BUILD_FAILURE = auto() 48 TEST_FAILURE = auto() 49 50def get_kernel_root_path() -> str: 51 path = sys.argv[0] if not __file__ else __file__ 52 parts = os.path.realpath(path).split('tools/testing/kunit') 53 if len(parts) != 2: 54 sys.exit(1) 55 return parts[0] 56 57def config_tests(linux: kunit_kernel.LinuxSourceTree, 58 request: KunitConfigRequest) -> KunitResult: 59 kunit_parser.print_with_timestamp('Configuring KUnit Kernel ...') 60 61 config_start = time.time() 62 success = linux.build_reconfig(request.build_dir, request.make_options) 63 config_end = time.time() 64 if not success: 65 return KunitResult(KunitStatus.CONFIG_FAILURE, 66 'could not configure kernel', 67 config_end - config_start) 68 return KunitResult(KunitStatus.SUCCESS, 69 'configured kernel successfully', 70 config_end - config_start) 71 72def build_tests(linux: kunit_kernel.LinuxSourceTree, 73 request: KunitBuildRequest) -> KunitResult: 74 kunit_parser.print_with_timestamp('Building KUnit Kernel ...') 75 76 build_start = time.time() 77 success = linux.build_kernel(request.alltests, 78 request.jobs, 79 request.build_dir, 80 request.make_options) 81 build_end = time.time() 82 if not success: 83 return KunitResult(KunitStatus.BUILD_FAILURE, 84 'could not build kernel', 85 build_end - build_start) 86 if not success: 87 return KunitResult(KunitStatus.BUILD_FAILURE, 88 'could not build kernel', 89 build_end - build_start) 90 return KunitResult(KunitStatus.SUCCESS, 91 'built kernel successfully', 92 build_end - build_start) 93 94def _list_tests(linux: kunit_kernel.LinuxSourceTree, request: KunitExecRequest) -> List[str]: 95 args = ['kunit.action=list'] 96 if request.kernel_args: 97 args.extend(request.kernel_args) 98 99 output = linux.run_kernel(args=args, 100 timeout=None if request.alltests else request.timeout, 101 filter_glob=request.filter_glob, 102 build_dir=request.build_dir) 103 lines = kunit_parser.extract_tap_lines(output) 104 # Hack! Drop the dummy TAP version header that the executor prints out. 105 lines.pop() 106 107 # Filter out any extraneous non-test output that might have gotten mixed in. 108 return [l for l in lines if re.match('^[^\s.]+\.[^\s.]+$', l)] 109 110def _suites_from_test_list(tests: List[str]) -> List[str]: 111 """Extracts all the suites from an ordered list of tests.""" 112 suites = [] # type: List[str] 113 for t in tests: 114 parts = t.split('.', maxsplit=2) 115 if len(parts) != 2: 116 raise ValueError(f'internal KUnit error, test name should be of the form "<suite>.<test>", got "{t}"') 117 suite, case = parts 118 if not suites or suites[-1] != suite: 119 suites.append(suite) 120 return suites 121 122 123 124def exec_tests(linux: kunit_kernel.LinuxSourceTree, request: KunitExecRequest, 125 parse_request: KunitParseRequest) -> KunitResult: 126 filter_globs = [request.filter_glob] 127 if request.run_isolated: 128 tests = _list_tests(linux, request) 129 if request.run_isolated == 'test': 130 filter_globs = tests 131 if request.run_isolated == 'suite': 132 filter_globs = _suites_from_test_list(tests) 133 # Apply the test-part of the user's glob, if present. 134 if '.' in request.filter_glob: 135 test_glob = request.filter_glob.split('.', maxsplit=2)[1] 136 filter_globs = [g + '.'+ test_glob for g in filter_globs] 137 138 test_counts = kunit_parser.TestCounts() 139 exec_time = 0.0 140 for i, filter_glob in enumerate(filter_globs): 141 kunit_parser.print_with_timestamp('Starting KUnit Kernel ({}/{})...'.format(i+1, len(filter_globs))) 142 143 test_start = time.time() 144 run_result = linux.run_kernel( 145 args=request.kernel_args, 146 timeout=None if request.alltests else request.timeout, 147 filter_glob=filter_glob, 148 build_dir=request.build_dir) 149 150 result = parse_tests(parse_request, run_result) 151 # run_kernel() doesn't block on the kernel exiting. 152 # That only happens after we get the last line of output from `run_result`. 153 # So exec_time here actually contains parsing + execution time, which is fine. 154 test_end = time.time() 155 exec_time += test_end - test_start 156 157 test_counts.add_subtest_counts(result.result.test.counts) 158 159 kunit_status = _map_to_overall_status(test_counts.get_status()) 160 return KunitResult(status=kunit_status, result=result.result, elapsed_time=exec_time) 161 162def _map_to_overall_status(test_status: kunit_parser.TestStatus) -> KunitStatus: 163 if test_status in (kunit_parser.TestStatus.SUCCESS, kunit_parser.TestStatus.SKIPPED): 164 return KunitStatus.SUCCESS 165 else: 166 return KunitStatus.TEST_FAILURE 167 168def parse_tests(request: KunitParseRequest, input_data: Iterable[str]) -> KunitResult: 169 parse_start = time.time() 170 171 test_result = kunit_parser.TestResult(kunit_parser.TestStatus.SUCCESS, 172 kunit_parser.Test(), 173 'Tests not Parsed.') 174 175 if request.raw_output: 176 # Treat unparsed results as one passing test. 177 test_result.test.status = kunit_parser.TestStatus.SUCCESS 178 test_result.test.counts.passed = 1 179 180 output: Iterable[str] = input_data 181 if request.raw_output == 'all': 182 pass 183 elif request.raw_output == 'kunit': 184 output = kunit_parser.extract_tap_lines(output) 185 else: 186 print(f'Unknown --raw_output option "{request.raw_output}"', file=sys.stderr) 187 for line in output: 188 print(line.rstrip()) 189 190 else: 191 test_result = kunit_parser.parse_run_tests(input_data) 192 parse_end = time.time() 193 194 if request.json: 195 json_obj = kunit_json.get_json_result( 196 test_result=test_result, 197 def_config='kunit_defconfig', 198 build_dir=request.build_dir, 199 json_path=request.json) 200 if request.json == 'stdout': 201 print(json_obj) 202 203 if test_result.status != kunit_parser.TestStatus.SUCCESS: 204 return KunitResult(KunitStatus.TEST_FAILURE, test_result, 205 parse_end - parse_start) 206 207 return KunitResult(KunitStatus.SUCCESS, test_result, 208 parse_end - parse_start) 209 210def run_tests(linux: kunit_kernel.LinuxSourceTree, 211 request: KunitRequest) -> KunitResult: 212 run_start = time.time() 213 214 config_request = KunitConfigRequest(request.build_dir, 215 request.make_options) 216 config_result = config_tests(linux, config_request) 217 if config_result.status != KunitStatus.SUCCESS: 218 return config_result 219 220 build_request = KunitBuildRequest(request.jobs, request.build_dir, 221 request.alltests, 222 request.make_options) 223 build_result = build_tests(linux, build_request) 224 if build_result.status != KunitStatus.SUCCESS: 225 return build_result 226 227 exec_request = KunitExecRequest(request.timeout, request.build_dir, 228 request.alltests, request.filter_glob, 229 request.kernel_args, request.run_isolated) 230 parse_request = KunitParseRequest(request.raw_output, 231 request.build_dir, 232 request.json) 233 234 exec_result = exec_tests(linux, exec_request, parse_request) 235 236 run_end = time.time() 237 238 kunit_parser.print_with_timestamp(( 239 'Elapsed time: %.3fs total, %.3fs configuring, %.3fs ' + 240 'building, %.3fs running\n') % ( 241 run_end - run_start, 242 config_result.elapsed_time, 243 build_result.elapsed_time, 244 exec_result.elapsed_time)) 245 return exec_result 246 247# Problem: 248# $ kunit.py run --json 249# works as one would expect and prints the parsed test results as JSON. 250# $ kunit.py run --json suite_name 251# would *not* pass suite_name as the filter_glob and print as json. 252# argparse will consider it to be another way of writing 253# $ kunit.py run --json=suite_name 254# i.e. it would run all tests, and dump the json to a `suite_name` file. 255# So we hackily automatically rewrite --json => --json=stdout 256pseudo_bool_flag_defaults = { 257 '--json': 'stdout', 258 '--raw_output': 'kunit', 259} 260def massage_argv(argv: Sequence[str]) -> Sequence[str]: 261 def massage_arg(arg: str) -> str: 262 if arg not in pseudo_bool_flag_defaults: 263 return arg 264 return f'{arg}={pseudo_bool_flag_defaults[arg]}' 265 return list(map(massage_arg, argv)) 266 267def add_common_opts(parser) -> None: 268 parser.add_argument('--build_dir', 269 help='As in the make command, it specifies the build ' 270 'directory.', 271 type=str, default='.kunit', metavar='build_dir') 272 parser.add_argument('--make_options', 273 help='X=Y make option, can be repeated.', 274 action='append') 275 parser.add_argument('--alltests', 276 help='Run all KUnit tests through allyesconfig', 277 action='store_true') 278 parser.add_argument('--kunitconfig', 279 help='Path to Kconfig fragment that enables KUnit tests.' 280 ' If given a directory, (e.g. lib/kunit), "/.kunitconfig" ' 281 'will get automatically appended.', 282 metavar='kunitconfig') 283 284 parser.add_argument('--arch', 285 help=('Specifies the architecture to run tests under. ' 286 'The architecture specified here must match the ' 287 'string passed to the ARCH make param, ' 288 'e.g. i386, x86_64, arm, um, etc. Non-UML ' 289 'architectures run on QEMU.'), 290 type=str, default='um', metavar='arch') 291 292 parser.add_argument('--cross_compile', 293 help=('Sets make\'s CROSS_COMPILE variable; it should ' 294 'be set to a toolchain path prefix (the prefix ' 295 'of gcc and other tools in your toolchain, for ' 296 'example `sparc64-linux-gnu-` if you have the ' 297 'sparc toolchain installed on your system, or ' 298 '`$HOME/toolchains/microblaze/gcc-9.2.0-nolibc/microblaze-linux/bin/microblaze-linux-` ' 299 'if you have downloaded the microblaze toolchain ' 300 'from the 0-day website to a directory in your ' 301 'home directory called `toolchains`).'), 302 metavar='cross_compile') 303 304 parser.add_argument('--qemu_config', 305 help=('Takes a path to a path to a file containing ' 306 'a QemuArchParams object.'), 307 type=str, metavar='qemu_config') 308 309def add_build_opts(parser) -> None: 310 parser.add_argument('--jobs', 311 help='As in the make command, "Specifies the number of ' 312 'jobs (commands) to run simultaneously."', 313 type=int, default=8, metavar='jobs') 314 315def add_exec_opts(parser) -> None: 316 parser.add_argument('--timeout', 317 help='maximum number of seconds to allow for all tests ' 318 'to run. This does not include time taken to build the ' 319 'tests.', 320 type=int, 321 default=300, 322 metavar='timeout') 323 parser.add_argument('filter_glob', 324 help='Filter which KUnit test suites/tests run at ' 325 'boot-time, e.g. list* or list*.*del_test', 326 type=str, 327 nargs='?', 328 default='', 329 metavar='filter_glob') 330 parser.add_argument('--kernel_args', 331 help='Kernel command-line parameters. Maybe be repeated', 332 action='append') 333 parser.add_argument('--run_isolated', help='If set, boot the kernel for each ' 334 'individual suite/test. This is can be useful for debugging ' 335 'a non-hermetic test, one that might pass/fail based on ' 336 'what ran before it.', 337 type=str, 338 choices=['suite', 'test']), 339 340def add_parse_opts(parser) -> None: 341 parser.add_argument('--raw_output', help='If set don\'t format output from kernel. ' 342 'If set to --raw_output=kunit, filters to just KUnit output.', 343 type=str, nargs='?', const='all', default=None) 344 parser.add_argument('--json', 345 nargs='?', 346 help='Stores test results in a JSON, and either ' 347 'prints to stdout or saves to file if a ' 348 'filename is specified', 349 type=str, const='stdout', default=None) 350 351def main(argv, linux=None): 352 parser = argparse.ArgumentParser( 353 description='Helps writing and running KUnit tests.') 354 subparser = parser.add_subparsers(dest='subcommand') 355 356 # The 'run' command will config, build, exec, and parse in one go. 357 run_parser = subparser.add_parser('run', help='Runs KUnit tests.') 358 add_common_opts(run_parser) 359 add_build_opts(run_parser) 360 add_exec_opts(run_parser) 361 add_parse_opts(run_parser) 362 363 config_parser = subparser.add_parser('config', 364 help='Ensures that .config contains all of ' 365 'the options in .kunitconfig') 366 add_common_opts(config_parser) 367 368 build_parser = subparser.add_parser('build', help='Builds a kernel with KUnit tests') 369 add_common_opts(build_parser) 370 add_build_opts(build_parser) 371 372 exec_parser = subparser.add_parser('exec', help='Run a kernel with KUnit tests') 373 add_common_opts(exec_parser) 374 add_exec_opts(exec_parser) 375 add_parse_opts(exec_parser) 376 377 # The 'parse' option is special, as it doesn't need the kernel source 378 # (therefore there is no need for a build_dir, hence no add_common_opts) 379 # and the '--file' argument is not relevant to 'run', so isn't in 380 # add_parse_opts() 381 parse_parser = subparser.add_parser('parse', 382 help='Parses KUnit results from a file, ' 383 'and parses formatted results.') 384 add_parse_opts(parse_parser) 385 parse_parser.add_argument('file', 386 help='Specifies the file to read results from.', 387 type=str, nargs='?', metavar='input_file') 388 389 cli_args = parser.parse_args(massage_argv(argv)) 390 391 if get_kernel_root_path(): 392 os.chdir(get_kernel_root_path()) 393 394 if cli_args.subcommand == 'run': 395 if not os.path.exists(cli_args.build_dir): 396 os.mkdir(cli_args.build_dir) 397 398 if not linux: 399 linux = kunit_kernel.LinuxSourceTree(cli_args.build_dir, 400 kunitconfig_path=cli_args.kunitconfig, 401 arch=cli_args.arch, 402 cross_compile=cli_args.cross_compile, 403 qemu_config_path=cli_args.qemu_config) 404 405 request = KunitRequest(cli_args.raw_output, 406 cli_args.timeout, 407 cli_args.jobs, 408 cli_args.build_dir, 409 cli_args.alltests, 410 cli_args.filter_glob, 411 cli_args.kernel_args, 412 cli_args.run_isolated, 413 cli_args.json, 414 cli_args.make_options) 415 result = run_tests(linux, request) 416 if result.status != KunitStatus.SUCCESS: 417 sys.exit(1) 418 elif cli_args.subcommand == 'config': 419 if cli_args.build_dir and ( 420 not os.path.exists(cli_args.build_dir)): 421 os.mkdir(cli_args.build_dir) 422 423 if not linux: 424 linux = kunit_kernel.LinuxSourceTree(cli_args.build_dir, 425 kunitconfig_path=cli_args.kunitconfig, 426 arch=cli_args.arch, 427 cross_compile=cli_args.cross_compile, 428 qemu_config_path=cli_args.qemu_config) 429 430 request = KunitConfigRequest(cli_args.build_dir, 431 cli_args.make_options) 432 result = config_tests(linux, request) 433 kunit_parser.print_with_timestamp(( 434 'Elapsed time: %.3fs\n') % ( 435 result.elapsed_time)) 436 if result.status != KunitStatus.SUCCESS: 437 sys.exit(1) 438 elif cli_args.subcommand == 'build': 439 if not linux: 440 linux = kunit_kernel.LinuxSourceTree(cli_args.build_dir, 441 kunitconfig_path=cli_args.kunitconfig, 442 arch=cli_args.arch, 443 cross_compile=cli_args.cross_compile, 444 qemu_config_path=cli_args.qemu_config) 445 446 request = KunitBuildRequest(cli_args.jobs, 447 cli_args.build_dir, 448 cli_args.alltests, 449 cli_args.make_options) 450 result = build_tests(linux, request) 451 kunit_parser.print_with_timestamp(( 452 'Elapsed time: %.3fs\n') % ( 453 result.elapsed_time)) 454 if result.status != KunitStatus.SUCCESS: 455 sys.exit(1) 456 elif cli_args.subcommand == 'exec': 457 if not linux: 458 linux = kunit_kernel.LinuxSourceTree(cli_args.build_dir, 459 kunitconfig_path=cli_args.kunitconfig, 460 arch=cli_args.arch, 461 cross_compile=cli_args.cross_compile, 462 qemu_config_path=cli_args.qemu_config) 463 464 exec_request = KunitExecRequest(cli_args.timeout, 465 cli_args.build_dir, 466 cli_args.alltests, 467 cli_args.filter_glob, 468 cli_args.kernel_args, 469 cli_args.run_isolated) 470 parse_request = KunitParseRequest(cli_args.raw_output, 471 cli_args.build_dir, 472 cli_args.json) 473 result = exec_tests(linux, exec_request, parse_request) 474 kunit_parser.print_with_timestamp(( 475 'Elapsed time: %.3fs\n') % (result.elapsed_time)) 476 if result.status != KunitStatus.SUCCESS: 477 sys.exit(1) 478 elif cli_args.subcommand == 'parse': 479 if cli_args.file == None: 480 sys.stdin.reconfigure(errors='backslashreplace') # pytype: disable=attribute-error 481 kunit_output = sys.stdin 482 else: 483 with open(cli_args.file, 'r', errors='backslashreplace') as f: 484 kunit_output = f.read().splitlines() 485 request = KunitParseRequest(cli_args.raw_output, 486 None, 487 cli_args.json) 488 result = parse_tests(request, kunit_output) 489 if result.status != KunitStatus.SUCCESS: 490 sys.exit(1) 491 else: 492 parser.print_help() 493 494if __name__ == '__main__': 495 main(sys.argv[1:]) 496