kunit_tool: update to support kunit_watcher
Return objects when building config, building kernel, and parsing
test log in order to provide full context of kunit run.
Change-Id: If410ac7e2b747c80c952fd2a9983bb2f4577fd33
Google-Bug-Id: 116626752
Signed-off-by: Avi Kondareddy <avikr@google.com>
diff --git a/tools/testing/kunit/kunit.py b/tools/testing/kunit/kunit.py
index b9bacd0..7651d38 100755
--- a/tools/testing/kunit/kunit.py
+++ b/tools/testing/kunit/kunit.py
@@ -13,34 +13,47 @@
import kunit_new_template
import kunit_parser
-def run_tests(cli_args, linux):
+from collections import namedtuple
+
+KunitRequest = namedtuple('KunitRequest', ['raw_output','timeout'])
+
+KunitResult = namedtuple('KunitResult', ['status','result'])
+
+class KunitStatus(object):
+ SUCCESS = 'SUCCESS'
+ CONFIG_FAILURE = 'CONFIG_FAILURE'
+ BUILD_FAILURE = 'BUILD_FAILURE'
+ TEST_FAILURE = 'TEST_FAILURE'
+
+def run_tests(linux: kunit_kernel.LinuxSourceTree,
+ request: KunitRequest) -> KunitResult:
config_start = time.time()
- success = linux.build_reconfig()
+ config_result = linux.build_reconfig()
config_end = time.time()
- if not success:
- return
+ if config_result.status != kunit_kernel.ConfigStatus.SUCCESS:
+ return KunitResult(KunitStatus.CONFIG_FAILURE, config_result)
print(kunit_parser.timestamp('Building KUnit Kernel ...'))
build_start = time.time()
- success = linux.build_um_kernel()
+ build_result = linux.build_um_kernel()
build_end = time.time()
- if not success:
- return
+ if build_result.status != kunit_kernel.BuildStatus.SUCCESS:
+ return KunitResult(KunitStatus.BUILD_FAILURE, build_result)
print(kunit_parser.timestamp('Starting KUnit Kernel ...'))
test_start = time.time()
- if cli_args.raw_output:
+ test_result = kunit_parser.TestResult(kunit_parser.TestStatus.SUCCESS,
+ [],
+ 'Tests not Parsed.')
+ if request.raw_output:
kunit_parser.raw_output(
- linux.run_kernel(timeout=cli_args.timeout))
+ linux.run_kernel(timeout=request.timeout))
else:
- for line in kunit_parser.parse_run_tests(
- kunit_parser.isolate_kunit_output(
- linux.run_kernel(
- timeout=cli_args.timeout))):
- print(line)
-
+ test_result = kunit_parser.parse_run_tests(
+ kunit_parser.isolate_kunit_output(
+ linux.run_kernel(timeout=request.timeout)))
test_end = time.time()
print(kunit_parser.timestamp((
@@ -51,13 +64,18 @@
build_end - build_start,
test_end - test_start)))
+ if test_result.status != kunit_parser.TestStatus.SUCCESS:
+ return KunitResult(KunitStatus.TEST_FAILURE, test_result)
+ else:
+ return KunitResult(KunitStatus.SUCCESS, test_result)
+
def print_test_skeletons(cli_args):
kunit_new_template.create_skeletons_from_path(
cli_args.path,
namespace_prefix=cli_args.namespace_prefix,
print_test_only=cli_args.print_test_only)
-def main(argv, linux=kunit_kernel.LinuxSourceTree()):
+def main(argv, linux):
parser = argparse.ArgumentParser(
description='Helps writing and running KUnit tests.')
subparser = parser.add_subparsers(dest='subcommand')
@@ -94,9 +112,10 @@
if cli_args.subcommand == 'new':
print_test_skeletons(cli_args)
elif cli_args.subcommand == 'run':
- run_tests(cli_args, linux)
+ request = KunitRequest(cli_args.raw_output, cli_args.timeout)
+ run_tests(linux, request)
else:
parser.print_help()
if __name__ == '__main__':
- main(sys.argv[1:])
+ main(sys.argv[1:], kunit_kernel.LinuxSourceTree())
diff --git a/tools/testing/kunit/kunit_config.py b/tools/testing/kunit/kunit_config.py
index 183bd5e..4196966 100644
--- a/tools/testing/kunit/kunit_config.py
+++ b/tools/testing/kunit/kunit_config.py
@@ -58,3 +58,10 @@
def read_from_file(self, path: str) -> None:
with open(path, 'r') as f:
self.parse_from_string(f.read())
+
+class KunitConfigProvider(object):
+
+ def get_kconfig(self) -> Kconfig:
+ kconfig = Kconfig()
+ kconfig.read_from_file('kunitconfig')
+ return kconfig
diff --git a/tools/testing/kunit/kunit_kernel.py b/tools/testing/kunit/kunit_kernel.py
index 5ef8fcd..8a51ba3 100644
--- a/tools/testing/kunit/kunit_kernel.py
+++ b/tools/testing/kunit/kunit_kernel.py
@@ -5,9 +5,24 @@
import os
import kunit_config
+import kunit_parser
KCONFIG_PATH = '.config'
+from collections import namedtuple
+
+ConfigResult = namedtuple('ConfigResult', ['status','info'])
+
+BuildResult = namedtuple('BuildResult', ['status','info'])
+
+class ConfigStatus(object):
+ SUCCESS = 'SUCCESS'
+ FAILURE = 'FAILURE'
+
+class BuildStatus(object):
+ SUCCESS = 'SUCCESS'
+ FAILURE = 'FAILURE'
+
class ConfigError(Exception):
"""Represents an error trying to configure the Linux kernel."""
@@ -56,16 +71,22 @@
except subprocess.TimeoutExpired:
process.terminate()
timed_out = True
- return process, timed_out
+ output, _ = process.communicate()
+ output = output.decode('ascii')
+ if timed_out:
+ output += kunit_parser.TIMED_OUT_LOG_ENTRY + '\n'
+
+ return output
class LinuxSourceTree(object):
"""Represents a Linux kernel source tree with KUnit tests."""
- def __init__(self):
- self._kconfig = kunit_config.Kconfig()
- self._kconfig.read_from_file('kunitconfig')
- self._ops = LinuxSourceTreeOperations()
+ def __init__(self,
+ kconfig_provider=kunit_config.KunitConfigProvider(),
+ linux_build_operations=LinuxSourceTreeOperations()):
+ self._kconfig = kconfig_provider.get_kconfig()
+ self._ops = linux_build_operations
def clean(self):
try:
@@ -81,13 +102,14 @@
self._ops.make_olddefconfig()
except ConfigError as e:
logging.error(e)
- return False
+ return ConfigResult(ConfigStatus.FAILURE, e.message)
validated_kconfig = kunit_config.Kconfig()
validated_kconfig.read_from_file(KCONFIG_PATH)
if not self._kconfig.is_subset_of(validated_kconfig):
- logging.error('Provided Kconfig is not contained in validated .config!')
- return False
- return True
+ message = 'Provided Kconfig is not contained in validated .config!'
+ logging.error(message)
+ return ConfigResult(ConfigStatus.FAILURE, message)
+ return ConfigResult(ConfigStatus.SUCCESS, 'Build config!')
def build_reconfig(self):
"""Creates a new .config if it is not a subset of the kunitconfig."""
@@ -99,7 +121,7 @@
os.remove(KCONFIG_PATH)
return self.build_config()
else:
- return True
+ return ConfigResult(ConfigStatus.SUCCESS, 'Already built.')
else:
print('Generating .config ...')
return self.build_config()
@@ -110,21 +132,19 @@
self._ops.make()
except (ConfigError, BuildError) as e:
logging.error(e)
- return False
+ return BuildResult(BuildStatus.FAILURE, e.message)
used_kconfig = kunit_config.Kconfig()
used_kconfig.read_from_file(KCONFIG_PATH)
if not self._kconfig.is_subset_of(used_kconfig):
- logging.error('Provided Kconfig is not contained in final config!')
- return False
- return True
+ message = 'Provided Kconfig is not contained in final config!'
+ logging.error(message)
+ return BuildResult(BuildStatus.FAILURE, message)
+ return BuildResult(BuildStatus.SUCCESS, 'Built kernel!')
def run_kernel(self, args=[], timeout=None):
args.extend(['mem=256M'])
- process, timed_out = self._ops.linux_bin(args, timeout)
+ raw_log = self._ops.linux_bin(args, timeout)
with open('test.log', 'w') as f:
- for line in process.stdout:
- f.write(line.rstrip().decode('ascii') + '\n')
- yield line.rstrip().decode('ascii')
- if timed_out:
- f.write('Timeout Reached - Process Terminated\n')
- yield 'Timeout Reached - Process Terminated\n'
+ for line in raw_log.split('\n'):
+ f.write(line.rstrip() + '\n')
+ yield line.rstrip()
diff --git a/tools/testing/kunit/kunit_parser.py b/tools/testing/kunit/kunit_parser.py
index d94b144..02b4abe 100644
--- a/tools/testing/kunit/kunit_parser.py
+++ b/tools/testing/kunit/kunit_parser.py
@@ -1,9 +1,26 @@
import re
from datetime import datetime
+from collections import namedtuple
+
+TestResult = namedtuple('TestResult', ['status','modules','log'])
+
+TestModule = namedtuple('TestModule', ['status','name','cases'])
+
+TestCase = namedtuple('TestCase', ['status','name','log'])
+
+class TestStatus(object):
+ SUCCESS = 'SUCCESS'
+ FAILURE = 'FAILURE'
+ TEST_CRASHED = 'TEST_CRASHED'
+ TIMED_OUT = 'TIMED_OUT'
+ KERNEL_CRASHED = 'KERNEL_CRASHED'
+
kunit_start_re = re.compile('console .* enabled')
kunit_end_re = re.compile('List of all partitions:')
+TIMED_OUT_LOG_ENTRY = 'Timeout Reached - Process Terminated'
+
class KernelCrashException(Exception):
pass
@@ -47,8 +64,8 @@
def parse_run_tests(kernel_output):
test_case_output = re.compile('^kunit .*?: (.*)$')
- test_module_success = re.compile('^kunit .*: all tests passed')
- test_module_fail = re.compile('^kunit .*: one or more tests failed')
+ test_module_success = re.compile('^kunit (.*): all tests passed')
+ test_module_fail = re.compile('^kunit (.*): one or more tests failed')
test_case_success = re.compile('^kunit (.*): (.*) passed')
test_case_fail = re.compile('^kunit (.*): (.*) failed')
@@ -58,30 +75,59 @@
failed_tests = set()
crashed_tests = set()
+ test_status = TestStatus.SUCCESS
+ modules = []
+ log_list = []
+
+ def get_test_module_name(match):
+ return match.group(1)
+ def get_test_case_name(match):
+ return match.group(2)
+
def get_test_name(match):
return match.group(1) + ":" + match.group(2)
current_case_log = []
+ current_module_cases = []
did_kernel_crash = False
did_timeout = False
def end_one_test(match, log):
- log.clear()
+ del log[:]
total_tests.add(get_test_name(match))
- yield timestamp(DIVIDER)
+ print(timestamp(DIVIDER))
try:
for line in kernel_output:
+ log_list.append(line)
+
# Ignore module output:
- if (test_module_success.match(line) or
- test_module_fail.match(line)):
- yield timestamp(DIVIDER)
+ module_success = test_module_success.match(line)
+ if module_success:
+ print(timestamp(DIVIDER))
+ modules.append(
+ TestModule(TestStatus.SUCCESS,
+ get_test_module_name(module_success),
+ current_module_cases))
+ current_module_cases = []
+ continue
+ module_fail = test_module_fail.match(line)
+ if module_fail:
+ print(timestamp(DIVIDER))
+ modules.append(
+ TestModule(TestStatus.FAILURE,
+ get_test_module_name(module_fail),
+ current_module_cases))
+ current_module_cases = []
continue
match = re.match(test_case_success, line)
if match:
- yield timestamp(green("[PASSED] ") +
- get_test_name(match))
+ print(timestamp(green("[PASSED] ") + get_test_name(match)))
+ current_module_cases.append(
+ TestCase(TestStatus.SUCCESS,
+ get_test_case_name(match),
+ '\n'.join(current_case_log)))
end_one_test(match, current_case_log)
continue
@@ -90,26 +136,38 @@
# want to show and count it once.
if match and get_test_name(match) not in crashed_tests:
failed_tests.add(get_test_name(match))
- yield timestamp(red("[FAILED] " +
- get_test_name(match)))
- yield from timestamp_log(map(yellow, current_case_log))
- yield timestamp("")
+ print(timestamp(red("[FAILED] " + get_test_name(match))))
+ for out in timestamp_log(map(yellow, current_case_log)):
+ print(out)
+ print(timestamp(""))
+ current_module_cases.append(
+ TestCase(TestStatus.FAILURE,
+ get_test_case_name(match),
+ '\n'.join(current_case_log)))
+ if test_status != TestStatus.TEST_CRASHED:
+ test_status = TestStatus.FAILURE
end_one_test(match, current_case_log)
continue
match = re.match(test_case_crash, line)
if match:
crashed_tests.add(get_test_name(match))
- yield timestamp(yellow("[CRASH] " +
- get_test_name(match)))
- yield from timestamp_log(current_case_log)
- yield timestamp("")
+ print(timestamp(yellow("[CRASH] " + get_test_name(match))))
+ for out in timestamp_log(current_case_log):
+ print(out)
+ print(timestamp(""))
+ current_module_cases.append(
+ TestCase(TestStatus.TEST_CRASHED,
+ get_test_case_name(match),
+ '\n'.join(current_case_log)))
+ test_status = TestStatus.TEST_CRASHED
end_one_test(match, current_case_log)
continue
- if line == 'Timeout Reached - Process Terminated\n':
- yield timestamp(red("[TIMED-OUT] Process Terminated"))
+ if line.strip() == TIMED_OUT_LOG_ENTRY:
+ print(timestamp(red("[TIMED-OUT] Process Terminated")))
did_timeout = True
+ test_status = TestStatus.TIMED_OUT
break
# Strip off the `kunit module-name:` prefix
@@ -120,14 +178,15 @@
current_case_log.append(line)
except KernelCrashException:
did_kernel_crash = True
- yield timestamp(
- red("The KUnit kernel crashed unexpectedly and was " +
- "unable to finish running tests!"))
- yield timestamp(red("These are the logs from the most " +
- "recently running test:"))
- yield timestamp(DIVIDER)
- yield from timestamp_log(current_case_log)
- yield timestamp(DIVIDER)
+ test_status = TestStatus.KERNEL_CRASHED
+ print(timestamp(red("The KUnit kernel crashed unexpectedly and was " +
+ "unable to finish running tests!")))
+ print(timestamp(red("These are the logs from the most " +
+ "recently running test:")))
+ print(timestamp(DIVIDER))
+ for out in timestamp_log(current_case_log):
+ print(out)
+ print(timestamp(DIVIDER))
fmt = green if (len(failed_tests) + len(crashed_tests) == 0
and not did_kernel_crash and not did_timeout) else red
@@ -137,7 +196,7 @@
elif did_timeout:
message = "Before timing out:"
- yield timestamp(
- fmt(message + " %d tests run. %d failed. %d crashed." %
- (len(total_tests), len(failed_tests), len(crashed_tests))))
+ print(timestamp(fmt(message + " %d tests run. %d failed. %d crashed." %
+ (len(total_tests), len(failed_tests), len(crashed_tests)))))
+ return TestResult(test_status, modules, '\n'.join(log_list))
diff --git a/tools/testing/kunit/kunit_test.py b/tools/testing/kunit/kunit_test.py
index e7baf2868..f143358 100755
--- a/tools/testing/kunit/kunit_test.py
+++ b/tools/testing/kunit/kunit_test.py
@@ -9,6 +9,7 @@
import kunit_config
import kunit_parser
+import kunit_kernel
import kunit
test_tmpdir = ''
@@ -104,9 +105,9 @@
'test_data/test_is_test_passed-all_passed.log')
file = open(all_passed_log)
result = kunit_parser.parse_run_tests(file.readlines())
- self.assertContains(
- 'Testing complete. 3 tests run. 0 failed. 0 crashed.',
- result)
+ self.assertEqual(
+ kunit_parser.TestStatus.SUCCESS,
+ result.status)
file.close()
def test_parse_failed_test_log(self):
@@ -114,9 +115,9 @@
'test_data/test_is_test_passed-failure.log')
file = open(failed_log)
result = kunit_parser.parse_run_tests(file.readlines())
- self.assertContains(
- 'Testing complete. 3 tests run. 1 failed. 0 crashed.',
- result)
+ self.assertEqual(
+ kunit_parser.TestStatus.FAILURE,
+ result.status)
file.close()
def test_broken_test(self):
@@ -125,9 +126,9 @@
file = open(broken_log)
result = kunit_parser.parse_run_tests(
kunit_parser.isolate_kunit_output(file.readlines()))
- self.assertContains(
- 'Before the crash: 3 tests run. 0 failed. 0 crashed.',
- result)
+ self.assertEqual(
+ kunit_parser.TestStatus.KERNEL_CRASHED,
+ result.status)
file.close()
def test_no_tests(self):
@@ -136,9 +137,10 @@
file = open(empty_log)
result = kunit_parser.parse_run_tests(
kunit_parser.isolate_kunit_output(file.readlines()))
- self.assertContains(
- 'Testing complete. 0 tests run. 0 failed. 0 crashed.',
- result)
+ self.assertEqual(0, len(result.modules))
+ self.assertEqual(
+ kunit_parser.TestStatus.SUCCESS,
+ result.status)
file.close()
def test_crashed_test(self):
@@ -146,9 +148,9 @@
'test_data/test_is_test_passed-crash.log')
file = open(crashed_log)
result = kunit_parser.parse_run_tests(file.readlines())
- self.assertContains(
- 'Testing complete. 3 tests run. 0 failed. 1 crashed.',
- result)
+ self.assertEqual(
+ kunit_parser.TestStatus.TEST_CRASHED,
+ result.status)
file.close()
def test_timed_out_test(self):
@@ -156,9 +158,9 @@
'test_data/test_is_test_passed-timed_out.log')
file = open(timed_out_log)
result = kunit_parser.parse_run_tests(file.readlines())
- self.assertContains(
- 'Before timing out: 3 tests run. 0 failed. 0 crashed.',
- result)
+ self.assertEqual(
+ kunit_parser.TestStatus.TIMED_OUT,
+ result.status)
file.close()
class StrContains(str):
@@ -170,7 +172,12 @@
self.print_patch = mock.patch('builtins.print')
self.print_mock = self.print_patch.start()
self.linux_source_mock = mock.Mock()
- self.linux_source_mock.build_reconfig = mock.Mock()
+ self.linux_source_mock.build_reconfig = mock.Mock(
+ return_value=kunit_kernel.ConfigResult(
+ kunit_kernel.ConfigStatus.SUCCESS, ''))
+ self.linux_source_mock.build_um_kernel = mock.Mock(
+ return_value=kunit_kernel.BuildResult(
+ kunit_kernel.BuildStatus.SUCCESS, ''))
self.linux_source_mock.run_kernel = mock.Mock(return_value=[
'console 0 enabled',
'List of all partitions:'])
