kunit.py (50501936288d6a29d7ef78f25d00e33240fad45f) kunit.py (723c8258c8fe167191b53e274dea435c4522e4d7)
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>

--- 41 unchanged lines hidden (view full) ---

50 raw_output: Optional[str]
51 json: Optional[str]
52
53@dataclass
54class KunitExecRequest(KunitParseRequest):
55 build_dir: str
56 timeout: int
57 filter_glob: str
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>

--- 41 unchanged lines hidden (view full) ---

50 raw_output: Optional[str]
51 json: Optional[str]
52
53@dataclass
54class KunitExecRequest(KunitParseRequest):
55 build_dir: str
56 timeout: int
57 filter_glob: str
58 filter: str
59 filter_action: Optional[str]
58 kernel_args: Optional[List[str]]
59 run_isolated: Optional[str]
60 kernel_args: Optional[List[str]]
61 run_isolated: Optional[str]
62 list_tests: bool
63 list_tests_attr: bool
60
61@dataclass
62class KunitRequest(KunitExecRequest, KunitBuildRequest):
63 pass
64
65
66def get_kernel_root_path() -> str:
67 path = sys.argv[0] if not __file__ else __file__

--- 29 unchanged lines hidden (view full) ---

97 config_result = config_tests(linux, request)
98 if config_result.status != KunitStatus.SUCCESS:
99 return config_result
100
101 return build_tests(linux, request)
102
103def _list_tests(linux: kunit_kernel.LinuxSourceTree, request: KunitExecRequest) -> List[str]:
104 args = ['kunit.action=list']
64
65@dataclass
66class KunitRequest(KunitExecRequest, KunitBuildRequest):
67 pass
68
69
70def get_kernel_root_path() -> str:
71 path = sys.argv[0] if not __file__ else __file__

--- 29 unchanged lines hidden (view full) ---

101 config_result = config_tests(linux, request)
102 if config_result.status != KunitStatus.SUCCESS:
103 return config_result
104
105 return build_tests(linux, request)
106
107def _list_tests(linux: kunit_kernel.LinuxSourceTree, request: KunitExecRequest) -> List[str]:
108 args = ['kunit.action=list']
109
105 if request.kernel_args:
106 args.extend(request.kernel_args)
107
108 output = linux.run_kernel(args=args,
109 timeout=request.timeout,
110 filter_glob=request.filter_glob,
110 if request.kernel_args:
111 args.extend(request.kernel_args)
112
113 output = linux.run_kernel(args=args,
114 timeout=request.timeout,
115 filter_glob=request.filter_glob,
116 filter=request.filter,
117 filter_action=request.filter_action,
111 build_dir=request.build_dir)
112 lines = kunit_parser.extract_tap_lines(output)
113 # Hack! Drop the dummy TAP version header that the executor prints out.
114 lines.pop()
115
116 # Filter out any extraneous non-test output that might have gotten mixed in.
118 build_dir=request.build_dir)
119 lines = kunit_parser.extract_tap_lines(output)
120 # Hack! Drop the dummy TAP version header that the executor prints out.
121 lines.pop()
122
123 # Filter out any extraneous non-test output that might have gotten mixed in.
117 return [l for l in lines if re.match(r'^[^\s.]+\.[^\s.]+$', l)]
124 return [l for l in output if re.match(r'^[^\s.]+\.[^\s.]+$', l)]
118
125
126def _list_tests_attr(linux: kunit_kernel.LinuxSourceTree, request: KunitExecRequest) -> Iterable[str]:
127 args = ['kunit.action=list_attr']
128
129 if request.kernel_args:
130 args.extend(request.kernel_args)
131
132 output = linux.run_kernel(args=args,
133 timeout=request.timeout,
134 filter_glob=request.filter_glob,
135 filter=request.filter,
136 filter_action=request.filter_action,
137 build_dir=request.build_dir)
138 lines = kunit_parser.extract_tap_lines(output)
139 # Hack! Drop the dummy TAP version header that the executor prints out.
140 lines.pop()
141
142 # Filter out any extraneous non-test output that might have gotten mixed in.
143 return lines
144
119def _suites_from_test_list(tests: List[str]) -> List[str]:
120 """Extracts all the suites from an ordered list of tests."""
121 suites = [] # type: List[str]
122 for t in tests:
123 parts = t.split('.', maxsplit=2)
124 if len(parts) != 2:
125 raise ValueError(f'internal KUnit error, test name should be of the form "<suite>.<test>", got "{t}"')
126 suite, _ = parts
127 if not suites or suites[-1] != suite:
128 suites.append(suite)
129 return suites
130
145def _suites_from_test_list(tests: List[str]) -> List[str]:
146 """Extracts all the suites from an ordered list of tests."""
147 suites = [] # type: List[str]
148 for t in tests:
149 parts = t.split('.', maxsplit=2)
150 if len(parts) != 2:
151 raise ValueError(f'internal KUnit error, test name should be of the form "<suite>.<test>", got "{t}"')
152 suite, _ = parts
153 if not suites or suites[-1] != suite:
154 suites.append(suite)
155 return suites
156
131
132
133def exec_tests(linux: kunit_kernel.LinuxSourceTree, request: KunitExecRequest) -> KunitResult:
134 filter_globs = [request.filter_glob]
157def exec_tests(linux: kunit_kernel.LinuxSourceTree, request: KunitExecRequest) -> KunitResult:
158 filter_globs = [request.filter_glob]
159 if request.list_tests:
160 output = _list_tests(linux, request)
161 for line in output:
162 print(line.rstrip())
163 return KunitResult(status=KunitStatus.SUCCESS, elapsed_time=0.0)
164 if request.list_tests_attr:
165 attr_output = _list_tests_attr(linux, request)
166 for line in attr_output:
167 print(line.rstrip())
168 return KunitResult(status=KunitStatus.SUCCESS, elapsed_time=0.0)
135 if request.run_isolated:
136 tests = _list_tests(linux, request)
137 if request.run_isolated == 'test':
138 filter_globs = tests
139 elif request.run_isolated == 'suite':
140 filter_globs = _suites_from_test_list(tests)
141 # Apply the test-part of the user's glob, if present.
142 if '.' in request.filter_glob:

