Commit db167981 authored by Daniel Latypov's avatar Daniel Latypov Committed by Shuah Khan
Browse files

kunit: tool: use dataclass instead of collections.namedtuple 



namedtuple is a terse way of defining a collection of fields.
However, it does not allow us to annotate the type of these fields.
It also doesn't let us have any sort of inheritance between types.

Since commit df4b0807 ("kunit: tool: Assert the version
requirement"), kunit.py has asserted that it's running on python >=3.7.

So in that case use a 3.7 feature, dataclasses, to replace these.

Changes in detail:
* Make KunitExecRequest contain all the fields needed for exec_tests
* Use inheritance to dedupe fields
  * also allows us to e.g. pass a KUnitRequest in as a KUnitParseRequest
  * this has changed around the order of some fields
* Use named arguments when constructing all request objects in kunit.py
  * This is to prevent accidentally mixing up fields, etc.

Signed-off-by: default avatarDaniel Latypov <dlatypov@google.com>
Reviewed-by: default avatarBrendan Higgins <brendanhiggins@google.com>
Signed-off-by: default avatarShuah Khan <skhan@linuxfoundation.org>
parent 7fa7ffcf

tools/testing/kunit/kunit.py

+72 −67
Original line number Original line Diff line number Diff line
@@ -15,38 +15,57 @@ import time 




assert sys.version_info >= (3, 7), "Python version is too old"

assert sys.version_info >= (3, 7), "Python version is too old"





from collections import namedtuple

from dataclasses import dataclass

from enum import Enum, auto

from enum import Enum, auto

from typing import Iterable, Sequence, List

from typing import Any, Iterable, Sequence, List, Optional





import kunit_json

import kunit_json

import kunit_kernel

import kunit_kernel

import kunit_parser

import kunit_parser





KunitResult = namedtuple('KunitResult', ['status','result','elapsed_time'])



KunitConfigRequest = namedtuple('KunitConfigRequest',

				['build_dir', 'make_options'])

KunitBuildRequest = namedtuple('KunitBuildRequest',

			       ['jobs', 'build_dir', 'alltests',

				'make_options'])

KunitExecRequest = namedtuple('KunitExecRequest',

			      ['timeout', 'build_dir', 'alltests',

			       'filter_glob', 'kernel_args', 'run_isolated'])

KunitParseRequest = namedtuple('KunitParseRequest',

			       ['raw_output', 'build_dir', 'json'])

KunitRequest = namedtuple('KunitRequest', ['raw_output','timeout', 'jobs',

					   'build_dir', 'alltests', 'filter_glob',

					   'kernel_args', 'run_isolated', 'json', 'make_options'])



KernelDirectoryPath = sys.argv[0].split('tools/testing/kunit/')[0]



class KunitStatus(Enum):

class KunitStatus(Enum):

	SUCCESS = auto()

	SUCCESS = auto()

	CONFIG_FAILURE = auto()

	CONFIG_FAILURE = auto()

	BUILD_FAILURE = auto()

	BUILD_FAILURE = auto()

	TEST_FAILURE = auto()

	TEST_FAILURE = auto()





@dataclass

class KunitResult:

	status: KunitStatus

	result: Any

	elapsed_time: float



@dataclass

class KunitConfigRequest:

	build_dir: str

	make_options: Optional[List[str]]



@dataclass

class KunitBuildRequest(KunitConfigRequest):

	jobs: int

	alltests: bool



@dataclass

class KunitParseRequest:

	raw_output: Optional[str]

	build_dir: str

	json: Optional[str]



@dataclass

class KunitExecRequest(KunitParseRequest):

	timeout: int

	alltests: bool

	filter_glob: str

	kernel_args: Optional[List[str]]

	run_isolated: Optional[str]



@dataclass

class KunitRequest(KunitExecRequest, KunitBuildRequest):

	pass





KernelDirectoryPath = sys.argv[0].split('tools/testing/kunit/')[0]



def get_kernel_root_path() -> str:

def get_kernel_root_path() -> str:

	path = sys.argv[0] if not __file__ else __file__

	path = sys.argv[0] if not __file__ else __file__

	parts = os.path.realpath(path).split('tools/testing/kunit')

	parts = os.path.realpath(path).split('tools/testing/kunit')
@@ -121,8 +140,7 @@ def _suites_from_test_list(tests: List[str]) -> List[str]: 












def exec_tests(linux: kunit_kernel.LinuxSourceTree, request: KunitExecRequest,

def exec_tests(linux: kunit_kernel.LinuxSourceTree, request: KunitExecRequest) -> KunitResult:

	       parse_request: KunitParseRequest) -> KunitResult:

	filter_globs = [request.filter_glob]

	filter_globs = [request.filter_glob]

	if request.run_isolated:

	if request.run_isolated:

		tests = _list_tests(linux, request)

		tests = _list_tests(linux, request)
@@ -147,7 +165,7 @@ def exec_tests(linux: kunit_kernel.LinuxSourceTree, request: KunitExecRequest, 
			filter_glob=filter_glob,

			filter_glob=filter_glob,

			build_dir=request.build_dir)

			build_dir=request.build_dir)





		result = parse_tests(parse_request, run_result)

		result = parse_tests(request, run_result)

		# run_kernel() doesn't block on the kernel exiting.

		# run_kernel() doesn't block on the kernel exiting.

		# That only happens after we get the last line of output from `run_result`.

		# That only happens after we get the last line of output from `run_result`.

		# So exec_time here actually contains parsing + execution time, which is fine.

		# So exec_time here actually contains parsing + execution time, which is fine.
