| # Copyright (C) 2018-2021 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 os |
| import logging |
| import time |
| |
| from webkitcorepy import string_utils |
| from webkitcorepy import TaskPool |
| |
| from webkitpy.common.iteration_compatibility import iteritems |
| from webkitpy.port.server_process import ServerProcess, _log as server_process_logger |
| |
| _log = logging.getLogger(__name__) |
| |
| |
| def setup_shard(port=None, devices=None): |
| if devices and getattr(port, 'DEVICE_MANAGER', None): |
| port.DEVICE_MANAGER.AVAILABLE_DEVICES = devices.get('available_devices', []) |
| port.DEVICE_MANAGER.INITIALIZED_DEVICES = devices.get('initialized_devices', None) |
| |
| return _Worker.setup(port=port) |
| |
| |
| def run_shard(name, *tests): |
| return _Worker.instance.run(name, *tests) |
| |
| |
| def report_result(worker, test, status, output): |
| if status == Runner.STATUS_PASSED and (not output or Runner.instance.port.get_option('quiet')): |
| Runner.instance.printer.write_update('{} {} {}'.format(worker, test, Runner.NAME_FOR_STATUS[status])) |
| else: |
| Runner.instance.printer.writeln('{} {} {}'.format(worker, test, Runner.NAME_FOR_STATUS[status])) |
| Runner.instance.results[test] = status, output |
| |
| |
| def teardown_shard(): |
| return _Worker.teardown() |
| |
| |
| class Runner(object): |
| STATUS_PASSED = 0 |
| STATUS_FAILED = 1 |
| STATUS_CRASHED = 2 |
| STATUS_TIMEOUT = 3 |
| STATUS_DISABLED = 4 |
| STATUS_RUNNING = 5 |
| |
| NAME_FOR_STATUS = [ |
| 'Passed', |
| 'Failed', |
| 'Crashed', |
| 'Timeout', |
| 'Disabled', |
| ] |
| |
| instance = None |
| |
| def __init__(self, port, printer): |
| self.port = port |
| self.printer = printer |
| self.tests_run = 0 |
| self._num_workers = 1 |
| self._has_logged_for_test = True # Suppress an empty line between "Running tests" and the first test's output. |
| self.results = {} |
| |
| # FIXME API tests should run as an app, we won't need this function <https://bugs.webkit.org/show_bug.cgi?id=175204> |
| @staticmethod |
| def command_for_port(port, args): |
| if (port.get_option('force')): |
| args.append('--gtest_also_run_disabled_tests=1') |
| if getattr(port, 'DEVICE_MANAGER', None): |
| assert port.DEVICE_MANAGER.INITIALIZED_DEVICES |
| return ['/usr/bin/xcrun', 'simctl', 'spawn', port.DEVICE_MANAGER.INITIALIZED_DEVICES[0].udid] + args |
| elif 'device' in port.port_name: |
| raise RuntimeError('Running api tests on {} is not supported'.format(port.port_name)) |
| elif port.host.platform.is_win(): |
| args[0] = os.path.splitext(args[0])[0] + '.exe' |
| return args |
| |
| @staticmethod |
| def _shard_tests(tests): |
| shards = {} |
| for test in tests: |
| shard_prefix = '.'.join(test.split('.')[:-1]) |
| if shard_prefix not in shards: |
| shards[shard_prefix] = [] |
| shards[shard_prefix].append(test) |
| return shards |
| |
| def run(self, tests, num_workers): |
| if not tests: |
| return |
| |
| self.printer.write_update('Sharding tests ...') |
| shards = Runner._shard_tests(tests) |
| |
| original_level = server_process_logger.level |
| server_process_logger.setLevel(logging.CRITICAL) |
| |
| try: |
| if Runner.instance: |
| raise RuntimeError('Cannot nest API test runners') |
| Runner.instance = self |
| self._num_workers = min(num_workers, len(shards)) |
| |
| devices = None |
| if getattr(self.port, 'DEVICE_MANAGER', None): |
| devices = dict( |
| available_devices=self.port.DEVICE_MANAGER.AVAILABLE_DEVICES, |
| initialized_devices=self.port.DEVICE_MANAGER.INITIALIZED_DEVICES, |
| ) |
| |
| with TaskPool( |
| workers=self._num_workers, |
| setup=setup_shard, setupkwargs=dict(port=self.port, devices=devices), teardown=teardown_shard, |
| ) as pool: |
| for name, tests in iteritems(shards): |
| pool.do(run_shard, name, *tests) |
| pool.wait() |
| |
| finally: |
| server_process_logger.setLevel(original_level) |
| Runner.instance = None |
| |
| def result_map_by_status(self, status=None): |
| map = {} |
| for test_name, result in iteritems(self.results): |
| if result[0] == status: |
| map[test_name] = result[1] |
| return map |
| |
| |
| class _Worker(object): |
| instance = None |
| |
| @classmethod |
| def setup(cls, port=None): |
| cls.instance = cls(port) |
| |
| @classmethod |
| def teardown(cls): |
| cls.instance = None |
| |
| def __init__(self, port): |
| self._port = port |
| self.host = port.host |
| |
| # ServerProcess doesn't allow for a timeout of 'None,' this uses a week instead of None. |
| self._timeout = int(self._port.get_option('timeout')) if self._port.get_option('timeout') else 60 * 24 * 7 |
| |
| @classmethod |
| def _filter_noisy_output(cls, output): |
| result = '' |
| for line in output.splitlines(): |
| if line.lstrip().startswith('objc['): |
| continue |
| result += line + '\n' |
| return result |
| |
| def _run_single_test(self, binary_name, test): |
| server_process = ServerProcess( |
| self._port, binary_name, |
| Runner.command_for_port(self._port, [self._port._build_path(binary_name), '--gtest_filter={}'.format(test)]), |
| env=self._port.environment_for_api_tests()) |
| |
| status = Runner.STATUS_RUNNING |
| if test.split('.')[1].startswith('DISABLED_') and not self._port.get_option('force'): |
| status = Runner.STATUS_DISABLED |
| |
| stdout_buffer = '' |
| stderr_buffer = '' |
| |
| try: |
| deadline = time.time() + self._timeout |
| if status != Runner.STATUS_DISABLED: |
| server_process.start() |
| |
| while status == Runner.STATUS_RUNNING: |
| stdout_line, stderr_line = server_process.read_either_stdout_or_stderr_line(deadline) |
| if not stderr_line and not stdout_line: |
| break |
| |
| if stderr_line: |
| stderr_line = string_utils.decode(stderr_line, target_type=str) |
| stderr_buffer += stderr_line |
| _log.error(stderr_line[:-1]) |
| if stdout_line: |
| stdout_line = string_utils.decode(stdout_line, target_type=str) |
| if '**PASS**' in stdout_line: |
| status = Runner.STATUS_PASSED |
| elif '**FAIL**' in stdout_line: |
| status = Runner.STATUS_FAILED |
| else: |
| stdout_buffer += stdout_line |
| _log.error(stdout_line[:-1]) |
| |
| if status == Runner.STATUS_DISABLED: |
| pass |
| elif server_process.timed_out: |
| status = Runner.STATUS_TIMEOUT |
| elif server_process.has_crashed(): |
| status = Runner.STATUS_CRASHED |
| elif status == Runner.STATUS_RUNNING: |
| status = Runner.STATUS_FAILED |
| |
| finally: |
| remaining_stderr = string_utils.decode(server_process.pop_all_buffered_stderr(), target_type=str) |
| remaining_stdout = string_utils.decode(server_process.pop_all_buffered_stdout(), target_type=str) |
| for line in (remaining_stdout + remaining_stderr).splitlines(False): |
| _log.error(line) |
| output_buffer = stderr_buffer + stdout_buffer + remaining_stderr + remaining_stdout |
| server_process.stop() |
| |
| TaskPool.Process.queue.send(TaskPool.Task( |
| report_result, None, TaskPool.Process.name, |
| '{}.{}'.format(binary_name, test), |
| status, |
| self._filter_noisy_output(output_buffer), |
| )) |
| |
| def run(self, name, *tests): |
| binary_name = name.split('.')[0] |
| remaining_tests = ['.'.join(test.split('.')[1:]) for test in tests] |
| |
| # Try to run the shard in a single process. |
| while remaining_tests and not self._port.get_option('run_singly'): |
| starting_length = len(remaining_tests) |
| server_process = ServerProcess( |
| self._port, binary_name, |
| Runner.command_for_port(self._port, [ |
| self._port._build_path(binary_name), '--gtest_filter={}'.format(':'.join(remaining_tests)) |
| ]), env=self._port.environment_for_api_tests()) |
| |
| try: |
| deadline = time.time() + self._timeout |
| last_test = None |
| last_status = None |
| stdout_buffer = '' |
| |
| server_process.start() |
| while remaining_tests: |
| stdout = string_utils.decode(server_process.read_stdout_line(deadline), target_type=str) |
| |
| # If we've triggered a timeout, we don't know which test caused it. Break out and run singly. |
| if stdout is None and server_process.timed_out: |
| break |
| |
| if stdout is None and server_process.has_crashed(): |
| # It's possible we crashed before printing anything. |
| if last_status == Runner.STATUS_PASSED: |
| last_test = None |
| else: |
| last_status = Runner.STATUS_CRASHED |
| break |
| |
| assert stdout is not None |
| stdout_split = stdout.rstrip().split(' ') |
| if len(stdout_split) != 2 or not (stdout_split[0].startswith('**') and stdout_split[0].endswith('**')): |
| stdout_buffer += stdout |
| continue |
| if last_test is not None: |
| remaining_tests.remove(last_test) |
| |
| for line in stdout_buffer.splitlines(False): |
| _log.error(line) |
| TaskPool.Process.queue.send(TaskPool.Task( |
| report_result, None, TaskPool.Process.name, |
| '{}.{}'.format(binary_name, last_test), |
| last_status, stdout_buffer, |
| )) |
| deadline = time.time() + self._timeout |
| stdout_buffer = '' |
| |
| if '**PASS**' == stdout_split[0]: |
| last_status = Runner.STATUS_PASSED |
| else: |
| last_status = Runner.STATUS_FAILED |
| last_test = stdout_split[1] |
| |
| # We assume that stderr is only relevant if there is a crash (meaning we triggered an assert) |
| if last_test: |
| remaining_tests.remove(last_test) |
| stdout_buffer += string_utils.decode(server_process.pop_all_buffered_stdout(), target_type=str) |
| stderr_buffer = string_utils.decode(server_process.pop_all_buffered_stderr(), target_type=str) if last_status == Runner.STATUS_CRASHED else '' |
| for line in (stdout_buffer + stderr_buffer).splitlines(keepends=False): |
| _log.error(line) |
| |
| TaskPool.Process.queue.send(TaskPool.Task( |
| report_result, None, TaskPool.Process.name, |
| '{}.{}'.format(binary_name, last_test), |
| last_status, |
| self._filter_noisy_output(stdout_buffer + stderr_buffer), |
| )) |
| |
| if server_process.timed_out: |
| break |
| |
| # If we weren't able to determine the results for any tests, we need to run what remains singly. |
| if starting_length == len(remaining_tests): |
| break |
| finally: |
| server_process.stop() |
| |
| # Now, just try and run the rest of the tests singly. |
| for test in remaining_tests: |
| self._run_single_test(binary_name, test) |
