blob: b7cf2e4efd42e098206da276ecad7565cef23ab9 [file] [log] [blame]
import re
from datetime import datetime
from collections import namedtuple
class TestStatus(object):
SUCCESS = 'SUCCESS'
FAILURE = 'FAILURE'
TEST_CRASHED = 'TEST_CRASHED'
TIMED_OUT = 'TIMED_OUT'
KERNEL_CRASHED = 'KERNEL_CRASHED'
class TestResult(object):
def __init__(self, status=TestStatus.SUCCESS, modules=None, kernel_log='',
log_lines=None):
self.status = status
self.modules = modules or []
self.kernel_log = kernel_log
self.log_lines = log_lines or []
def pretty_log(self, msg):
self.log_lines.append(msg)
def print_pretty_log(self):
print('\n'.join(self.log_lines))
TestModule = namedtuple('TestModule', ['status','name','cases'])
TestCase = namedtuple('TestCase', ['status','name','log'])
kunit_start_re = re.compile('console .* enabled')
kunit_end_re = re.compile('reboot: System halted')
TIMED_OUT_LOG_ENTRY = 'Timeout Reached - Process Terminated'
class KernelCrashException(Exception):
pass
def isolate_kunit_output(kernel_output):
started = False
for line in kernel_output:
if kunit_start_re.match(line):
started = True
elif kunit_end_re.match(line):
return
elif started:
yield line
# Output ended without encountering end marker, kernel probably panicked
# or crashed unexpectedly.
raise KernelCrashException()
def raw_output(kernel_output):
for line in kernel_output:
print(line)
DIVIDER = "=" * 30
RESET = '\033[0;0m'
def red(text):
return '\033[1;31m' + text + RESET
def yellow(text):
return '\033[1;33m' + text + RESET
def green(text):
return '\033[1;32m' + text + RESET
def timestamp(message):
return '[%s] %s' % (datetime.now().strftime('%H:%M:%S'), message)
def timestamp_log(log):
for m in log:
yield timestamp(m)
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_case_success = re.compile('^kunit (.*): (.*) passed')
test_case_fail = re.compile('^kunit (.*): (.*) failed')
test_case_crash = re.compile('^kunit (.*): (.*) crashed')
total_tests = set()
failed_tests = set()
crashed_tests = set()
test_result = TestResult()
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):
del log[:]
total_tests.add(get_test_name(match))
test_result.pretty_log(timestamp(DIVIDER))
try:
for line in kernel_output:
log_list.append(line)
# Ignore module output:
module_success = test_module_success.match(line)
if module_success:
test_result.pretty_log(timestamp(DIVIDER))
test_result.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:
test_result.pretty_log(timestamp(DIVIDER))
test_result.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:
test_result.pretty_log(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
match = re.match(test_case_fail, line)
# Crashed tests will report as both failed and crashed. We only
# 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))
test_result.pretty_log(timestamp(red("[FAILED] " +
get_test_name(match))))
for out in timestamp_log(map(yellow, current_case_log)):
test_result.pretty_log(out)
test_result.pretty_log(timestamp(""))
current_module_cases.append(
TestCase(TestStatus.FAILURE,
get_test_case_name(match),
'\n'.join(current_case_log)))
if test_result.status != TestStatus.TEST_CRASHED:
test_result.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))
test_result.pretty_log(timestamp(yellow("[CRASH] " +
get_test_name(match))))
for out in timestamp_log(current_case_log):
test_result.pretty_log(out)
test_result.pretty_log(timestamp(""))
current_module_cases.append(
TestCase(TestStatus.TEST_CRASHED,
get_test_case_name(match),
'\n'.join(current_case_log)))
test_result.status = TestStatus.TEST_CRASHED
end_one_test(match, current_case_log)
continue
if line.strip() == TIMED_OUT_LOG_ENTRY:
test_result.pretty_log(timestamp(red("[TIMED-OUT] " +
"Process Terminated")))
did_timeout = True
test_result.status = TestStatus.TIMED_OUT
break
# Strip off the `kunit module-name:` prefix
match = re.match(test_case_output, line)
if match:
current_case_log.append(match.group(1))
else:
current_case_log.append(line)
except KernelCrashException:
did_kernel_crash = True
test_result.status = TestStatus.KERNEL_CRASHED
test_result.pretty_log(timestamp(red("The KUnit kernel crashed " +
"unexpectedly and was unable " +
"to finish running tests!")))
test_result.pretty_log(timestamp(red("These are the logs from the most " +
"recently running test:")))
test_result.pretty_log(timestamp(DIVIDER))
for out in timestamp_log(current_case_log):
test_result.pretty_log(out)
test_result.pretty_log(timestamp(DIVIDER))
fmt = green if (len(failed_tests) + len(crashed_tests) == 0
and not did_kernel_crash and not did_timeout) else red
message = "Testing complete."
if did_kernel_crash:
message = "Before the crash:"
elif did_timeout:
message = "Before timing out:"
test_result.pretty_log(timestamp(fmt(message + " %d tests run. %d failed. %d crashed." %
(len(total_tests), len(failed_tests), len(crashed_tests)))))
test_result.kernel_log = '\n'.join(log_list)
return test_result