mirrors/linux
2
0
Fork 0
mirror of https://git.kernel.org/pub/scm/linux/kernel/git/torvalds/linux.git synced 2026-05-16 03:11:11 -04:00
Code Issues Packages Projects Releases Wiki Activity

Merge tag 'linux_kselftest-kunit-7.1-rc1' of git://git.kernel.org/pub/scm/linux/kernel/git/shuah/linux-kselftest

Browse Source 
Pull kunit tool updates from Shuah Khan:

 - terminate kernel under test on SIGINT when it catches SIGINT to make
   sure the TTY isn't messed up and terminate the running kernel

 - recommend --raw_output=all when KTAP header isn't found in the kernel
   output, it's useful to re-run the test with --raw_output=all to find
   out the reasons why the test didn't complete.

 - skip stty when stdin is not a tty to avoid writing noise to stderr.

 - show suites when user runs --list_suites option instead of entire
   list of tests to make the output user friendly and concise.

* tag 'linux_kselftest-kunit-7.1-rc1' of git://git.kernel.org/pub/scm/linux/kernel/git/shuah/linux-kselftest:
  kunit: tool: Terminate kernel under test on SIGINT
  kunit: tool: skip stty when stdin is not a tty
  kunit: tool: Recommend --raw_output=all if no KTAP found
  kunit: Add --list_suites to show suites
This commit is contained in:
Linus Torvalds
2026-04-14 17:39:42 -07:00
parent 88b29f3f57 8f260b02ee
commit 6198c86a97
4 changed files with 98 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
tools/testing/kunit/kunit.py
View File

@@ -63,6 +63,7 @@ class KunitExecRequest(KunitParseRequest):
run_isolated: Optional[str]
list_tests: bool
list_tests_attr: bool
list_suites: bool
@dataclass
class KunitRequest(KunitExecRequest, KunitBuildRequest):
@@ -168,6 +169,12 @@ def exec_tests(linux: kunit_kernel.LinuxSourceTree, request: KunitExecRequest) -
for line in attr_output:
print(line.rstrip())
return KunitResult(status=KunitStatus.SUCCESS, elapsed_time=0.0)
if request.list_suites:
tests = _list_tests(linux, request)
output = _suites_from_test_list(tests)
for line in output:
print(line.rstrip())
return KunitResult(status=KunitStatus.SUCCESS, elapsed_time=0.0)
if request.run_isolated:
tests = _list_tests(linux, request)
if request.run_isolated == 'test':
@@ -438,6 +445,9 @@ def add_exec_opts(parser: argparse.ArgumentParser) -> None:
parser.add_argument('--list_tests_attr', help='If set, list all tests and test '
'attributes.',
action='store_true')
parser.add_argument('--list_suites', help='If set, list all suites that will be '
'run.',
action='store_true')
def add_parse_opts(parser: argparse.ArgumentParser) -> None:
parser.add_argument('--raw_output', help='If set don\'t parse output from kernel. '
@@ -501,7 +511,8 @@ def run_handler(cli_args: argparse.Namespace) -> None:
kernel_args=cli_args.kernel_args,
run_isolated=cli_args.run_isolated,
list_tests=cli_args.list_tests,
list_tests_attr=cli_args.list_tests_attr)
list_tests_attr=cli_args.list_tests_attr,
list_suites=cli_args.list_suites)
result = run_tests(linux, request)
if result.status != KunitStatus.SUCCESS:
sys.exit(1)
@@ -550,7 +561,8 @@ def exec_handler(cli_args: argparse.Namespace) -> None:
kernel_args=cli_args.kernel_args,
run_isolated=cli_args.run_isolated,
list_tests=cli_args.list_tests,
list_tests_attr=cli_args.list_tests_attr)
list_tests_attr=cli_args.list_tests_attr,
list_suites=cli_args.list_suites)
result = exec_tests(linux, exec_request)
stdout.print_with_timestamp((
'Elapsed time: %.3fs\n') % (result.elapsed_time))

38
tools/testing/kunit/kunit_kernel.py
View File

