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
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
|
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this file,
# You can obtain one at http://mozilla.org/MPL/2.0/.
import os
import tempfile
import shutil
import subprocess
import fxos_appgen
import gaiatest
import mozdevice
import moznetwork
import mozrunner
from marionette import expected
from marionette.by import By
from marionette.wait import Wait
from mozprofile import FirefoxProfile, Preferences
from .base import get_free_port, BrowserError, Browser, ExecutorBrowser
from ..executors.executormarionette import MarionetteTestharnessExecutor
from ..hosts import HostsFile, HostsLine
from ..environment import hostnames
here = os.path.split(__file__)[0]
__wptrunner__ = {"product": "b2g",
"check_args": "check_args",
"browser": "B2GBrowser",
"executor": {"testharness": "B2GMarionetteTestharnessExecutor"},
"browser_kwargs": "browser_kwargs",
"executor_kwargs": "executor_kwargs",
"env_options": "env_options"}
def check_args(**kwargs):
pass
def browser_kwargs(test_environment, **kwargs):
return {"prefs_root": kwargs["prefs_root"],
"no_backup": kwargs.get("b2g_no_backup", False)}
def executor_kwargs(test_type, server_config, cache_manager, run_info_data,
**kwargs):
timeout_multiplier = kwargs["timeout_multiplier"]
if timeout_multiplier is None:
timeout_multiplier = 2
executor_kwargs = {"server_config": server_config,
"timeout_multiplier": timeout_multiplier,
"close_after_done": False}
if test_type == "reftest":
executor_kwargs["cache_manager"] = cache_manager
return executor_kwargs
def env_options():
return {"host": "web-platform.test",
"bind_hostname": "false",
"test_server_port": False}
class B2GBrowser(Browser):
used_ports = set()
init_timeout = 180
def __init__(self, logger, prefs_root, no_backup=False):
Browser.__init__(self, logger)
logger.info("Waiting for device")
subprocess.call(["adb", "wait-for-device"])
self.device = mozdevice.DeviceManagerADB()
self.marionette_port = get_free_port(2828, exclude=self.used_ports)
self.used_ports.add(self.marionette_port)
self.cert_test_app = None
self.runner = None
self.prefs_root = prefs_root
self.no_backup = no_backup
self.backup_path = None
self.backup_paths = []
self.backup_dirs = []
def setup(self):
self.logger.info("Running B2G setup")
self.backup_path = tempfile.mkdtemp()
self.logger.debug("Backing up device to %s" % (self.backup_path,))
if not self.no_backup:
self.backup_dirs = [("/data/local", os.path.join(self.backup_path, "local")),
("/data/b2g/mozilla", os.path.join(self.backup_path, "profile"))]
self.backup_paths = [("/system/etc/hosts", os.path.join(self.backup_path, "hosts"))]
for remote, local in self.backup_dirs:
self.device.getDirectory(remote, local)
for remote, local in self.backup_paths:
self.device.getFile(remote, local)
self.setup_hosts()
def start(self):
profile = FirefoxProfile()
profile.set_preferences({"dom.disable_open_during_load": False,
"marionette.defaultPrefs.enabled": True})
self.logger.debug("Creating device runner")
self.runner = mozrunner.B2GDeviceRunner(profile=profile)
self.logger.debug("Starting device runner")
self.runner.start()
self.logger.debug("Device runner started")
def setup_hosts(self):
host_ip = moznetwork.get_ip()
temp_dir = tempfile.mkdtemp()
hosts_path = os.path.join(temp_dir, "hosts")
remote_path = "/system/etc/hosts"
try:
self.device.getFile("/system/etc/hosts", hosts_path)
with open(hosts_path) as f:
hosts_file = HostsFile.from_file(f)
for canonical_hostname in hostnames:
hosts_file.set_host(HostsLine(host_ip, canonical_hostname))
with open(hosts_path, "w") as f:
hosts_file.to_file(f)
self.logger.info("Installing hosts file")
self.device.remount()
self.device.removeFile(remote_path)
self.device.pushFile(hosts_path, remote_path)
finally:
os.unlink(hosts_path)
os.rmdir(temp_dir)
def load_prefs(self):
prefs_path = os.path.join(self.prefs_root, "prefs_general.js")
if os.path.exists(prefs_path):
preferences = Preferences.read_prefs(prefs_path)
else:
self.logger.warning("Failed to find base prefs file in %s" % prefs_path)
preferences = []
return preferences
def stop(self):
pass
def on_output(self):
raise NotImplementedError
def cleanup(self):
self.logger.debug("Running browser cleanup steps")
self.device.remount()
for remote, local in self.backup_dirs:
self.device.removeDir(remote)
self.device.pushDir(local, remote)
for remote, local in self.backup_paths:
self.device.removeFile(remote)
self.device.pushFile(local, remote)
shutil.rmtree(self.backup_path)
self.device.reboot(wait=True)
def pid(self):
return None
def is_alive(self):
return True
def executor_browser(self):
return B2GExecutorBrowser, {"marionette_port": self.marionette_port}
class B2GExecutorBrowser(ExecutorBrowser):
# The following methods are called from a different process
def __init__(self, *args, **kwargs):
ExecutorBrowser.__init__(self, *args, **kwargs)
import sys, subprocess
self.device = mozdevice.ADBB2G()
self.device.forward("tcp:%s" % self.marionette_port,
"tcp:2828")
self.executor = None
self.marionette = None
self.gaia_device = None
self.gaia_apps = None
def after_connect(self, executor):
self.executor = executor
self.marionette = executor.marionette
self.executor.logger.debug("Running browser.after_connect steps")
self.gaia_apps = gaiatest.GaiaApps(marionette=executor.marionette)
self.executor.logger.debug("Waiting for homescreen to load")
# Moved out of gaia_test temporarily
self.executor.logger.info("Waiting for B2G to be ready")
self.wait_for_homescreen(timeout=60)
self.install_cert_app()
self.use_cert_app()
def install_cert_app(self):
"""Install the container app used to run the tests"""
if fxos_appgen.is_installed("CertTest App"):
self.executor.logger.info("CertTest App is already installed")
return
self.executor.logger.info("Installing CertTest App")
app_path = os.path.join(here, "b2g_setup", "certtest_app.zip")
fxos_appgen.install_app("CertTest App", app_path, marionette=self.marionette)
self.executor.logger.debug("Install complete")
def use_cert_app(self):
"""Start the app used to run the tests"""
self.executor.logger.info("Homescreen loaded")
self.gaia_apps.launch("CertTest App")
def wait_for_homescreen(self, timeout):
self.executor.logger.info("Waiting for home screen to load")
Wait(self.marionette, timeout).until(expected.element_present(
By.CSS_SELECTOR, '#homescreen[loading-state=false]'))
class B2GMarionetteTestharnessExecutor(MarionetteTestharnessExecutor):
def after_connect(self):
self.browser.after_connect(self)
MarionetteTestharnessExecutor.after_connect(self)
|
