2 # SPDX-License-Identifier: GPL-2.0
4 # A thin wrapper on top of the KUnit Kernel
6 # Copyright (C) 2019, Google LLC.
7 # Author: Felix Guo <felixguoxiuping@gmail.com>
8 # Author: Brendan Higgins <brendanhiggins@google.com>
17 assert sys
.version_info
>= (3, 7), "Python version is too old"
19 from dataclasses
import dataclass
20 from enum
import Enum
, auto
21 from typing
import Iterable
, List
, Optional
, Sequence
, Tuple
26 from kunit_printer
import stdout
, null_printer
28 class KunitStatus(Enum
):
30 CONFIG_FAILURE
= auto()
31 BUILD_FAILURE
= auto()
40 class KunitConfigRequest
:
42 make_options
: Optional
[List
[str]]
45 class KunitBuildRequest(KunitConfigRequest
):
49 class KunitParseRequest
:
50 raw_output
: Optional
[str]
56 class KunitExecRequest(KunitParseRequest
):
61 filter_action
: Optional
[str]
62 kernel_args
: Optional
[List
[str]]
63 run_isolated
: Optional
[str]
68 class KunitRequest(KunitExecRequest
, KunitBuildRequest
):
72 def get_kernel_root_path() -> str:
73 path
= sys
.argv
[0] if not __file__
else __file__
74 parts
= os
.path
.realpath(path
).split('tools/testing/kunit')
79 def config_tests(linux
: kunit_kernel
.LinuxSourceTree
,
80 request
: KunitConfigRequest
) -> KunitResult
:
81 stdout
.print_with_timestamp('Configuring KUnit Kernel ...')
83 config_start
= time
.time()
84 success
= linux
.build_reconfig(request
.build_dir
, request
.make_options
)
85 config_end
= time
.time()
86 status
= KunitStatus
.SUCCESS
if success
else KunitStatus
.CONFIG_FAILURE
87 return KunitResult(status
, config_end
- config_start
)
89 def build_tests(linux
: kunit_kernel
.LinuxSourceTree
,
90 request
: KunitBuildRequest
) -> KunitResult
:
91 stdout
.print_with_timestamp('Building KUnit Kernel ...')
93 build_start
= time
.time()
94 success
= linux
.build_kernel(request
.jobs
,
97 build_end
= time
.time()
98 status
= KunitStatus
.SUCCESS
if success
else KunitStatus
.BUILD_FAILURE
99 return KunitResult(status
, build_end
- build_start
)
101 def config_and_build_tests(linux
: kunit_kernel
.LinuxSourceTree
,
102 request
: KunitBuildRequest
) -> KunitResult
:
103 config_result
= config_tests(linux
, request
)
104 if config_result
.status
!= KunitStatus
.SUCCESS
:
107 return build_tests(linux
, request
)
109 def _list_tests(linux
: kunit_kernel
.LinuxSourceTree
, request
: KunitExecRequest
) -> List
[str]:
110 args
= ['kunit.action=list']
112 if request
.kernel_args
:
113 args
.extend(request
.kernel_args
)
115 output
= linux
.run_kernel(args
=args
,
116 timeout
=request
.timeout
,
117 filter_glob
=request
.filter_glob
,
118 filter=request
.filter,
119 filter_action
=request
.filter_action
,
120 build_dir
=request
.build_dir
)
121 lines
= kunit_parser
.extract_tap_lines(output
)
122 # Hack! Drop the dummy TAP version header that the executor prints out.
125 # Filter out any extraneous non-test output that might have gotten mixed in.
126 return [l
for l
in output
if re
.match(r
'^[^\s.]+\.[^\s.]+$', l
)]
128 def _list_tests_attr(linux
: kunit_kernel
.LinuxSourceTree
, request
: KunitExecRequest
) -> Iterable
[str]:
129 args
= ['kunit.action=list_attr']
131 if request
.kernel_args
:
132 args
.extend(request
.kernel_args
)
134 output
= linux
.run_kernel(args
=args
,
135 timeout
=request
.timeout
,
136 filter_glob
=request
.filter_glob
,
137 filter=request
.filter,
138 filter_action
=request
.filter_action
,
139 build_dir
=request
.build_dir
)
140 lines
= kunit_parser
.extract_tap_lines(output
)
141 # Hack! Drop the dummy TAP version header that the executor prints out.
144 # Filter out any extraneous non-test output that might have gotten mixed in.
147 def _suites_from_test_list(tests
: List
[str]) -> List
[str]:
148 """Extracts all the suites from an ordered list of tests."""
149 suites
= [] # type: List[str]
151 parts
= t
.split('.', maxsplit
=2)
153 raise ValueError(f
'internal KUnit error, test name should be of the form "<suite>.<test>", got "{t}"')
155 if not suites
or suites
[-1] != suite
:
159 def exec_tests(linux
: kunit_kernel
.LinuxSourceTree
, request
: KunitExecRequest
) -> KunitResult
:
160 filter_globs
= [request
.filter_glob
]
161 if request
.list_tests
:
162 output
= _list_tests(linux
, request
)
165 return KunitResult(status
=KunitStatus
.SUCCESS
, elapsed_time
=0.0)
166 if request
.list_tests_attr
:
167 attr_output
= _list_tests_attr(linux
, request
)
168 for line
in attr_output
:
170 return KunitResult(status
=KunitStatus
.SUCCESS
, elapsed_time
=0.0)
171 if request
.run_isolated
:
172 tests
= _list_tests(linux
, request
)
173 if request
.run_isolated
== 'test':
175 elif request
.run_isolated
== 'suite':
176 filter_globs
= _suites_from_test_list(tests
)
177 # Apply the test-part of the user's glob, if present.
178 if '.' in request
.filter_glob
:
179 test_glob
= request
.filter_glob
.split('.', maxsplit
=2)[1]
180 filter_globs
= [g
+ '.'+ test_glob
for g
in filter_globs
]
182 metadata
= kunit_json
.Metadata(arch
=linux
.arch(), build_dir
=request
.build_dir
, def_config
='kunit_defconfig')
184 test_counts
= kunit_parser
.TestCounts()
186 for i
, filter_glob
in enumerate(filter_globs
):
187 stdout
.print_with_timestamp('Starting KUnit Kernel ({}/{})...'.format(i
+1, len(filter_globs
)))
189 test_start
= time
.time()
190 run_result
= linux
.run_kernel(
191 args
=request
.kernel_args
,
192 timeout
=request
.timeout
,
193 filter_glob
=filter_glob
,
194 filter=request
.filter,
195 filter_action
=request
.filter_action
,
196 build_dir
=request
.build_dir
)
198 _
, test_result
= parse_tests(request
, metadata
, run_result
)
199 # run_kernel() doesn't block on the kernel exiting.
200 # That only happens after we get the last line of output from `run_result`.
201 # So exec_time here actually contains parsing + execution time, which is fine.
202 test_end
= time
.time()
203 exec_time
+= test_end
- test_start
205 test_counts
.add_subtest_counts(test_result
.counts
)
207 if len(filter_globs
) == 1 and test_counts
.crashed
> 0:
208 bd
= request
.build_dir
209 print('The kernel seems to have crashed; you can decode the stack traces with:')
210 print('$ scripts/decode_stacktrace.sh {}/vmlinux {} < {} | tee {}/decoded.log | {} parse'.format(
211 bd
, bd
, kunit_kernel
.get_outfile_path(bd
), bd
, sys
.argv
[0]))
213 kunit_status
= _map_to_overall_status(test_counts
.get_status())
214 return KunitResult(status
=kunit_status
, elapsed_time
=exec_time
)
216 def _map_to_overall_status(test_status
: kunit_parser
.TestStatus
) -> KunitStatus
:
217 if test_status
in (kunit_parser
.TestStatus
.SUCCESS
, kunit_parser
.TestStatus
.SKIPPED
):
218 return KunitStatus
.SUCCESS
219 return KunitStatus
.TEST_FAILURE
221 def parse_tests(request
: KunitParseRequest
, metadata
: kunit_json
.Metadata
, input_data
: Iterable
[str]) -> Tuple
[KunitResult
, kunit_parser
.Test
]:
222 parse_start
= time
.time()
224 if request
.raw_output
:
225 # Treat unparsed results as one passing test.
226 fake_test
= kunit_parser
.Test()
227 fake_test
.status
= kunit_parser
.TestStatus
.SUCCESS
228 fake_test
.counts
.passed
= 1
230 output
: Iterable
[str] = input_data
231 if request
.raw_output
== 'all':
233 elif request
.raw_output
== 'kunit':
234 output
= kunit_parser
.extract_tap_lines(output
)
237 parse_time
= time
.time() - parse_start
238 return KunitResult(KunitStatus
.SUCCESS
, parse_time
), fake_test
240 default_printer
= stdout
241 if request
.summary
or request
.failed
:
242 default_printer
= null_printer
244 # Actually parse the test results.
245 test
= kunit_parser
.parse_run_tests(input_data
, default_printer
)
246 parse_time
= time
.time() - parse_start
249 kunit_parser
.print_test(test
, request
.failed
, stdout
)
250 kunit_parser
.print_summary_line(test
, stdout
)
253 json_str
= kunit_json
.get_json_result(
256 if request
.json
== 'stdout':
259 with
open(request
.json
, 'w') as f
:
261 stdout
.print_with_timestamp("Test results stored in %s" %
262 os
.path
.abspath(request
.json
))
264 if test
.status
!= kunit_parser
.TestStatus
.SUCCESS
:
265 return KunitResult(KunitStatus
.TEST_FAILURE
, parse_time
), test
267 return KunitResult(KunitStatus
.SUCCESS
, parse_time
), test
269 def run_tests(linux
: kunit_kernel
.LinuxSourceTree
,
270 request
: KunitRequest
) -> KunitResult
:
271 run_start
= time
.time()
273 config_result
= config_tests(linux
, request
)
274 if config_result
.status
!= KunitStatus
.SUCCESS
:
277 build_result
= build_tests(linux
, request
)
278 if build_result
.status
!= KunitStatus
.SUCCESS
:
281 exec_result
= exec_tests(linux
, request
)
283 run_end
= time
.time()
285 stdout
.print_with_timestamp((
286 'Elapsed time: %.3fs total, %.3fs configuring, %.3fs ' +
287 'building, %.3fs running\n') % (
289 config_result
.elapsed_time
,
290 build_result
.elapsed_time
,
291 exec_result
.elapsed_time
))
295 # $ kunit.py run --json
296 # works as one would expect and prints the parsed test results as JSON.
297 # $ kunit.py run --json suite_name
298 # would *not* pass suite_name as the filter_glob and print as json.
299 # argparse will consider it to be another way of writing
300 # $ kunit.py run --json=suite_name
301 # i.e. it would run all tests, and dump the json to a `suite_name` file.
302 # So we hackily automatically rewrite --json => --json=stdout
303 pseudo_bool_flag_defaults
= {
305 '--raw_output': 'kunit',
307 def massage_argv(argv
: Sequence
[str]) -> Sequence
[str]:
308 def massage_arg(arg
: str) -> str:
309 if arg
not in pseudo_bool_flag_defaults
:
311 return f
'{arg}={pseudo_bool_flag_defaults[arg]}'
312 return list(map(massage_arg
, argv
))
314 def get_default_jobs() -> int:
315 return len(os
.sched_getaffinity(0))
317 def add_common_opts(parser
: argparse
.ArgumentParser
) -> None:
318 parser
.add_argument('--build_dir',
319 help='As in the make command, it specifies the build '
321 type=str, default
='.kunit', metavar
='DIR')
322 parser
.add_argument('--make_options',
323 help='X=Y make option, can be repeated.',
324 action
='append', metavar
='X=Y')
325 parser
.add_argument('--alltests',
326 help='Run all KUnit tests via tools/testing/kunit/configs/all_tests.config',
328 parser
.add_argument('--kunitconfig',
329 help='Path to Kconfig fragment that enables KUnit tests.'
330 ' If given a directory, (e.g. lib/kunit), "/.kunitconfig" '
331 'will get automatically appended. If repeated, the files '
332 'blindly concatenated, which might not work in all cases.',
333 action
='append', metavar
='PATHS')
334 parser
.add_argument('--kconfig_add',
335 help='Additional Kconfig options to append to the '
336 '.kunitconfig, e.g. CONFIG_KASAN=y. Can be repeated.',
337 action
='append', metavar
='CONFIG_X=Y')
339 parser
.add_argument('--arch',
340 help=('Specifies the architecture to run tests under. '
341 'The architecture specified here must match the '
342 'string passed to the ARCH make param, '
343 'e.g. i386, x86_64, arm, um, etc. Non-UML '
344 'architectures run on QEMU.'),
345 type=str, default
='um', metavar
='ARCH')
347 parser
.add_argument('--cross_compile',
348 help=('Sets make\'s CROSS_COMPILE variable; it should '
349 'be set to a toolchain path prefix (the prefix '
350 'of gcc and other tools in your toolchain, for '
351 'example `sparc64-linux-gnu-` if you have the '
352 'sparc toolchain installed on your system, or '
353 '`$HOME/toolchains/microblaze/gcc-9.2.0-nolibc/microblaze-linux/bin/microblaze-linux-` '
354 'if you have downloaded the microblaze toolchain '
355 'from the 0-day website to a directory in your '
356 'home directory called `toolchains`).'),
359 parser
.add_argument('--qemu_config',
360 help=('Takes a path to a path to a file containing '
361 'a QemuArchParams object.'),
362 type=str, metavar
='FILE')
364 parser
.add_argument('--qemu_args',
365 help='Additional QEMU arguments, e.g. "-smp 8"',
366 action
='append', metavar
='')
368 def add_build_opts(parser
: argparse
.ArgumentParser
) -> None:
369 parser
.add_argument('--jobs',
370 help='As in the make command, "Specifies the number of '
371 'jobs (commands) to run simultaneously."',
372 type=int, default
=get_default_jobs(), metavar
='N')
374 def add_exec_opts(parser
: argparse
.ArgumentParser
) -> None:
375 parser
.add_argument('--timeout',
376 help='maximum number of seconds to allow for all tests '
377 'to run. This does not include time taken to build the '
382 parser
.add_argument('filter_glob',
383 help='Filter which KUnit test suites/tests run at '
384 'boot-time, e.g. list* or list*.*del_test',
388 metavar
='filter_glob')
389 parser
.add_argument('--filter',
390 help='Filter KUnit tests with attributes, '
391 'e.g. module=example or speed>slow',
394 parser
.add_argument('--filter_action',
395 help='If set to skip, filtered tests will be skipped, '
396 'e.g. --filter_action=skip. Otherwise they will not run.',
399 parser
.add_argument('--kernel_args',
400 help='Kernel command-line parameters. Maybe be repeated',
401 action
='append', metavar
='')
402 parser
.add_argument('--run_isolated', help='If set, boot the kernel for each '
403 'individual suite/test. This is can be useful for debugging '
404 'a non-hermetic test, one that might pass/fail based on '
405 'what ran before it.',
407 choices
=['suite', 'test'])
408 parser
.add_argument('--list_tests', help='If set, list all tests that will be '
411 parser
.add_argument('--list_tests_attr', help='If set, list all tests and test '
415 def add_parse_opts(parser
: argparse
.ArgumentParser
) -> None:
416 parser
.add_argument('--raw_output', help='If set don\'t parse output from kernel. '
417 'By default, filters to just KUnit output. Use '
418 '--raw_output=all to show everything',
419 type=str, nargs
='?', const
='all', default
=None, choices
=['all', 'kunit'])
420 parser
.add_argument('--json',
422 help='Prints parsed test results as JSON to stdout or a file if '
423 'a filename is specified. Does nothing if --raw_output is set.',
424 type=str, const
='stdout', default
=None, metavar
='FILE')
425 parser
.add_argument('--summary',
426 help='Prints only the summary line for parsed test results.'
427 'Does nothing if --raw_output is set.',
429 parser
.add_argument('--failed',
430 help='Prints only the failed parsed test results and summary line.'
431 'Does nothing if --raw_output is set.',
435 def tree_from_args(cli_args
: argparse
.Namespace
) -> kunit_kernel
.LinuxSourceTree
:
436 """Returns a LinuxSourceTree based on the user's arguments."""
437 # Allow users to specify multiple arguments in one string, e.g. '-smp 8'
438 qemu_args
: List
[str] = []
439 if cli_args
.qemu_args
:
440 for arg
in cli_args
.qemu_args
:
441 qemu_args
.extend(shlex
.split(arg
))
443 kunitconfigs
= cli_args
.kunitconfig
if cli_args
.kunitconfig
else []
444 if cli_args
.alltests
:
445 # Prepend so user-specified options take prio if we ever allow
446 # --kunitconfig options to have differing options.
447 kunitconfigs
= [kunit_kernel
.ALL_TESTS_CONFIG_PATH
] + kunitconfigs
449 return kunit_kernel
.LinuxSourceTree(cli_args
.build_dir
,
450 kunitconfig_paths
=kunitconfigs
,
451 kconfig_add
=cli_args
.kconfig_add
,
453 cross_compile
=cli_args
.cross_compile
,
454 qemu_config_path
=cli_args
.qemu_config
,
455 extra_qemu_args
=qemu_args
)
458 def run_handler(cli_args
: argparse
.Namespace
) -> None:
459 if not os
.path
.exists(cli_args
.build_dir
):
460 os
.mkdir(cli_args
.build_dir
)
462 linux
= tree_from_args(cli_args
)
463 request
= KunitRequest(build_dir
=cli_args
.build_dir
,
464 make_options
=cli_args
.make_options
,
466 raw_output
=cli_args
.raw_output
,
468 summary
=cli_args
.summary
,
469 failed
=cli_args
.failed
,
470 timeout
=cli_args
.timeout
,
471 filter_glob
=cli_args
.filter_glob
,
472 filter=cli_args
.filter,
473 filter_action
=cli_args
.filter_action
,
474 kernel_args
=cli_args
.kernel_args
,
475 run_isolated
=cli_args
.run_isolated
,
476 list_tests
=cli_args
.list_tests
,
477 list_tests_attr
=cli_args
.list_tests_attr
)
478 result
= run_tests(linux
, request
)
479 if result
.status
!= KunitStatus
.SUCCESS
:
483 def config_handler(cli_args
: argparse
.Namespace
) -> None:
484 if cli_args
.build_dir
and (
485 not os
.path
.exists(cli_args
.build_dir
)):
486 os
.mkdir(cli_args
.build_dir
)
488 linux
= tree_from_args(cli_args
)
489 request
= KunitConfigRequest(build_dir
=cli_args
.build_dir
,
490 make_options
=cli_args
.make_options
)
491 result
= config_tests(linux
, request
)
492 stdout
.print_with_timestamp((
493 'Elapsed time: %.3fs\n') % (
494 result
.elapsed_time
))
495 if result
.status
!= KunitStatus
.SUCCESS
:
499 def build_handler(cli_args
: argparse
.Namespace
) -> None:
500 linux
= tree_from_args(cli_args
)
501 request
= KunitBuildRequest(build_dir
=cli_args
.build_dir
,
502 make_options
=cli_args
.make_options
,
504 result
= config_and_build_tests(linux
, request
)
505 stdout
.print_with_timestamp((
506 'Elapsed time: %.3fs\n') % (
507 result
.elapsed_time
))
508 if result
.status
!= KunitStatus
.SUCCESS
:
512 def exec_handler(cli_args
: argparse
.Namespace
) -> None:
513 linux
= tree_from_args(cli_args
)
514 exec_request
= KunitExecRequest(raw_output
=cli_args
.raw_output
,
515 build_dir
=cli_args
.build_dir
,
517 summary
=cli_args
.summary
,
518 failed
=cli_args
.failed
,
519 timeout
=cli_args
.timeout
,
520 filter_glob
=cli_args
.filter_glob
,
521 filter=cli_args
.filter,
522 filter_action
=cli_args
.filter_action
,
523 kernel_args
=cli_args
.kernel_args
,
524 run_isolated
=cli_args
.run_isolated
,
525 list_tests
=cli_args
.list_tests
,
526 list_tests_attr
=cli_args
.list_tests_attr
)
527 result
= exec_tests(linux
, exec_request
)
528 stdout
.print_with_timestamp((
529 'Elapsed time: %.3fs\n') % (result
.elapsed_time
))
530 if result
.status
!= KunitStatus
.SUCCESS
:
534 def parse_handler(cli_args
: argparse
.Namespace
) -> None:
535 if cli_args
.file is None:
536 sys
.stdin
.reconfigure(errors
='backslashreplace') # type: ignore
537 kunit_output
= sys
.stdin
# type: Iterable[str]
539 with
open(cli_args
.file, 'r', errors
='backslashreplace') as f
:
540 kunit_output
= f
.read().splitlines()
541 # We know nothing about how the result was created!
542 metadata
= kunit_json
.Metadata()
543 request
= KunitParseRequest(raw_output
=cli_args
.raw_output
,
544 json
=cli_args
.json
, summary
=cli_args
.summary
,
545 failed
=cli_args
.failed
)
546 result
, _
= parse_tests(request
, metadata
, kunit_output
)
547 if result
.status
!= KunitStatus
.SUCCESS
:
551 subcommand_handlers_map
= {
553 'config': config_handler
,
554 'build': build_handler
,
555 'exec': exec_handler
,
556 'parse': parse_handler
560 def main(argv
: Sequence
[str]) -> None:
561 parser
= argparse
.ArgumentParser(
562 description
='Helps writing and running KUnit tests.')
563 subparser
= parser
.add_subparsers(dest
='subcommand')
565 # The 'run' command will config, build, exec, and parse in one go.
566 run_parser
= subparser
.add_parser('run', help='Runs KUnit tests.')
567 add_common_opts(run_parser
)
568 add_build_opts(run_parser
)
569 add_exec_opts(run_parser
)
570 add_parse_opts(run_parser
)
572 config_parser
= subparser
.add_parser('config',
573 help='Ensures that .config contains all of '
574 'the options in .kunitconfig')
575 add_common_opts(config_parser
)
577 build_parser
= subparser
.add_parser('build', help='Builds a kernel with KUnit tests')
578 add_common_opts(build_parser
)
579 add_build_opts(build_parser
)
581 exec_parser
= subparser
.add_parser('exec', help='Run a kernel with KUnit tests')
582 add_common_opts(exec_parser
)
583 add_exec_opts(exec_parser
)
584 add_parse_opts(exec_parser
)
586 # The 'parse' option is special, as it doesn't need the kernel source
587 # (therefore there is no need for a build_dir, hence no add_common_opts)
588 # and the '--file' argument is not relevant to 'run', so isn't in
590 parse_parser
= subparser
.add_parser('parse',
591 help='Parses KUnit results from a file, '
592 'and parses formatted results.')
593 add_parse_opts(parse_parser
)
594 parse_parser
.add_argument('file',
595 help='Specifies the file to read results from.',
596 type=str, nargs
='?', metavar
='input_file')
598 cli_args
= parser
.parse_args(massage_argv(argv
))
600 if get_kernel_root_path():
601 os
.chdir(get_kernel_root_path())
603 subcomand_handler
= subcommand_handlers_map
.get(cli_args
.subcommand
, None)
605 if subcomand_handler
is None:
609 subcomand_handler(cli_args
)
612 if __name__
== '__main__':