@@ -16,7 +16,7 @@ import shutil
import signal
import sys
import threading
from typing import Iterator, List, Optional, Tuple
from typing import Iterator, List, Optional, Tuple, Any
from types import FrameType
import kunit_config
@@ -265,6 +265,7 @@ class LinuxSourceTree:
if kconfig_add:
kconfig = kunit_config.parse_from_string('\n'.join(kconfig_add))
self._kconfig.merge_in_entries(kconfig)
self._process : Optional[subprocess.Popen[Any]] = None
def arch(self) -> str:
return self._arch
@@ -345,6 +346,12 @@ class LinuxSourceTree:
return False
return self.validate_config(build_dir)
def _restore_terminal_if_tty(self) -> None:
# stty requires a controlling terminal; skip headless runs.
if sys.stdin is None or not sys.stdin.isatty():
return
subprocess.call(['stty', 'sane'])
def run_kernel(self, args: Optional[List[str]]=None, build_dir: str='', filter_glob: str='', filter: str='', filter_action: Optional[str]=None, timeout: Optional[int]=None) -> Iterator[str]:
# Copy to avoid mutating the caller-supplied list. exec_tests() reuses
# the same args across repeated run_kernel() calls (e.g. --run_isolated),
@@ -358,36 +365,45 @@ class LinuxSourceTree:
args.append('kunit.filter_action=' + filter_action)
args.append('kunit.enable=1')
process = self._ops.start(args, build_dir)
assert process.stdout is not None # tell mypy it's set
self._process = self._ops.start(args, build_dir)
assert self._process is not None # tell mypy it's set
assert self._process.stdout is not None # tell mypy it's set
# Enforce the timeout in a background thread.
def _wait_proc() -> None:
try:
process.wait(timeout=timeout)
if self._process:
self._process.wait(timeout=timeout)
except Exception as e:
print(e)
process.terminate()
process.wait()
if self._process:
self._process.terminate()
self._process.wait()
waiter = threading.Thread(target=_wait_proc)
waiter.start()
output = open(get_outfile_path(build_dir), 'w')
try:
# Tee the output to the file and to our caller in real time.
for line in process.stdout:
for line in self._process.stdout:
output.write(line)
yield line
# This runs even if our caller doesn't consume every line.
finally:
# Flush any leftover output to the file
output.write(process.stdout.read())
if self._process:
if self._process.stdout:
output.write(self._process.stdout.read())
self._process.stdout.close()
self._process = None
output.close()
process.stdout.close()
waiter.join()
subprocess.call(['stty', 'sane'])
self._restore_terminal_if_tty()
def signal_handler(self, unused_sig: int, unused_frame: Optional[FrameType]) -> None:
logging.error('Build interruption occurred. Cleaning console.')
subprocess.call(['stty', 'sane'])
if self._process:
self._process.terminate()
self._process.wait()
self._restore_terminal_if_tty()

3
tools/testing/kunit/kunit_parser.py
View File

@@ -857,7 +857,8 @@ def parse_run_tests(kernel_output: Iterable[str], printer: Printer) -> Test:
test = Test()
if not lines:
test.name = '<missing>'
test.add_error(printer, 'Could not find any KTAP output. Did any KUnit tests run?')
test.add_error(printer, 'Could not find any KTAP output. Did any KUnit tests run?\n' +
'Try running with the --raw_output=all option to see any log messages.')
test.status = TestStatus.FAILURE_TO_PARSE_TESTS
else:
test = parse_test(lines, 0, [], False, printer)

58
tools/testing/kunit/kunit_tool_test.py
View File