@@ -217,27 +235,15 @@ def run_tests(linux: kunit_kernel.LinuxSourceTree, 
	      request: KunitRequest) -> KunitResult:

	      request: KunitRequest) -> KunitResult:

	run_start = time.time()

	run_start = time.time()





	config_request = KunitConfigRequest(request.build_dir,

	config_result = config_tests(linux, request)

					    request.make_options)

	config_result = config_tests(linux, config_request)

	if config_result.status != KunitStatus.SUCCESS:

	if config_result.status != KunitStatus.SUCCESS:

		return config_result

		return config_result





	build_request = KunitBuildRequest(request.jobs, request.build_dir,

	build_result = build_tests(linux, request)

					  request.alltests,

					  request.make_options)

	build_result = build_tests(linux, build_request)

	if build_result.status != KunitStatus.SUCCESS:

	if build_result.status != KunitStatus.SUCCESS:

		return build_result

		return build_result





	exec_request = KunitExecRequest(request.timeout, request.build_dir,

	exec_result = exec_tests(linux, request)

				 request.alltests, request.filter_glob,

				 request.kernel_args, request.run_isolated)

	parse_request = KunitParseRequest(request.raw_output,

					  request.build_dir,

					  request.json)



	exec_result = exec_tests(linux, exec_request, parse_request)





	run_end = time.time()

	run_end = time.time()
@@ -413,16 +419,16 @@ def main(argv, linux=None): 
					cross_compile=cli_args.cross_compile,

					cross_compile=cli_args.cross_compile,

					qemu_config_path=cli_args.qemu_config)

					qemu_config_path=cli_args.qemu_config)





		request = KunitRequest(cli_args.raw_output,

		request = KunitRequest(build_dir=cli_args.build_dir,

				       cli_args.timeout,

				       make_options=cli_args.make_options,

				       cli_args.jobs,

				       jobs=cli_args.jobs,

				       cli_args.build_dir,

				       alltests=cli_args.alltests,

				       cli_args.alltests,

				       raw_output=cli_args.raw_output,

				       cli_args.filter_glob,

				       json=cli_args.json,

				       cli_args.kernel_args,

				       timeout=cli_args.timeout,

				       cli_args.run_isolated,

				       filter_glob=cli_args.filter_glob,

				       cli_args.json,

				       kernel_args=cli_args.kernel_args,

				       cli_args.make_options)

				       run_isolated=cli_args.run_isolated)

		result = run_tests(linux, request)

		result = run_tests(linux, request)

		if result.status != KunitStatus.SUCCESS:

		if result.status != KunitStatus.SUCCESS:

			sys.exit(1)

			sys.exit(1)
