kunit: tool: fix pre-existing `mypy --strict` errors and update run_checks.py (1da2e622) · Commits · git / linux-net

tools/testing/kunit/kunit.py

+12 −12

Original line number	Diff line number	Diff line
		@@ -269,7 +269,7 @@ def massage_argv(argv: Sequence[str]) -> Sequence[str]:
		def get_default_jobs() -> int:
		return len(os.sched_getaffinity(0))

		def add_common_opts(parser) -> None:
		def add_common_opts(parser: argparse.ArgumentParser) -> None:
		parser.add_argument('--build_dir',
		help='As in the make command, it specifies the build '
		'directory.',
		@@ -320,13 +320,13 @@ def add_common_opts(parser) -> None:
		help='Additional QEMU arguments, e.g. "-smp 8"',
		action='append', metavar='')

		def add_build_opts(parser) -> None:
		def add_build_opts(parser: argparse.ArgumentParser) -> None:
		parser.add_argument('--jobs',
		help='As in the make command, "Specifies the number of '
		'jobs (commands) to run simultaneously."',
		type=int, default=get_default_jobs(), metavar='N')

		def add_exec_opts(parser) -> None:
		def add_exec_opts(parser: argparse.ArgumentParser) -> None:
		parser.add_argument('--timeout',
		help='maximum number of seconds to allow for all tests '
		'to run. This does not include time taken to build the '
		@@ -351,7 +351,7 @@ def add_exec_opts(parser) -> None:
		type=str,
		choices=['suite', 'test'])

		def add_parse_opts(parser) -> None:
		def add_parse_opts(parser: argparse.ArgumentParser) -> None:
		parser.add_argument('--raw_output', help='If set don\'t parse output from kernel. '
		'By default, filters to just KUnit output. Use '
		'--raw_output=all to show everything',
		@@ -386,7 +386,7 @@ def tree_from_args(cli_args: argparse.Namespace) -> kunit_kernel.LinuxSourceTree
		extra_qemu_args=qemu_args)


		def run_handler(cli_args):
		def run_handler(cli_args: argparse.Namespace) -> None:
		if not os.path.exists(cli_args.build_dir):
		os.mkdir(cli_args.build_dir)

		@@ -405,7 +405,7 @@ def run_handler(cli_args):
		sys.exit(1)


		def config_handler(cli_args):
		def config_handler(cli_args: argparse.Namespace) -> None:
		if cli_args.build_dir and (
		not os.path.exists(cli_args.build_dir)):
		os.mkdir(cli_args.build_dir)
		@@ -421,7 +421,7 @@ def config_handler(cli_args):
		sys.exit(1)


		def build_handler(cli_args):
		def build_handler(cli_args: argparse.Namespace) -> None:
		linux = tree_from_args(cli_args)
		request = KunitBuildRequest(build_dir=cli_args.build_dir,
		make_options=cli_args.make_options,
		@@ -434,7 +434,7 @@ def build_handler(cli_args):
		sys.exit(1)


		def exec_handler(cli_args):
		def exec_handler(cli_args: argparse.Namespace) -> None:
		linux = tree_from_args(cli_args)
		exec_request = KunitExecRequest(raw_output=cli_args.raw_output,
		build_dir=cli_args.build_dir,
		@@ -450,10 +450,10 @@ def exec_handler(cli_args):
		sys.exit(1)


		def parse_handler(cli_args):
		def parse_handler(cli_args: argparse.Namespace) -> None:
		if cli_args.file is None:
		sys.stdin.reconfigure(errors='backslashreplace') # pytype: disable=attribute-error
		kunit_output = sys.stdin
		sys.stdin.reconfigure(errors='backslashreplace') # type: ignore
		kunit_output = sys.stdin # type: Iterable[str]
		else:
		with open(cli_args.file, 'r', errors='backslashreplace') as f:
		kunit_output = f.read().splitlines()
		@@ -475,7 +475,7 @@ subcommand_handlers_map = {
		}


		def main(argv):
		def main(argv: Sequence[str]) -> None:
		parser = argparse.ArgumentParser(
		description='Helps writing and running KUnit tests.')
		subparser = parser.add_subparsers(dest='subcommand')

tools/testing/kunit/kunit_config.py

+2 −2

Original line number	Diff line number	Diff line
		@@ -8,7 +8,7 @@

		from dataclasses import dataclass
		import re
		from typing import Dict, Iterable, List, Tuple
		from typing import Any, Dict, Iterable, List, Tuple

		CONFIG_IS_NOT_SET_PATTERN = r'^# CONFIG_(\w+) is not set$'
		CONFIG_PATTERN = r'^CONFIG_(\w+)=(\S+\|".*")$'
		@@ -34,7 +34,7 @@ class Kconfig:
		def __init__(self) -> None:
		self._entries = {} # type: Dict[str, str]

		def __eq__(self, other) -> bool:
		def __eq__(self, other: Any) -> bool:
		if not isinstance(other, self.__class__):
		return False
		return self._entries == other._entries

tools/testing/kunit/kunit_kernel.py

+15 −14

Original line number	Diff line number	Diff line
		@@ -16,6 +16,7 @@ import shutil
		import signal
		import threading
		from typing import Iterator, List, Optional, Tuple
		from types import FrameType

		import kunit_config
		import qemu_config
		@@ -56,7 +57,7 @@ class LinuxSourceTreeOperations:
		def make_arch_config(self, base_kunitconfig: kunit_config.Kconfig) -> kunit_config.Kconfig:
		return base_kunitconfig

		def make_olddefconfig(self, build_dir: str, make_options) -> None:
		def make_olddefconfig(self, build_dir: str, make_options: Optional[List[str]]) -> None:
		command = ['make', 'ARCH=' + self._linux_arch, 'O=' + build_dir, 'olddefconfig']
		if self._cross_compile:
		command += ['CROSS_COMPILE=' + self._cross_compile]
		@@ -70,7 +71,7 @@ class LinuxSourceTreeOperations:
		except subprocess.CalledProcessError as e:
		raise ConfigError(e.output.decode())

		def make(self, jobs, build_dir: str, make_options) -> None:
		def make(self, jobs: int, build_dir: str, make_options: Optional[List[str]]) -> None:
		command = ['make', 'ARCH=' + self._linux_arch, 'O=' + build_dir, '--jobs=' + str(jobs)]
		if make_options:
		command.extend(make_options)
		@@ -132,7 +133,7 @@ class LinuxSourceTreeOperationsQemu(LinuxSourceTreeOperations):
		class LinuxSourceTreeOperationsUml(LinuxSourceTreeOperations):
		"""An abstraction over command line operations performed on a source tree."""

		def __init__(self, cross_compile=None):
		def __init__(self, cross_compile: Optional[str]=None):
		super().__init__(linux_arch='um', cross_compile=cross_compile)

		def make_arch_config(self, base_kunitconfig: kunit_config.Kconfig) -> kunit_config.Kconfig:
		@@ -215,7 +216,7 @@ def _get_qemu_ops(config_path: str,

		if not hasattr(config, 'QEMU_ARCH'):
		raise ValueError('qemu_config module missing "QEMU_ARCH": ' + config_path)
		params: qemu_config.QemuArchParams = config.QEMU_ARCH # type: ignore
		params: qemu_config.QemuArchParams = config.QEMU_ARCH
		if extra_qemu_args:
		params.extra_qemu_params.extend(extra_qemu_args)
		return params.linux_arch, LinuxSourceTreeOperationsQemu(
		@@ -229,10 +230,10 @@ class LinuxSourceTree:
		build_dir: str,
		kunitconfig_paths: Optional[List[str]]=None,
		kconfig_add: Optional[List[str]]=None,
		arch=None,
		cross_compile=None,
		qemu_config_path=None,
		extra_qemu_args=None) -> None:
		arch: Optional[str]=None,
		cross_compile: Optional[str]=None,
		qemu_config_path: Optional[str]=None,
		extra_qemu_args: Optional[List[str]]=None) -> None:
		signal.signal(signal.SIGINT, self.signal_handler)
		if qemu_config_path:
		self._arch, self._ops = _get_qemu_ops(qemu_config_path, extra_qemu_args, cross_compile)
		@@ -275,7 +276,7 @@ class LinuxSourceTree:
		logging.error(message)
		return False

		def build_config(self, build_dir: str, make_options) -> bool:
		def build_config(self, build_dir: str, make_options: Optional[List[str]]) -> bool:
		kconfig_path = get_kconfig_path(build_dir)
		if build_dir and not os.path.exists(build_dir):
		os.mkdir(build_dir)
		@@ -303,7 +304,7 @@ class LinuxSourceTree:
		old_kconfig = kunit_config.parse_file(old_path)
		return old_kconfig != self._kconfig

		def build_reconfig(self, build_dir: str, make_options) -> bool:
		def build_reconfig(self, build_dir: str, make_options: Optional[List[str]]) -> bool:
		"""Creates a new .config if it is not a subset of the .kunitconfig."""
		kconfig_path = get_kconfig_path(build_dir)
		if not os.path.exists(kconfig_path):
		@@ -319,7 +320,7 @@ class LinuxSourceTree:
		os.remove(kconfig_path)
		return self.build_config(build_dir, make_options)

		def build_kernel(self, jobs, build_dir: str, make_options) -> bool:
		def build_kernel(self, jobs: int, build_dir: str, make_options: Optional[List[str]]) -> bool:
		try:
		self._ops.make_olddefconfig(build_dir, make_options)
		self._ops.make(jobs, build_dir, make_options)
		@@ -328,7 +329,7 @@ class LinuxSourceTree:
		return False
		return self.validate_config(build_dir)

		def run_kernel(self, args=None, build_dir='', filter_glob='', timeout=None) -> Iterator[str]:
		def run_kernel(self, args: Optional[List[str]]=None, build_dir: str='', filter_glob: str='', timeout: Optional[int]=None) -> Iterator[str]:
		if not args:
		args = []
		if filter_glob:
		@@ -339,7 +340,7 @@ class LinuxSourceTree:
		assert process.stdout is not None # tell mypy it's set

		# Enforce the timeout in a background thread.
		def _wait_proc():
		def _wait_proc() -> None:
		try:
		process.wait(timeout=timeout)
		except Exception as e:
		@@ -365,6 +366,6 @@ class LinuxSourceTree:
		waiter.join()
		subprocess.call(['stty', 'sane'])

		def signal_handler(self, unused_sig, unused_frame) -> None:
		def signal_handler(self, unused_sig: int, unused_frame: Optional[FrameType]) -> None:
		logging.error('Build interruption occurred. Cleaning console.')
		subprocess.call(['stty', 'sane'])

tools/testing/kunit/run_checks.py

+2 −2

Original line number	Diff line number	Diff line
		@@ -23,7 +23,7 @@ commands: Dict[str, Sequence[str]] = {
		'kunit_tool_test.py': ['./kunit_tool_test.py'],
		'kunit smoke test': ['./kunit.py', 'run', '--kunitconfig=lib/kunit', '--build_dir=kunit_run_checks'],
		'pytype': ['/bin/sh', '-c', 'pytype *.py'],
		'mypy': ['/bin/sh', '-c', 'mypy *.py'],
		'mypy': ['mypy', '--strict', '--exclude', '_test.py$', '--exclude', 'qemu_configs/', '.'],
		}

		# The user might not have mypy or pytype installed, skip them if so.
		@@ -73,7 +73,7 @@ def main(argv: Sequence[str]) -> None:
		sys.exit(1)


		def run_cmd(argv: Sequence[str]):
		def run_cmd(argv: Sequence[str]) -> None:
		subprocess.check_output(argv, stderr=subprocess.STDOUT, cwd=ABS_TOOL_PATH, timeout=TIMEOUT)