| # Copyright 2017 The Cobalt Authors. All Rights Reserved. |
| # |
| # Licensed under the Apache License, Version 2.0 (the "License"); |
| # you may not use this file except in compliance with the License. |
| # You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, software |
| # distributed under the License is distributed on an "AS IS" BASIS, |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| # See the License for the specific language governing permissions and |
| # limitations under the License. |
| |
| |
| from __future__ import absolute_import |
| from __future__ import division |
| from __future__ import print_function |
| |
| import argparse |
| import importlib |
| import logging |
| import random |
| import socket |
| import sys |
| import unittest |
| |
| import _env # pylint: disable=unused-import |
| from cobalt.black_box_tests import black_box_cobalt_runner |
| from cobalt.tools.automated_testing import cobalt_runner |
| from proxy_server import ProxyServer |
| from starboard.tools import abstract_launcher |
| from starboard.tools import build |
| |
| _PORT_SELECTION_RETRY_LIMIT = 10 |
| _PORT_SELECTION_RANGE = [5000, 7000] |
| _SERVER_EXIT_TIMEOUT_SECONDS = 30 |
| # These tests can only be run on platforms whose app launcher can send suspend/ |
| # resume signals. |
| _TESTS_NEEDING_SYSTEM_SIGNAL = [ |
| 'cancel_sync_loads_when_suspended', |
| 'preload_font', |
| 'preload_visibility', |
| 'signal_handler_doesnt_crash', |
| 'suspend_visibility', |
| 'timer_hit_after_preload', |
| 'timer_hit_in_preload', |
| ] |
| # These tests only need app launchers with webdriver. |
| _TESTS_NO_SIGNAL = [ |
| 'allow_eval', |
| 'disable_eval_with_csp', |
| 'persistent_cookie', |
| 'web_debugger', |
| 'web_platform_tests', |
| ] |
| # Location of test files. |
| _TEST_DIR_PATH = 'cobalt.black_box_tests.tests.' |
| # Platform dependent device parameters. |
| _device_params = None |
| # Binding address used to create the test server. |
| _binding_address = None |
| |
| |
| def GetDeviceParams(): |
| |
| global _device_params |
| _device_params = cobalt_runner.GetDeviceParamsFromCommandLine() |
| # Keep other modules from seeing these args |
| sys.argv = sys.argv[:1] |
| |
| |
| class BlackBoxTestCase(unittest.TestCase): |
| |
| def __init__(self, *args, **kwargs): |
| super(BlackBoxTestCase, self).__init__(*args, **kwargs) |
| self.device_params = _device_params |
| self.platform_config = build.GetPlatformConfig(_device_params.platform) |
| self.cobalt_config = self.platform_config.GetApplicationConfiguration( |
| 'cobalt') |
| |
| @classmethod |
| def setUpClass(cls): |
| super(BlackBoxTestCase, cls).setUpClass() |
| print('Running ' + cls.__name__) |
| |
| @classmethod |
| def tearDownClass(cls): |
| super(BlackBoxTestCase, cls).tearDownClass() |
| print('Done ' + cls.__name__) |
| |
| def CreateCobaltRunner(self, url, target_params=None): |
| all_target_params = list(target_params) if target_params else [] |
| if _device_params.target_params is not None: |
| all_target_params += _device_params.target_params |
| new_runner = black_box_cobalt_runner.BlackBoxCobaltRunner( |
| device_params=_device_params, url=url, target_params=all_target_params) |
| return new_runner |
| |
| def GetBindingAddress(self): |
| return _binding_address |
| |
| |
| def LoadTests(platform, config, device_id, out_directory): |
| |
| launcher = abstract_launcher.LauncherFactory( |
| platform, |
| 'cobalt', |
| config, |
| device_id=device_id, |
| target_params=None, |
| output_file=None, |
| out_directory=out_directory) |
| |
| if launcher.SupportsSuspendResume(): |
| test_targets = _TESTS_NEEDING_SYSTEM_SIGNAL + _TESTS_NO_SIGNAL |
| else: |
| test_targets = _TESTS_NO_SIGNAL |
| |
| test_suite = unittest.TestSuite() |
| for test in test_targets: |
| test_suite.addTest(unittest.TestLoader().loadTestsFromModule( |
| importlib.import_module(_TEST_DIR_PATH + test))) |
| return test_suite |
| |
| |
| class BlackBoxTests(object): |
| """Helper class to run all black box tests and return results.""" |
| |
| def __init__(self, test_name=None, proxy_port_number=None): |
| logging.basicConfig(level=logging.DEBUG) |
| GetDeviceParams() |
| |
| # Port number used to create the proxy server. If the --proxy target param |
| # is not set, a random free port is used. |
| if proxy_port_number is None: |
| proxy_port_number = str(self.GetUnusedPort(_binding_address)) |
| _device_params.target_params.append('--proxy=%s:%s' % (_binding_address, |
| proxy_port_number)) |
| |
| self.test_name = test_name |
| self.proxy_port_number = proxy_port_number |
| |
| # Test domains used in web platform tests to be resolved to the server |
| # binding address. |
| hosts = [ |
| 'web-platform.test', |
| 'www.web-platform.test', |
| 'www1.web-platform.test', |
| 'www2.web-platform.test', |
| 'xn--n8j6ds53lwwkrqhv28a.web-platform.test', |
| 'xn--lve-6lad.web-platform.test' |
| ] |
| self.host_resolve_map = dict([(host, _binding_address) for host in hosts]) |
| |
| def Run(self): |
| if self.proxy_port_number == '-1': |
| return 1 |
| logging.info('Using proxy port number: %s', self.proxy_port_number) |
| |
| with ProxyServer(port=self.proxy_port_number, |
| host_resolve_map=self.host_resolve_map): |
| if self.test_name: |
| suite = unittest.TestLoader().loadTestsFromModule( |
| importlib.import_module(_TEST_DIR_PATH + self.test_name)) |
| else: |
| suite = LoadTests(_device_params.platform, _device_params.config, |
| _device_params.device_id, |
| _device_params.out_directory) |
| return_code = not unittest.TextTestRunner( |
| verbosity=0, stream=sys.stdout).run(suite).wasSuccessful() |
| return return_code |
| |
| def GetUnusedPort(self, machine_address): |
| """Find a free port on the machine address by pinging with socket.""" |
| sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) |
| try: |
| for i in range(1, _PORT_SELECTION_RETRY_LIMIT): |
| port_number = random.randint(_PORT_SELECTION_RANGE[0], |
| _PORT_SELECTION_RANGE[1]) |
| result = sock.connect_ex((machine_address, port_number)) |
| if result != 0: |
| return port_number |
| if i == _PORT_SELECTION_RETRY_LIMIT - 1: |
| logging.error( |
| 'Can not find unused port on target machine within %s attempts.', |
| _PORT_SELECTION_RETRY_LIMIT) |
| return -1 |
| finally: |
| sock.close() |
| |
| |
| def main(): |
| parser = argparse.ArgumentParser() |
| parser.add_argument('--test_name', |
| help=('Name of test to be run. If not specified, all ' |
| 'tests are run.')) |
| parser.add_argument('--server_binding_address', |
| default='127.0.0.1', |
| help='Binding address used to create the test server.') |
| parser.add_argument('--proxy_port_number', |
| help=('Port number used to create the proxy http server' |
| 'that all black box tests are run through. If not' |
| 'specified, a random free port is used.')) |
| args, _ = parser.parse_known_args() |
| |
| global _binding_address |
| _binding_address = args.server_binding_address |
| test_object = BlackBoxTests(args.test_name, args.proxy_port_number) |
| sys.exit(test_object.Run()) |
| |
| |
| if __name__ == '__main__': |
| # Running this script on the command line and importing this file are |
| # different and create two modules. |
| # Import this module to ensure we are using the same module as the tests to |
| # make module-owned variables like device_param accessible to the tests. |
| main_module = importlib.import_module( |
| 'cobalt.black_box_tests.black_box_tests') |
| sys.exit(main_module.main()) |