@@ -439,8 +445,8 @@ def main(argv, linux=None): 
					cross_compile=cli_args.cross_compile,

					cross_compile=cli_args.cross_compile,

					qemu_config_path=cli_args.qemu_config)

					qemu_config_path=cli_args.qemu_config)





		request = KunitConfigRequest(cli_args.build_dir,

		request = KunitConfigRequest(build_dir=cli_args.build_dir,

					     cli_args.make_options)

					     make_options=cli_args.make_options)

		result = config_tests(linux, request)

		result = config_tests(linux, request)

		kunit_parser.print_with_timestamp((

		kunit_parser.print_with_timestamp((

			'Elapsed time: %.3fs\n') % (

			'Elapsed time: %.3fs\n') % (
@@ -456,10 +462,10 @@ def main(argv, linux=None): 
					cross_compile=cli_args.cross_compile,

					cross_compile=cli_args.cross_compile,

					qemu_config_path=cli_args.qemu_config)

					qemu_config_path=cli_args.qemu_config)





		request = KunitBuildRequest(cli_args.jobs,

		request = KunitBuildRequest(build_dir=cli_args.build_dir,

					    cli_args.build_dir,

					    make_options=cli_args.make_options,

					    cli_args.alltests,

					    jobs=cli_args.jobs,

					    cli_args.make_options)

					    alltests=cli_args.alltests)

		result = build_tests(linux, request)

		result = build_tests(linux, request)

		kunit_parser.print_with_timestamp((

		kunit_parser.print_with_timestamp((

			'Elapsed time: %.3fs\n') % (

			'Elapsed time: %.3fs\n') % (
@@ -475,16 +481,15 @@ def main(argv, linux=None): 
					cross_compile=cli_args.cross_compile,

					cross_compile=cli_args.cross_compile,

					qemu_config_path=cli_args.qemu_config)

					qemu_config_path=cli_args.qemu_config)





		exec_request = KunitExecRequest(cli_args.timeout,

		exec_request = KunitExecRequest(raw_output=cli_args.raw_output,

						cli_args.build_dir,

						build_dir=cli_args.build_dir,

						cli_args.alltests,

						json=cli_args.json,

						cli_args.filter_glob,

						timeout=cli_args.timeout,

						cli_args.kernel_args,

						alltests=cli_args.alltests,

						cli_args.run_isolated)

						filter_glob=cli_args.filter_glob,

		parse_request = KunitParseRequest(cli_args.raw_output,

						kernel_args=cli_args.kernel_args,

						  cli_args.build_dir,

						run_isolated=cli_args.run_isolated)

						  cli_args.json)

		result = exec_tests(linux, exec_request)

		result = exec_tests(linux, exec_request, parse_request)

		kunit_parser.print_with_timestamp((

		kunit_parser.print_with_timestamp((

			'Elapsed time: %.3fs\n') % (result.elapsed_time))

			'Elapsed time: %.3fs\n') % (result.elapsed_time))

		if result.status != KunitStatus.SUCCESS:

		if result.status != KunitStatus.SUCCESS:
@@ -496,9 +501,9 @@ def main(argv, linux=None): 
		else:

		else:

			with open(cli_args.file, 'r', errors='backslashreplace') as f:

			with open(cli_args.file, 'r', errors='backslashreplace') as f:

				kunit_output = f.read().splitlines()

				kunit_output = f.read().splitlines()

		request = KunitParseRequest(cli_args.raw_output,

		request = KunitParseRequest(raw_output=cli_args.raw_output,

					    None,

					    build_dir='',

					    cli_args.json)

					    json=cli_args.json)

		result = parse_tests(request, kunit_output)

		result = parse_tests(request, kunit_output)

		if result.status != KunitStatus.SUCCESS:

		if result.status != KunitStatus.SUCCESS:

			sys.exit(1)

			sys.exit(1)

tools/testing/kunit/kunit_tool_test.py

+3 −3
Original line number Original line Diff line number Diff line
@@ -691,7 +691,7 @@ class KUnitMainTest(unittest.TestCase): 
		self.linux_source_mock.run_kernel.return_value = ['TAP version 14', 'init: random output'] + want

		self.linux_source_mock.run_kernel.return_value = ['TAP version 14', 'init: random output'] + want





		got = kunit._list_tests(self.linux_source_mock,

		got = kunit._list_tests(self.linux_source_mock,

				     kunit.KunitExecRequest(300, '.kunit', False, 'suite*', None, 'suite'))

				     kunit.KunitExecRequest(None, '.kunit', None, 300, False, 'suite*', None, 'suite'))





		self.assertEqual(got, want)

		self.assertEqual(got, want)

		# Should respect the user's filter glob when listing tests.

		# Should respect the user's filter glob when listing tests.
@@ -706,7 +706,7 @@ class KUnitMainTest(unittest.TestCase): 




		# Should respect the user's filter glob when listing tests.

		# Should respect the user's filter glob when listing tests.

		mock_tests.assert_called_once_with(mock.ANY,

		mock_tests.assert_called_once_with(mock.ANY,

				     kunit.KunitExecRequest(300, '.kunit', False, 'suite*.test*', None, 'suite'))

				     kunit.KunitExecRequest(None, '.kunit', None, 300, False, 'suite*.test*', None, 'suite'))

		self.linux_source_mock.run_kernel.assert_has_calls([

		self.linux_source_mock.run_kernel.assert_has_calls([

			mock.call(args=None, build_dir='.kunit', filter_glob='suite.test*', timeout=300),

			mock.call(args=None, build_dir='.kunit', filter_glob='suite.test*', timeout=300),

			mock.call(args=None, build_dir='.kunit', filter_glob='suite2.test*', timeout=300),

			mock.call(args=None, build_dir='.kunit', filter_glob='suite2.test*', timeout=300),
@@ -719,7 +719,7 @@ class KUnitMainTest(unittest.TestCase): 




		# Should respect the user's filter glob when listing tests.

		# Should respect the user's filter glob when listing tests.

		mock_tests.assert_called_once_with(mock.ANY,

		mock_tests.assert_called_once_with(mock.ANY,

				     kunit.KunitExecRequest(300, '.kunit', False, 'suite*', None, 'test'))

				     kunit.KunitExecRequest(None, '.kunit', None, 300, False, 'suite*', None, 'test'))

		self.linux_source_mock.run_kernel.assert_has_calls([

		self.linux_source_mock.run_kernel.assert_has_calls([

			mock.call(args=None, build_dir='.kunit', filter_glob='suite.test1', timeout=300),

			mock.call(args=None, build_dir='.kunit', filter_glob='suite.test1', timeout=300),

			mock.call(args=None, build_dir='.kunit', filter_glob='suite.test2', timeout=300),

			mock.call(args=None, build_dir='.kunit', filter_glob='suite.test2', timeout=300),