blob: 2e86d0fe8231d6c351916e6c5b0d1439d07b60db [file] [log] [blame]
#!/usr/bin/env python3
# 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.
"""Contains utilities to create black box tests and run them."""
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
from cobalt.black_box_tests import black_box_cobalt_runner
from cobalt.black_box_tests.proxy_server import ProxyServer
from starboard.tools import abstract_launcher
from starboard.tools import build
from starboard.tools import command_line
from starboard.tools import log_level
_DISABLED_BLACKBOXTEST_CONFIGS = [
'android-arm/devel',
'android-arm64/devel',
'android-x86/devel',
'evergreen-arm/devel',
'evergreen-x64/devel',
'raspi-0/devel',
]
_EVERGREEN_COMPATIBLE_CONFIGS = [
# TODO(b/283788059): enable when there are GitHub jobs to run integration
# and Black Box Tests on evergreen-arm-hardfp.
#'evergreen-arm/devel',
# TODO(b/283144901): enable when the Starboard 16 binaries are released for
# Evergreen.
#'evergreen-x64/devel',
]
_PORT_SELECTION_RETRY_LIMIT = 10
_PORT_SELECTION_RANGE = [5000, 7000]
# List of blocked ports.
_RESTRICTED_PORTS = [6000, 6665, 6666, 6667, 6668, 6669, 6697]
_SERVER_EXIT_TIMEOUT_SECONDS = 30
_SOCKET_SUCCESS = 0
# 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',
'pointer_test',
'preload_font',
'preload_visibility',
'preload_launch_parameter',
# TODO(b/254502632): Investigate the cause of the flakiness from this test.
# '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',
'compression_test',
'default_site_can_load',
'disable_eval_with_csp',
'h5vcc_storage_write_verify_test',
'http_cache',
'persistent_cookie',
'scroll',
'service_worker_add_to_cache_test',
'service_worker_cache_keys_test',
'service_worker_controller_activation_test',
'service_worker_get_registrations_test',
'service_worker_fetch_main_resource_test',
'service_worker_fetch_test',
'service_worker_message_test',
'service_worker_post_message_test',
'service_worker_test',
'service_worker_persist_test',
'soft_mic_platform_service_test',
'telemetry_test',
'text_encoding_test',
'wasm_basic_test',
'web_debugger',
'web_worker_test',
'worker_csp_test',
'worker_load_test',
'worker_post_message_test',
]
# These are very different and require a custom config + proxy
_WPT_TESTS = [
'web_platform_tests',
]
# These tests can only be run on platforms whose app launcher can send deep
# links.
_TESTS_NEEDING_DEEP_LINK = [
'deep_links',
]
# These tests can only run on Evergreen-compatible platforms.
_TESTS_EVERGREEN_END_TO_END = [
'evergreen_verify_qa_channel_update_test',
]
# Location of test files.
_TEST_DIR_PATH = 'cobalt.black_box_tests.tests.'
_LAUNCH_TARGET = 'cobalt'
# Platform configuration and device information parameters.
_launcher_params = None
# Binding address used to create the test server.
_server_binding_address = None
# Port used to create the web platform test http server.
_wpt_http_port = None
class BlackBoxTestCase(unittest.TestCase):
"""Base class for Cobalt black box test cases."""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.launcher_params = _launcher_params
self.platform_config = build.GetPlatformConfig(_launcher_params.platform)
self.cobalt_config = self.platform_config.GetApplicationConfiguration(
'cobalt')
@classmethod
def setUpClass(cls):
super(BlackBoxTestCase, cls).setUpClass()
logging.info('\n\n\n%s\n\n', '=' * 40)
logging.info('Running %s', cls.__name__)
@classmethod
def tearDownClass(cls):
super(BlackBoxTestCase, cls).tearDownClass()
logging.info('Done %s', cls.__name__)
def CreateCobaltRunner(self, url=None, target_params=None, **kwargs):
all_target_params = list(target_params) if target_params else []
if _launcher_params.target_params is not None:
all_target_params += _launcher_params.target_params
return black_box_cobalt_runner.BlackBoxCobaltRunner(
launcher_params=_launcher_params,
url=url,
target_params=all_target_params,
**kwargs)
def GetBindingAddress(self):
return _server_binding_address
def GetWptHttpPort(self):
return _wpt_http_port
def LoadTests(launcher_params, test_set):
launcher = abstract_launcher.LauncherFactory(
launcher_params.platform,
_LAUNCH_TARGET,
launcher_params.config,
device_id=launcher_params.device_id,
target_params=None,
output_file=None,
out_directory=launcher_params.out_directory,
loader_platform=launcher_params.loader_platform,
loader_config=launcher_params.loader_config,
loader_out_directory=launcher_params.loader_out_directory)
test_targets = []
if test_set in ['all', 'blackbox']:
test_targets = _TESTS_NO_SIGNAL
if launcher.SupportsSuspendResume():
test_targets += _TESTS_NEEDING_SYSTEM_SIGNAL
if launcher.SupportsDeepLink():
test_targets += _TESTS_NEEDING_DEEP_LINK
if test_set in ['all', 'wpt']:
test_targets += _WPT_TESTS
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
def LoadEvergreenEndToEndTests(launcher_params):
launcher = abstract_launcher.LauncherFactory( # pylint: disable=unused-variable
launcher_params.platform,
_LAUNCH_TARGET,
launcher_params.config,
device_id=launcher_params.device_id,
target_params=None,
output_file=None,
out_directory=launcher_params.out_directory,
loader_platform=launcher_params.loader_platform,
loader_config=launcher_params.loader_config,
loader_out_directory=launcher_params.loader_out_directory,
# The more lightweight elf_loader_sandbox can't be used since it has no
# knowledge of updates or installations.
loader_target='loader_app')
test_targets = _TESTS_EVERGREEN_END_TO_END
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, args):
self.args = args
#TODO(b/137905502): These globals should be refactored
# Setup global variables used by test cases.
global _launcher_params
_launcher_params = command_line.CreateLauncherParams()
# Keep other modules from seeing these args.
sys.argv = sys.argv[:1]
global _server_binding_address
_server_binding_address = args.server_binding_address
# Port used to create the web platform test http server. If not specified,
# a random free port is used.
global _wpt_http_port
_wpt_http_port = args.wpt_http_port or str(
self.GetUnusedPort([_server_binding_address]))
# Proxy is only needed for WPT
self.use_proxy = args.test_set in ['all', 'wpt']
# TODO: Remove generation of --dev_servers_listen_ip once executable will
# be able to bind correctly with incomplete support of IPv6
if args.device_id and IsValidIpAddress(args.device_id):
_launcher_params.target_params.append(
f'--dev_servers_listen_ip={args.device_id}')
elif IsValidIpAddress(_server_binding_address):
_launcher_params.target_params.append(
f'--dev_servers_listen_ip={_server_binding_address}')
_launcher_params.target_params.append(
f'--web-platform-test-server=http://web-platform.test:{_wpt_http_port}')
# Port used to create the proxy server. If not specified, a random free
# port is used.
if self.use_proxy:
self.proxy_port = args.proxy_port or str(
self.GetUnusedPort([_server_binding_address]))
proxy_address = args.proxy_address or _server_binding_address
_launcher_params.target_params.append(
f'--proxy={proxy_address}:{self.proxy_port}')
self.device_ips = args.device_ips
# 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 = {host: _server_binding_address for host in hosts}
def Run(self):
if self.use_proxy and self.proxy_port == '-1':
return 1
run_cobalt_tests = True
run_evergreen_tests = False
launch_config = f'{_launcher_params.platform}/{_launcher_params.config}'
# TODO(b/135549281): Configuring this in Python is superfluous, the on/off
# flags can be in Github Actions code
if launch_config in _DISABLED_BLACKBOXTEST_CONFIGS:
run_cobalt_tests = False
logging.warning(
'Cobalt blackbox tests disabled for platform:%s config:%s',
_launcher_params.platform, _launcher_params.config)
if launch_config in _EVERGREEN_COMPATIBLE_CONFIGS:
run_evergreen_tests = self.args.test_set in ['all', 'evergreen']
if not (run_cobalt_tests or run_evergreen_tests):
return 0
def LoadAndRunTests():
if self.args.test_name:
suite = unittest.TestLoader().loadTestsFromName(_TEST_DIR_PATH +
self.args.test_name)
return_code = not unittest.TextTestRunner(
verbosity=2, stream=sys.stdout).run(suite).wasSuccessful()
return return_code
else:
cobalt_tests_return_code = 0
if run_cobalt_tests:
suite = LoadTests(_launcher_params, self.args.test_set)
# Using verbosity=2 to log individual test function names and results.
cobalt_tests_return_code = not unittest.TextTestRunner(
verbosity=2, stream=sys.stdout).run(suite).wasSuccessful()
evergreen_tests_return_code = 0
if run_evergreen_tests:
suite = LoadEvergreenEndToEndTests(_launcher_params)
evergreen_tests_return_code = not unittest.TextTestRunner(
verbosity=2, stream=sys.stdout).run(suite).wasSuccessful()
return cobalt_tests_return_code or evergreen_tests_return_code
if self.use_proxy:
logging.info('Using proxy port: %s', self.proxy_port)
with ProxyServer(
port=self.proxy_port,
host_resolve_map=self.host_resolve_map,
client_ips=self.args.device_ips):
return LoadAndRunTests()
else:
return LoadAndRunTests()
def GetUnusedPort(self, addresses):
"""Find a free port on the list of addresses by pinging with sockets."""
if not addresses:
logging.error('Can not find unused port on invalid addresses.')
return -1
socks = []
for address in addresses:
socks.append((address, socket.socket(socket.AF_INET, socket.SOCK_STREAM)))
try:
for _ in range(_PORT_SELECTION_RETRY_LIMIT):
port = random.choice([
i for i in range(_PORT_SELECTION_RANGE[0], _PORT_SELECTION_RANGE[1])
if i not in _RESTRICTED_PORTS
])
unused = True
for sock in socks:
result = sock[1].connect_ex((sock[0], port))
if result == _SOCKET_SUCCESS:
unused = False
break
if unused:
return port
logging.error('Can not find unused port on addresses within %s attempts.',
_PORT_SELECTION_RETRY_LIMIT)
return -1
finally:
for sock in socks:
sock[1].close()
def IsValidIpAddress(address):
"""Checks if address is valid IP address."""
return IsValidIpv4Address(address) or IsValidIpv6Address(address)
def IsValidIpv4Address(address):
"""Checks if address is valid IPv4 address."""
try:
socket.inet_pton(socket.AF_INET, address)
return True
except socket.error: # not a valid address
return False
def IsValidIpv6Address(address):
"""Checks if address is valid IPv6 address."""
try:
socket.inet_pton(socket.AF_INET6, address)
return True
except socket.error: # not a valid address
return False
def main():
parser = argparse.ArgumentParser()
parser.add_argument('-v', '--verbose', required=False, action='store_true')
parser.add_argument(
'--server_binding_address',
default='127.0.0.1',
help='Binding address used to create the test server.')
parser.add_argument(
'--proxy_address',
default=None,
help=('Address to the proxy server that all black box'
'tests are run through. If not specified, the'
'server binding address is used.'))
parser.add_argument(
'--proxy_port',
default=None,
help=('Port used to create the proxy server that all'
'black box tests are run through. If not'
'specified, a random free port is used.'))
parser.add_argument(
'--test_name',
default=None,
help=('Name of test to be run. If not specified, all '
'tests are run.'))
parser.add_argument(
'--wpt_http_port',
default=None,
help=('Port used to create the web platform test http'
'server. If not specified, a random free port is'
'used.'))
parser.add_argument(
'--device_id',
default=None,
help=('ID of test device to connect. If specified, it will be passed '
'as --dev_servers_listen_ip param on the test device.'))
parser.add_argument(
'--device_ips',
default=None,
nargs='*',
help=('IPs of test devices that will be allowed to connect. If not '
'specified, all IPs will be allowed to connect.'))
parser.add_argument(
'--test_set',
choices=['all', 'wpt', 'blackbox', 'evergreen'],
default='all')
args, _ = parser.parse_known_args()
log_level.InitializeLogging(args)
test_object = BlackBoxTests(args)
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 launcher_param accessible to the tests.
main_module = importlib.import_module(
'cobalt.black_box_tests.black_box_tests')
sys.exit(main_module.main())