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
|
# 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 subprocess
import tempfile
from mozprocess import ProcessHandler
from .base import Browser, require_arg, get_free_port, browser_command, ExecutorBrowser
from .servo import render_arg
from ..executors import executor_kwargs as base_executor_kwargs
from ..executors.executorservodriver import (ServoWebDriverTestharnessExecutor,
ServoWebDriverRefTestExecutor)
here = os.path.join(os.path.split(__file__)[0])
__wptrunner__ = {"product": "servodriver",
"check_args": "check_args",
"browser": "ServoWebDriverBrowser",
"executor": {"testharness": "ServoWebDriverTestharnessExecutor",
"reftest": "ServoWebDriverRefTestExecutor"},
"browser_kwargs": "browser_kwargs",
"executor_kwargs": "executor_kwargs",
"env_options": "env_options",
"run_info_extras": "run_info_extras",
"update_properties": "update_properties"}
hosts_text = """127.0.0.1 web-platform.test
127.0.0.1 www.web-platform.test
127.0.0.1 www1.web-platform.test
127.0.0.1 www2.web-platform.test
127.0.0.1 xn--n8j6ds53lwwkrqhv28a.web-platform.test
127.0.0.1 xn--lve-6lad.web-platform.test
"""
def check_args(**kwargs):
require_arg(kwargs, "binary")
def browser_kwargs(**kwargs):
return {"binary": kwargs["binary"],
"debug_info": kwargs["debug_info"],
"user_stylesheets": kwargs.get("user_stylesheets"),
"render_backend": kwargs.get("servo_backend")}
def executor_kwargs(test_type, server_config, cache_manager, run_info_data, **kwargs):
rv = base_executor_kwargs(test_type, server_config,
cache_manager, **kwargs)
return rv
def env_options():
return {"host": "127.0.0.1",
"external_host": "web-platform.test",
"bind_hostname": "true",
"testharnessreport": "testharnessreport-servodriver.js",
"supports_debugger": True}
def run_info_extras(**kwargs):
return {"backend": kwargs["servo_backend"]}
def update_properties():
return ["debug", "os", "version", "processor", "bits", "backend"], None
def make_hosts_file():
hosts_fd, hosts_path = tempfile.mkstemp()
with os.fdopen(hosts_fd, "w") as f:
f.write(hosts_text)
return hosts_path
class ServoWebDriverBrowser(Browser):
used_ports = set()
def __init__(self, logger, binary, debug_info=None, webdriver_host="127.0.0.1",
user_stylesheets=None, render_backend="cpu"):
Browser.__init__(self, logger)
self.binary = binary
self.webdriver_host = webdriver_host
self.webdriver_port = None
self.proc = None
self.debug_info = debug_info
self.hosts_path = make_hosts_file()
self.command = None
self.user_stylesheets = user_stylesheets if user_stylesheets else []
self.render_backend = render_backend
def start(self):
self.webdriver_port = get_free_port(4444, exclude=self.used_ports)
self.used_ports.add(self.webdriver_port)
env = os.environ.copy()
env["HOST_FILE"] = self.hosts_path
env["RUST_BACKTRACE"] = "1"
debug_args, command = browser_command(self.binary,
[render_arg(self.render_backend), "--hard-fail",
"--webdriver", str(self.webdriver_port),
"about:blank"],
self.debug_info)
for stylesheet in self.user_stylesheets:
command += ["--user-stylesheet", stylesheet]
self.command = command
self.command = debug_args + self.command
if not self.debug_info or not self.debug_info.interactive:
self.proc = ProcessHandler(self.command,
processOutputLine=[self.on_output],
env=env,
storeOutput=False)
self.proc.run()
else:
self.proc = subprocess.Popen(self.command, env=env)
self.logger.debug("Servo Started")
def stop(self):
self.logger.debug("Stopping browser")
if self.proc is not None:
try:
self.proc.kill()
except OSError:
# This can happen on Windows if the process is already dead
pass
def pid(self):
if self.proc is None:
return None
try:
return self.proc.pid
except AttributeError:
return None
def on_output(self, line):
"""Write a line of output from the process to the log"""
self.logger.process_output(self.pid(),
line.decode("utf8", "replace"),
command=" ".join(self.command))
def is_alive(self):
if self.runner:
return self.runner.is_running()
return False
def cleanup(self):
self.stop()
def executor_browser(self):
assert self.webdriver_port is not None
return ExecutorBrowser, {"webdriver_host": self.webdriver_host,
"webdriver_port": self.webdriver_port}
|
