-
Notifications
You must be signed in to change notification settings - Fork 3k
/
android_weblayer.py
134 lines (107 loc) · 5.1 KB
/
android_weblayer.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
import subprocess
from .base import Browser, ExecutorBrowser, require_arg
from .base import get_timeout_multiplier # noqa: F401
from .chrome import executor_kwargs as chrome_executor_kwargs
from ..webdriver_server import ChromeDriverServer
from ..executors.executorwebdriver import (WebDriverTestharnessExecutor, # noqa: F401
WebDriverRefTestExecutor) # noqa: F401
from ..executors.executorchrome import ChromeDriverWdspecExecutor # noqa: F401
__wptrunner__ = {"product": "android_weblayer",
"check_args": "check_args",
"browser": "WeblayerShell",
"executor": {"testharness": "WebDriverTestharnessExecutor",
"reftest": "WebDriverRefTestExecutor",
"wdspec": "ChromeDriverWdspecExecutor"},
"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, "webdriver_binary")
def browser_kwargs(test_type, run_info_data, config, **kwargs):
return {"binary": kwargs["binary"],
"device_serial": kwargs["device_serial"],
"webdriver_binary": kwargs["webdriver_binary"],
"webdriver_args": kwargs.get("webdriver_args")}
def executor_kwargs(test_type, server_config, cache_manager, run_info_data,
**kwargs):
# Use update() to modify the global list in place.
_wptserve_ports.update(set(
server_config['ports']['http'] + server_config['ports']['https'] +
server_config['ports']['ws'] + server_config['ports']['wss']
))
executor_kwargs = chrome_executor_kwargs(test_type, server_config,
cache_manager, run_info_data,
**kwargs)
del executor_kwargs["capabilities"]["goog:chromeOptions"]["prefs"]
capabilities = executor_kwargs["capabilities"]
# Note that for WebLayer, we launch a test shell and have the test shell use
# WebLayer.
# https://cs.chromium.org/chromium/src/weblayer/shell/android/shell_apk/
capabilities["goog:chromeOptions"]["androidPackage"] = \
"org.chromium.weblayer.shell"
capabilities["goog:chromeOptions"]["androidActivity"] = ".WebLayerShellActivity"
if kwargs.get('device_serial'):
capabilities["goog:chromeOptions"]["androidDeviceSerial"] = kwargs['device_serial']
# Workaround: driver.quit() cannot quit WeblayerShell.
executor_kwargs["pause_after_test"] = False
# Workaround: driver.close() is not supported.
executor_kwargs["restart_after_test"] = True
executor_kwargs["close_after_done"] = False
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"}
#TODO: refactor common elements of WeblayerShell and ChromeAndroidBrowser
class WeblayerShell(Browser):
"""Chrome is backed by chromedriver, which is supplied through
``wptrunner.webdriver.ChromeDriverServer``.
"""
def __init__(self, logger, binary, webdriver_binary="chromedriver",
device_serial=None,
webdriver_args=None):
"""Creates a new representation of Chrome. The `binary` argument gives
the browser binary to use for testing."""
Browser.__init__(self, logger)
self.binary = binary
self.device_serial = device_serial
self.server = ChromeDriverServer(self.logger,
binary=webdriver_binary,
args=webdriver_args)
self.setup_adb_reverse()
def _adb_run(self, args):
cmd = ['adb']
if self.device_serial:
cmd.extend(['-s', self.device_serial])
cmd.extend(args)
self.logger.info(' '.join(cmd))
subprocess.check_call(cmd)
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" basically forwards network connection from device to
# host.
for port in _wptserve_ports:
self._adb_run(['reverse', 'tcp:%d' % port, 'tcp:%d' % port])
def start(self, **kwargs):
self.server.start(block=False)
def stop(self, force=False):
self.server.stop(force=force)
def pid(self):
return self.server.pid
def is_alive(self):
# TODO(ato): This only indicates the driver is alive,
# and doesn't say anything about whether a browser session
# is active.
return self.server.is_alive()
def cleanup(self):
self.stop()
self._adb_run(['forward', '--remove-all'])
self._adb_run(['reverse', '--remove-all'])
def executor_browser(self):
return ExecutorBrowser, {"webdriver_url": self.server.url}