--- 7 unchanged lines hidden (view full) ---

150 for i, filter_glob in enumerate(filter_globs):
151 stdout.print_with_timestamp('Starting KUnit Kernel ({}/{})...'.format(i+1, len(filter_globs)))
152
153 test_start = time.time()
154 run_result = linux.run_kernel(
155 args=request.kernel_args,
156 timeout=request.timeout,
157 filter_glob=filter_glob,
169 if request.run_isolated:
170 tests = _list_tests(linux, request)
171 if request.run_isolated == 'test':
172 filter_globs = tests
173 elif request.run_isolated == 'suite':
174 filter_globs = _suites_from_test_list(tests)
175 # Apply the test-part of the user's glob, if present.
176 if '.' in request.filter_glob:

--- 7 unchanged lines hidden (view full) ---

184 for i, filter_glob in enumerate(filter_globs):
185 stdout.print_with_timestamp('Starting KUnit Kernel ({}/{})...'.format(i+1, len(filter_globs)))
186
187 test_start = time.time()
188 run_result = linux.run_kernel(
189 args=request.kernel_args,
190 timeout=request.timeout,
191 filter_glob=filter_glob,
192 filter=request.filter,
193 filter_action=request.filter_action,
158 build_dir=request.build_dir)
159
160 _, test_result = parse_tests(request, metadata, run_result)
161 # run_kernel() doesn't block on the kernel exiting.
162 # That only happens after we get the last line of output from `run_result`.
163 # So exec_time here actually contains parsing + execution time, which is fine.
164 test_end = time.time()
165 exec_time += test_end - test_start

--- 170 unchanged lines hidden (view full) ---

336 metavar='SECONDS')
337 parser.add_argument('filter_glob',
338 help='Filter which KUnit test suites/tests run at '
339 'boot-time, e.g. list* or list*.*del_test',
340 type=str,
341 nargs='?',
342 default='',
343 metavar='filter_glob')
194 build_dir=request.build_dir)
195
196 _, test_result = parse_tests(request, metadata, run_result)
197 # run_kernel() doesn't block on the kernel exiting.
198 # That only happens after we get the last line of output from `run_result`.
199 # So exec_time here actually contains parsing + execution time, which is fine.
200 test_end = time.time()
201 exec_time += test_end - test_start

--- 170 unchanged lines hidden (view full) ---

