blob: f67add88c93af994ec454bb3dd1feefb9010d975 [file] [log] [blame]
# Copyright (C) 2018-2019 Apple Inc. All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions
# are met:
# 1. Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
# 2. Redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer in the
# documentation and/or other materials provided with the distribution.
#
# THIS SOFTWARE IS PROVIDED BY APPLE INC. AND ITS CONTRIBUTORS ``AS IS'' AND ANY
# EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
# DISCLAIMED. IN NO EVENT SHALL APPLE INC. OR ITS CONTRIBUTORS BE LIABLE FOR ANY
# DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
# (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
# LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON
# ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
# SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
import json
import logging
import time
from webkitpy.api_tests.runner import Runner
from webkitpy.common.system.executive import ScriptError
from webkitpy.results.upload import Upload
from webkitpy.xcode.simulated_device import DeviceRequest, SimulatedDeviceManager
_log = logging.getLogger(__name__)
class Manager(object):
"""A class for managing running API and WTF tests
"""
SUCCESS = 0
FAILED_BUILD_CHECK = 1
FAILED_COLLECT_TESTS = 2
FAILED_TESTS = 3
FAILED_UPLOAD = 4
def __init__(self, port, options, stream):
self._port = port
self.host = port.host
self._options = options
self._stream = stream
@staticmethod
def _test_list_from_output(output, prefix=''):
result = []
current_test_suite = None
for line in output.split('\n'):
striped_line = line.lstrip().rstrip()
if not striped_line:
continue
if striped_line[-1] == '.':
current_test_suite = striped_line[:-1]
else:
striped_line = striped_line.lstrip()
if ' ' in striped_line:
continue
val = '{}{}.{}'.format(prefix, current_test_suite, striped_line)
if val not in result:
result.append(val)
return result
@staticmethod
def _find_test_subset(superset, arg_filter):
result = []
for arg in arg_filter:
split_arg = arg.split('.')
for test in superset:
# Might match <binary>.<suite>.<test> or just <suite>.<test>
split_test = test.split('.')
if len(split_arg) == 1:
if test not in result and (arg == split_test[0] or arg == split_test[1]):
result.append(test)
elif len(split_arg) == 2:
if test not in result and (split_arg == split_test[0:2] or split_arg == split_test[1:3]):
result.append(test)
else:
if arg == test and test not in result:
result.append(test)
return result
def _collect_tests(self, args):
available_tests = []
specified_binaries = self._binaries_for_arguments(args)
for canonicalized_binary, path in self._port.path_to_api_test_binaries().iteritems():
if canonicalized_binary not in specified_binaries:
continue
try:
output = self.host.executive.run_command(
Runner.command_for_port(self._port, [path, '--gtest_list_tests']),
env=self._port.environment_for_api_tests())
available_tests += Manager._test_list_from_output(output, '{}.'.format(canonicalized_binary))
except ScriptError:
_log.error('Failed to list {} tests'.format(canonicalized_binary))
raise
if len(args) == 0:
return sorted(available_tests)
return sorted(Manager._find_test_subset(available_tests, args))
@staticmethod
def _print_test_result(stream, test_name, output):
stream.writeln(' {}'.format(test_name))
has_output = False
for line in output.splitlines():
stream.writeln(' {}'.format(line))
has_output = True
if has_output:
stream.writeln('')
return not has_output
def _print_tests_result_with_status(self, status, runner):
mapping = runner.result_map_by_status(status)
if mapping:
self._stream.writeln(runner.NAME_FOR_STATUS[status])
self._stream.writeln('')
need_newline = False
for test, output in mapping.iteritems():
need_newline = Manager._print_test_result(self._stream, test, output)
if need_newline:
self._stream.writeln('')
def _initialize_devices(self):
if 'simulator' in self._port.port_name:
SimulatedDeviceManager.initialize_devices(DeviceRequest(self._port.DEVICE_TYPE, allow_incomplete_match=True), self.host, simulator_ui=False)
elif 'device' in self._port.port_name:
raise RuntimeError('Running api tests on {} is not supported'.format(self._port.port_name))
def _binaries_for_arguments(self, args):
if self._port.get_option('api_binary'):
return self._port.get_option('api_binary')
binaries = []
for arg in args:
candidate_binary = arg.split('.')[0]
if candidate_binary in binaries:
continue
if candidate_binary in self._port.path_to_api_test_binaries():
binaries.append(candidate_binary)
else:
# If the user specifies a test-name without a binary, we need to search both binaries
return self._port.path_to_api_test_binaries().keys()
return binaries or self._port.path_to_api_test_binaries().keys()
def run(self, args, json_output=None):
if json_output:
json_output = self.host.filesystem.abspath(json_output)
if not self.host.filesystem.isdir(self.host.filesystem.dirname(json_output)) or self.host.filesystem.isdir(json_output):
raise RuntimeError('Cannot write to {}'.format(json_output))
start_time = time.time()
self._stream.write_update('Checking build ...')
if not self._port.check_api_test_build(self._binaries_for_arguments(args)):
_log.error('Build check failed')
return Manager.FAILED_BUILD_CHECK
self._initialize_devices()
self._stream.write_update('Collecting tests ...')
try:
test_names = self._collect_tests(args)
except ScriptError:
self._stream.writeln('Failed to collect tests')
return Manager.FAILED_COLLECT_TESTS
self._stream.write_update('Found {} tests'.format(len(test_names)))
if len(test_names) == 0:
self._stream.writeln('No tests found')
return Manager.FAILED_COLLECT_TESTS
if self._port.get_option('dump'):
for test in test_names:
self._stream.writeln(test)
return Manager.SUCCESS
test_names = [test for test in test_names for _ in xrange(self._options.repeat_each)]
if self._options.repeat_each != 1:
_log.debug('Repeating each test {} times'.format(self._options.iterations))
try:
_log.info('Running tests')
runner = Runner(self._port, self._stream)
for i in xrange(self._options.iterations):
_log.debug('\nIteration {}'.format(i + 1))
runner.run(test_names, int(self._options.child_processes) if self._options.child_processes else self._port.default_child_processes())
except KeyboardInterrupt:
# If we receive a KeyboardInterrupt, print results.
self._stream.writeln('')
end_time = time.time()
successful = runner.result_map_by_status(runner.STATUS_PASSED)
disabled = len(runner.result_map_by_status(runner.STATUS_DISABLED))
_log.info('Ran {} tests of {} with {} successful'.format(len(runner.results) - disabled, len(test_names), len(successful)))
result_dictionary = {
'Skipped': [],
'Failed': [],
'Crashed': [],
'Timedout': [],
}
self._stream.writeln('-' * 30)
result = Manager.SUCCESS
if len(successful) * self._options.repeat_each + disabled == len(test_names):
self._stream.writeln('All tests successfully passed!')
if json_output:
self.host.filesystem.write_text_file(json_output, json.dumps(result_dictionary, indent=4))
else:
self._stream.writeln('Test suite failed')
self._stream.writeln('')
skipped = []
for test in test_names:
if test not in runner.results:
skipped.append(test)
result_dictionary['Skipped'].append({'name': test, 'output': None})
if skipped:
self._stream.writeln('Skipped {} tests'.format(len(skipped)))
self._stream.writeln('')
if self._options.verbose:
for test in skipped:
self._stream.writeln(' {}'.format(test))
self._print_tests_result_with_status(runner.STATUS_FAILED, runner)
self._print_tests_result_with_status(runner.STATUS_CRASHED, runner)
self._print_tests_result_with_status(runner.STATUS_TIMEOUT, runner)
for test, result in runner.results.iteritems():
status_to_string = {
runner.STATUS_FAILED: 'Failed',
runner.STATUS_CRASHED: 'Crashed',
runner.STATUS_TIMEOUT: 'Timedout',
}.get(result[0])
if not status_to_string:
continue
result_dictionary[status_to_string].append({'name': test, 'output': result[1]})
if json_output:
self.host.filesystem.write_text_file(json_output, json.dumps(result_dictionary, indent=4))
result = Manager.FAILED_TESTS
if self._options.report_urls:
self._stream.writeln('\n')
self._stream.write_update('Preparing upload data ...')
status_to_test_result = {
runner.STATUS_PASSED: None,
runner.STATUS_FAILED: Upload.Expectations.FAIL,
runner.STATUS_CRASHED: Upload.Expectations.CRASH,
runner.STATUS_TIMEOUT: Upload.Expectations.TIMEOUT,
}
upload = Upload(
suite='api-tests',
configuration=self._port.configuration_for_upload(self._port.target_host(0)),
details=Upload.create_details(options=self._options),
commits=self._port.commits_for_upload(),
run_stats=Upload.create_run_stats(
start_time=start_time,
end_time=end_time,
tests_skipped=len(result_dictionary['Skipped']),
),
results={test: Upload.create_test_result(actual=status_to_test_result[result[0]])
for test, result in runner.results.iteritems() if result[0] in status_to_test_result},
)
for url in self._options.report_urls:
self._stream.write_update('Uploading to {} ...'.format(url))
if not upload.upload(url, log_line_func=self._stream.writeln):
result = Manager.FAILED_UPLOAD
self._stream.writeln('Uploads completed!')
return result