Source code
Revision control
Copy as Markdown
Other Tools
# mypy: allow-untyped-defs
import mozprocess
import subprocess
from .base import OutputHandler, cmd_arg, require_arg
from .base import get_timeout_multiplier # noqa: F401
from .base import WebDriverBrowser # noqa: F401
from .chrome import executor_kwargs as chrome_executor_kwargs
from ..executors.base import WdspecExecutor # noqa: F401
from ..executors.executorchrome import ChromeDriverPrintRefTestExecutor # noqa: F401
from ..executors.executorwebdriver import (WebDriverCrashtestExecutor, # noqa: F401
WebDriverTestharnessExecutor, # noqa: F401
WebDriverRefTestExecutor) # noqa: F401
__wptrunner__ = {"product": "chrome_android",
"check_args": "check_args",
"browser": "ChromeAndroidBrowser",
"executor": {"testharness": "WebDriverTestharnessExecutor",
"reftest": "WebDriverRefTestExecutor",
"print-reftest": "ChromeDriverPrintRefTestExecutor",
"wdspec": "WdspecExecutor",
"crashtest": "WebDriverCrashtestExecutor"},
"browser_kwargs": "browser_kwargs",
"executor_kwargs": "executor_kwargs",
"env_extras": "env_extras",
"env_options": "env_options",
"timeout_multiplier": "get_timeout_multiplier"}
_wptserve_ports = set()
def check_args(**kwargs):
require_arg(kwargs, "package_name")
require_arg(kwargs, "webdriver_binary")
def browser_kwargs(logger, test_type, run_info_data, config, **kwargs):
return {"package_name": kwargs["package_name"],
"adb_binary": kwargs["adb_binary"],
"device_serial": kwargs["device_serial"],
"webdriver_binary": kwargs["webdriver_binary"],
"webdriver_args": kwargs.get("webdriver_args"),
"stackwalk_binary": kwargs.get("stackwalk_binary"),
"symbols_path": kwargs.get("symbols_path")}
def executor_kwargs(logger, test_type, test_environment, run_info_data,
**kwargs):
# Use update() to modify the global list in place.
_wptserve_ports.update(set(
test_environment.config['ports']['http'] + test_environment.config['ports']['https'] +
test_environment.config['ports']['ws'] + test_environment.config['ports']['wss']
))
executor_kwargs = chrome_executor_kwargs(logger, test_type, test_environment, run_info_data,
**kwargs)
# Remove unsupported options on mobile.
del executor_kwargs["capabilities"]["goog:chromeOptions"]["prefs"]
assert kwargs["package_name"], "missing --package-name"
capabilities = executor_kwargs["capabilities"]
capabilities["goog:chromeOptions"]["androidPackage"] = \
kwargs["package_name"]
capabilities["goog:chromeOptions"]["androidKeepAppDataDir"] = \
kwargs.get("keep_app_data_directory")
return executor_kwargs
def env_extras(**kwargs):
return []
def env_options():
# allow the use of host-resolver-rules in lieu of modifying /etc/hosts file
return {"server_host": "127.0.0.1"}
class LogcatRunner:
def __init__(self, logger, browser):
self.logger = logger
self.browser = browser
def start(self):
try:
self._run()
except KeyboardInterrupt:
self.stop()
def _run(self):
try:
# TODO: adb logcat -c fail randomly with message
# "failed to clear the 'main' log"
self.browser.clear_log()
except subprocess.CalledProcessError:
self.logger.error("Failed to clear logcat buffer")
self._cmd = self.browser.logcat_cmd()
self._output_handler = OutputHandler(self.logger, self._cmd)
self._proc = mozprocess.ProcessHandler(
self._cmd,
processOutputLine=self._output_handler,
storeOutput=False)
self._proc.run()
self._output_handler.after_process_start(self._proc.pid)
self._output_handler.start()
def stop(self, force=False):
if self.is_alive():
kill_result = self._proc.kill()
if force and kill_result != 0:
self._proc.kill(9)
self._output_handler.after_process_stop()
def is_alive(self):
return hasattr(self._proc, "proc") and self._proc.poll() is None
class ChromeAndroidBrowserBase(WebDriverBrowser):
def __init__(self,
logger,
*,
adb_binary=None,
device_serial=None,
stackwalk_binary=None,
symbols_path=None,
**kwargs):
super().__init__(logger, **kwargs)
self.adb_binary = adb_binary or "adb"
self.device_serial = device_serial[self.manager_number]
self.stackwalk_binary = stackwalk_binary
self.symbols_path = symbols_path
self.logcat_runner = LogcatRunner(self.logger, self)
def setup(self):
self.setup_adb_reverse()
self.logcat_runner.start()
def _adb_run(self, args):
cmd = [self.adb_binary]
if self.device_serial:
cmd.extend(['-s', self.device_serial])
cmd.extend(args)
self.logger.info(' '.join(cmd))
subprocess.check_call(cmd)
def make_command(self):
return [self.webdriver_binary,
cmd_arg("port", str(self.port)),
cmd_arg("url-base", self.base_path),
cmd_arg("enable-chrome-logs")] + self.webdriver_args
def cleanup(self):
super().cleanup()
self._adb_run(['forward', '--remove-all'])
self._adb_run(['reverse', '--remove-all'])
self.logcat_runner.stop(force=True)
def executor_browser(self):
cls, kwargs = super().executor_browser()
kwargs["capabilities"] = {
"goog:chromeOptions": {
"androidDeviceSerial": self.device_serial
}
}
return cls, kwargs
def clear_log(self):
self._adb_run(['logcat', '-c'])
def logcat_cmd(self):
cmd = [self.adb_binary]
if self.device_serial:
cmd.extend(['-s', self.device_serial])
cmd.extend(['logcat', '*:D'])
return cmd
def check_crash(self, process, test):
self.maybe_parse_tombstone()
# Existence of a tombstone does not necessarily mean test target has
# crashed. Always return False so we don't change the test results.
return False
def maybe_parse_tombstone(self):
if self.stackwalk_binary:
cmd = [self.stackwalk_binary, "-a", "-w"]
if self.device_serial:
cmd.extend(["--device", self.device_serial])
cmd.extend(["--output-directory", self.symbols_path])
raw_output = subprocess.check_output(cmd)
for line in raw_output.splitlines():
self.logger.process_output("TRACE", line, "logcat")
def setup_adb_reverse(self):
self._adb_run(['wait-for-device'])
self._adb_run(['forward', '--remove-all'])
self._adb_run(['reverse', '--remove-all'])
# "adb reverse" forwards network connection from device to host.
for port in self.wptserver_ports:
self._adb_run(['reverse', 'tcp:%d' % port, 'tcp:%d' % port])
class ChromeAndroidBrowser(ChromeAndroidBrowserBase):
"""Chrome is backed by chromedriver, which is supplied through
``wptrunner.webdriver.ChromeDriverServer``.
"""
def __init__(self, logger, *, package_name, **kwargs):
super().__init__(logger, **kwargs)
self.package_name = package_name
self.wptserver_ports = _wptserve_ports