372 metavar='SECONDS')
373 parser.add_argument('filter_glob',
374 help='Filter which KUnit test suites/tests run at '
375 'boot-time, e.g. list* or list*.*del_test',
376 type=str,
377 nargs='?',
378 default='',
379 metavar='filter_glob')
380 parser.add_argument('--filter',
381 help='Filter KUnit tests with attributes, '
382 'e.g. module=example or speed>slow',
383 type=str,
384 default='')
385 parser.add_argument('--filter_action',
386 help='If set to skip, filtered tests will be skipped, '
387 'e.g. --filter_action=skip. Otherwise they will not run.',
388 type=str,
389 choices=['skip'])
344 parser.add_argument('--kernel_args',
345 help='Kernel command-line parameters. Maybe be repeated',
346 action='append', metavar='')
347 parser.add_argument('--run_isolated', help='If set, boot the kernel for each '
348 'individual suite/test. This is can be useful for debugging '
349 'a non-hermetic test, one that might pass/fail based on '
350 'what ran before it.',
351 type=str,
352 choices=['suite', 'test'])
390 parser.add_argument('--kernel_args',
391 help='Kernel command-line parameters. Maybe be repeated',
392 action='append', metavar='')
393 parser.add_argument('--run_isolated', help='If set, boot the kernel for each '
394 'individual suite/test. This is can be useful for debugging '
395 'a non-hermetic test, one that might pass/fail based on '
396 'what ran before it.',
397 type=str,
398 choices=['suite', 'test'])
399 parser.add_argument('--list_tests', help='If set, list all tests that will be '
400 'run.',
401 action='store_true')
402 parser.add_argument('--list_tests_attr', help='If set, list all tests and test '
403 'attributes.',
404 action='store_true')
353
354def add_parse_opts(parser: argparse.ArgumentParser) -> None:
355 parser.add_argument('--raw_output', help='If set don\'t parse output from kernel. '
356 'By default, filters to just KUnit output. Use '
357 '--raw_output=all to show everything',
358 type=str, nargs='?', const='all', default=None, choices=['all', 'kunit'])
359 parser.add_argument('--json',
360 nargs='?',

--- 32 unchanged lines hidden (view full) ---

393 linux = tree_from_args(cli_args)
394 request = KunitRequest(build_dir=cli_args.build_dir,
395 make_options=cli_args.make_options,
396 jobs=cli_args.jobs,
397 raw_output=cli_args.raw_output,
398 json=cli_args.json,
399 timeout=cli_args.timeout,
400 filter_glob=cli_args.filter_glob,
405
406def add_parse_opts(parser: argparse.ArgumentParser) -> None:
407 parser.add_argument('--raw_output', help='If set don\'t parse output from kernel. '
408 'By default, filters to just KUnit output. Use '
409 '--raw_output=all to show everything',
410 type=str, nargs='?', const='all', default=None, choices=['all', 'kunit'])
411 parser.add_argument('--json',
412 nargs='?',

--- 32 unchanged lines hidden (view full) ---

445 linux = tree_from_args(cli_args)
446 request = KunitRequest(build_dir=cli_args.build_dir,
447 make_options=cli_args.make_options,
448 jobs=cli_args.jobs,
449 raw_output=cli_args.raw_output,
450 json=cli_args.json,
451 timeout=cli_args.timeout,
452 filter_glob=cli_args.filter_glob,
453 filter=cli_args.filter,
454 filter_action=cli_args.filter_action,
401 kernel_args=cli_args.kernel_args,
455 kernel_args=cli_args.kernel_args,
402 run_isolated=cli_args.run_isolated)
456 run_isolated=cli_args.run_isolated,
457 list_tests=cli_args.list_tests,
458 list_tests_attr=cli_args.list_tests_attr)
403 result = run_tests(linux, request)
404 if result.status != KunitStatus.SUCCESS:
405 sys.exit(1)
406
407
408def config_handler(cli_args: argparse.Namespace) -> None:
409 if cli_args.build_dir and (
410 not os.path.exists(cli_args.build_dir)):

--- 25 unchanged lines hidden (view full) ---

436
437def exec_handler(cli_args: argparse.Namespace) -> None:
438 linux = tree_from_args(cli_args)
439 exec_request = KunitExecRequest(raw_output=cli_args.raw_output,
440 build_dir=cli_args.build_dir,
441 json=cli_args.json,
442 timeout=cli_args.timeout,
443 filter_glob=cli_args.filter_glob,
459 result = run_tests(linux, request)
460 if result.status != KunitStatus.SUCCESS:
461 sys.exit(1)
462
463
464def config_handler(cli_args: argparse.Namespace) -> None:
465 if cli_args.build_dir and (
466 not os.path.exists(cli_args.build_dir)):

--- 25 unchanged lines hidden (view full) ---

492
493def exec_handler(cli_args: argparse.Namespace) -> None:
494 linux = tree_from_args(cli_args)
495 exec_request = KunitExecRequest(raw_output=cli_args.raw_output,
496 build_dir=cli_args.build_dir,
497 json=cli_args.json,
498 timeout=cli_args.timeout,
499 filter_glob=cli_args.filter_glob,
500 filter=cli_args.filter,
501 filter_action=cli_args.filter_action,
444 kernel_args=cli_args.kernel_args,
502 kernel_args=cli_args.kernel_args,
445 run_isolated=cli_args.run_isolated)
503 run_isolated=cli_args.run_isolated,
504 list_tests=cli_args.list_tests,
505 list_tests_attr=cli_args.list_tests_attr)
446 result = exec_tests(linux, exec_request)
447 stdout.print_with_timestamp((
448 'Elapsed time: %.3fs\n') % (result.elapsed_time))
449 if result.status != KunitStatus.SUCCESS:
450 sys.exit(1)
451
452
453def parse_handler(cli_args: argparse.Namespace) -> None:

--- 78 unchanged lines hidden --- 		506 result = exec_tests(linux, exec_request)
507 stdout.print_with_timestamp((
508 'Elapsed time: %.3fs\n') % (result.elapsed_time))
509 if result.status != KunitStatus.SUCCESS:
510 sys.exit(1)
511
512
513def parse_handler(cli_args: argparse.Namespace) -> None:

--- 78 unchanged lines hidden ---