@@ -529,6 +529,48 @@ class LinuxSourceTreeTest(unittest.TestCase):
self.assertIn('kunit.filter_glob=suite.test1', start_calls[0])
self.assertIn('kunit.filter_glob=suite.test2', start_calls[1])
def test_run_kernel_skips_terminal_reset_without_tty(self):
def fake_start(unused_args, unused_build_dir):
return subprocess.Popen(['printf', 'KTAP version 1\n'],
text=True, stdout=subprocess.PIPE)
non_tty_stdin = mock.Mock()
non_tty_stdin.isatty.return_value = False
with tempfile.TemporaryDirectory('') as build_dir:
tree = kunit_kernel.LinuxSourceTree(build_dir, kunitconfig_paths=[os.devnull])
with mock.patch.object(tree._ops, 'start', side_effect=fake_start), \
mock.patch.object(kunit_kernel.sys, 'stdin', non_tty_stdin), \
mock.patch.object(kunit_kernel.subprocess, 'call') as mock_call:
for _ in tree.run_kernel(build_dir=build_dir):
pass
mock_call.assert_not_called()
def test_signal_handler_skips_terminal_reset_without_tty(self):
non_tty_stdin = mock.Mock()
non_tty_stdin.isatty.return_value = False
tree = kunit_kernel.LinuxSourceTree('', kunitconfig_paths=[os.devnull])
with mock.patch.object(kunit_kernel.sys, 'stdin', non_tty_stdin), \
mock.patch.object(kunit_kernel.subprocess, 'call') as mock_call, \
mock.patch.object(kunit_kernel.logging, 'error') as mock_error:
tree.signal_handler(signal.SIGINT, None)
mock_error.assert_called_once()
mock_call.assert_not_called()
def test_signal_handler_resets_terminal_with_tty(self):
tty_stdin = mock.Mock()
tty_stdin.isatty.return_value = True
tree = kunit_kernel.LinuxSourceTree('', kunitconfig_paths=[os.devnull])
with mock.patch.object(kunit_kernel.sys, 'stdin', tty_stdin), \
mock.patch.object(kunit_kernel.subprocess, 'call') as mock_call, \
mock.patch.object(kunit_kernel.logging, 'error') as mock_error:
tree.signal_handler(signal.SIGINT, None)
mock_error.assert_called_once()
mock_call.assert_called_once_with(['stty', 'sane'])
def test_build_reconfig_no_config(self):
with tempfile.TemporaryDirectory('') as build_dir:
with open(kunit_kernel.get_kunitconfig_path(build_dir), 'w') as f:
@@ -881,7 +923,7 @@ class KUnitMainTest(unittest.TestCase):
self.linux_source_mock.run_kernel.return_value = ['TAP version 14', 'init: random output'] + want
got = kunit._list_tests(self.linux_source_mock,
kunit.KunitExecRequest(None, None, False, False, '.kunit', 300, 'suite*', '', None, None, 'suite', False, False))
kunit.KunitExecRequest(None, None, False, False, '.kunit', 300, 'suite*', '', None, None, 'suite', False, False, False))
self.assertEqual(got, want)
# Should respect the user's filter glob when listing tests.
self.linux_source_mock.run_kernel.assert_called_once_with(
@@ -894,7 +936,7 @@ class KUnitMainTest(unittest.TestCase):
# Should respect the user's filter glob when listing tests.
mock_tests.assert_called_once_with(mock.ANY,
kunit.KunitExecRequest(None, None, False, False, '.kunit', 300, 'suite*.test*', '', None, None, 'suite', False, False))
kunit.KunitExecRequest(None, None, False, False, '.kunit', 300, 'suite*.test*', '', None, None, 'suite', False, False, False))
self.linux_source_mock.run_kernel.assert_has_calls([
mock.call(args=None, build_dir='.kunit', filter_glob='suite.test*', filter='', filter_action=None, timeout=300),
mock.call(args=None, build_dir='.kunit', filter_glob='suite2.test*', filter='', filter_action=None, timeout=300),
@@ -907,13 +949,23 @@ class KUnitMainTest(unittest.TestCase):
# Should respect the user's filter glob when listing tests.
mock_tests.assert_called_once_with(mock.ANY,
kunit.KunitExecRequest(None, None, False, False, '.kunit', 300, 'suite*', '', None, None, 'test', False, False))
kunit.KunitExecRequest(None, None, False, False, '.kunit', 300, 'suite*', '', None, None, 'test', False, False, False))
self.linux_source_mock.run_kernel.assert_has_calls([
mock.call(args=None, build_dir='.kunit', filter_glob='suite.test1', filter='', filter_action=None, timeout=300),
mock.call(args=None, build_dir='.kunit', filter_glob='suite.test2', filter='', filter_action=None, timeout=300),
mock.call(args=None, build_dir='.kunit', filter_glob='suite2.test1', filter='', filter_action=None, timeout=300),
])
@mock.patch.object(kunit, '_list_tests')
@mock.patch.object(sys, 'stdout', new_callable=io.StringIO)
def test_list_suites(self, mock_stdout, mock_tests):
mock_tests.return_value = ['suite.test1', 'suite.test2', 'suite2.test1']
kunit.main(['run', '--list_suites'])
want = ['suite', 'suite2']
output = mock_stdout.getvalue().split()
self.assertEqual(output, want)
@mock.patch.object(sys, 'stdout', new_callable=io.StringIO)
def test_list_cmds(self, mock_stdout):
kunit.main(['--list-cmds'])