wpt: Create a base class for Servo process executors
This change creates a base class for the Servo process executors, to
handle shared functionality. The only thing that hasn't moved there yet
is the actual process execution, which can happen in a followup change.
The main motivation behind this change is consistently handling
`on_evironment_change` which is used to handle changes to the `prefs`
value stored in `__dir__.ini`. Inherited `prefs` (thos in `__dir__.ini`)
aren't passed when creating tests for reference HTML. This change takes
a similar appraoch to Gecko, which just listens to
`on_environment_change` to note when these prefs change.
diff --git a/tools/wptrunner/wptrunner/executors/executorservo.py b/tools/wptrunner/wptrunner/executors/executorservo.py
index 3250c74..dd1ea1a 100644
--- a/tools/wptrunner/wptrunner/executors/executorservo.py
+++ b/tools/wptrunner/wptrunner/executors/executorservo.py
@@ -27,53 +27,24 @@
webdriver = None
-def write_hosts_file(config):
- hosts_fd, hosts_path = tempfile.mkstemp()
- with os.fdopen(hosts_fd, "w") as f:
- f.write(make_hosts_file(config, "127.0.0.1"))
- return hosts_path
-
-
-def build_servo_command(test, test_url_func, browser, binary, pause_after_test, debug_info,
- extra_args=None, debug_opts="replace-surrogates"):
- args = [
- "--hard-fail", "-u", "Servo/wptrunner",
- # See https://github.com/servo/servo/issues/30080.
- # For some reason rustls does not like the certificate generated by the WPT tooling.
- "--ignore-certificate-errors",
- "-z", test_url_func(test),
- ]
- if debug_opts:
- args += ["-Z", debug_opts]
- for stylesheet in browser.user_stylesheets:
- args += ["--user-stylesheet", stylesheet]
- for pref, value in test.environment.get('prefs', {}).items():
- args += ["--pref", f"{pref}={value}"]
- if browser.ca_certificate_path:
- args += ["--certificate-path", browser.ca_certificate_path]
- if extra_args:
- args += extra_args
- args += browser.binary_args
- debug_args, command = browser_command(binary, args, debug_info)
- if pause_after_test:
- command.remove("-z")
- return debug_args + command
-
-
-
-class ServoTestharnessExecutor(ProcessTestExecutor):
- convert_result = testharness_result_converter
-
- def __init__(self, logger, browser, server_config, timeout_multiplier=1, debug_info=None,
- pause_after_test=False, **kwargs):
+class ServoExecutor(ProcessTestExecutor):
+ def __init__(self, logger, browser, server_config, timeout_multiplier, debug_info,
+ pause_after_test, reftest_screenshot="unexpected"):
ProcessTestExecutor.__init__(self, logger, browser, server_config,
timeout_multiplier=timeout_multiplier,
- debug_info=debug_info)
+ debug_info=debug_info,
+ reftest_screenshot=reftest_screenshot)
self.pause_after_test = pause_after_test
- self.result_data = None
- self.result_flag = None
+ self.environment = {}
self.protocol = ConnectionlessProtocol(self, browser)
- self.hosts_path = write_hosts_file(server_config)
+
+ hosts_fd, self.hosts_path = tempfile.mkstemp()
+ with os.fdopen(hosts_fd, "w") as f:
+ f.write(make_hosts_file(server_config, "127.0.0.1"))
+
+ self.env_for_tests = os.environ.copy()
+ self.env_for_tests["HOST_FILE"] = self.hosts_path
+ self.env_for_tests["RUST_BACKTRACE"] = "1"
def teardown(self):
try:
@@ -82,32 +53,70 @@
pass
ProcessTestExecutor.teardown(self)
+ def on_environment_change(self, new_environment):
+ self.environment = new_environment
+ return super().on_environment_change(new_environment)
+
+ def on_output(self, line):
+ line = line.decode("utf8", "replace")
+ if self.interactive:
+ print(line)
+ else:
+ self.logger.process_output(self.proc.pid, line, " ".join(self.command), self.test.url)
+
+ def build_servo_command(self, test, extra_args=None, debug_opts="replace-surrogates"):
+ args = [
+ "--hard-fail", "-u", "Servo/wptrunner",
+ # See https://github.com/servo/servo/issues/30080.
+ # For some reason rustls does not like the certificate generated by the WPT tooling.
+ "--ignore-certificate-errors",
+ "-z", self.test_url(test),
+ ]
+ if debug_opts:
+ args += ["-Z", debug_opts]
+ for stylesheet in self.browser.user_stylesheets:
+ args += ["--user-stylesheet", stylesheet]
+ for pref, value in self.environment.get('prefs', {}).items():
+ args += ["--pref", f"{pref}={value}"]
+ if self.browser.ca_certificate_path:
+ args += ["--certificate-path", self.browser.ca_certificate_path]
+ if extra_args:
+ args += extra_args
+ args += self.browser.binary_args
+ debug_args, command = browser_command(self.binary, args, self.debug_info)
+ if self.pause_after_test:
+ command.remove("-z")
+ return debug_args + command
+
+
+class ServoTestharnessExecutor(ServoExecutor):
+ convert_result = testharness_result_converter
+
+ def __init__(self, logger, browser, server_config, timeout_multiplier=1, debug_info=None,
+ pause_after_test=False, **kwargs):
+ ServoExecutor.__init__(self, logger, browser, server_config,
+ timeout_multiplier=timeout_multiplier,
+ debug_info=debug_info,
+ pause_after_test=pause_after_test)
+ self.result_data = None
+ self.result_flag = None
+
def do_test(self, test):
self.test = test
self.result_data = None
self.result_flag = threading.Event()
- self.command = build_servo_command(test,
- self.test_url,
- self.browser,
- self.binary,
- self.pause_after_test,
- self.debug_info)
-
- env = os.environ.copy()
- env["HOST_FILE"] = self.hosts_path
- env["RUST_BACKTRACE"] = "1"
-
+ self.command = self.build_servo_command(test)
if not self.interactive:
self.proc = ProcessHandler(self.command,
processOutputLine=[self.on_output],
onFinish=self.on_finish,
- env=env,
+ env=self.env_for_tests,
storeOutput=False)
self.proc.run()
else:
- self.proc = subprocess.Popen(self.command, env=env)
+ self.proc = subprocess.Popen(self.command, env=self.env_for_tests)
try:
timeout = test.timeout * self.timeout_multiplier
@@ -132,7 +141,6 @@
else:
result = (test.result_cls("TIMEOUT", None), [])
-
if proc_is_running:
if self.pause_after_test:
self.logger.info("Pausing until the browser exits")
@@ -147,17 +155,12 @@
def on_output(self, line):
prefix = "ALERT: RESULT: "
- line = line.decode("utf8", "replace")
- if line.startswith(prefix):
- self.result_data = json.loads(line[len(prefix):])
+ decoded_line = line.decode("utf8", "replace")
+ if decoded_line.startswith(prefix):
+ self.result_data = json.loads(decoded_line[len(prefix):])
self.result_flag.set()
else:
- if self.interactive:
- print(line)
- else:
- self.logger.process_output(self.proc.pid,
- line,
- " ".join(self.command), self.test.url)
+ ServoExecutor.on_output(self, line)
def on_finish(self):
self.result_flag.set()
@@ -179,37 +182,32 @@
pass
-class ServoRefTestExecutor(ProcessTestExecutor):
+class ServoRefTestExecutor(ServoExecutor):
convert_result = reftest_result_converter
def __init__(self, logger, browser, server_config, binary=None, timeout_multiplier=1,
screenshot_cache=None, debug_info=None, pause_after_test=False,
reftest_screenshot="unexpected", **kwargs):
- ProcessTestExecutor.__init__(self,
- logger,
- browser,
- server_config,
- timeout_multiplier=timeout_multiplier,
- debug_info=debug_info,
- reftest_screenshot=reftest_screenshot)
+ ServoExecutor.__init__(self,
+ logger,
+ browser,
+ server_config,
+ timeout_multiplier=timeout_multiplier,
+ debug_info=debug_info,
+ reftest_screenshot=reftest_screenshot,
+ pause_after_test=pause_after_test)
- self.protocol = ConnectionlessProtocol(self, browser)
self.screenshot_cache = screenshot_cache
self.reftest_screenshot = reftest_screenshot
self.implementation = RefTestImplementation(self)
self.tempdir = tempfile.mkdtemp()
- self.hosts_path = write_hosts_file(server_config)
def reset(self):
self.implementation.reset()
def teardown(self):
- try:
- os.unlink(self.hosts_path)
- except OSError:
- pass
os.rmdir(self.tempdir)
- ProcessTestExecutor.teardown(self)
+ ServoExecutor.teardown(self)
def screenshot(self, test, viewport_size, dpi, page_ranges):
with TempFilename(self.tempdir) as output_path:
@@ -221,24 +219,12 @@
if dpi:
extra_args += ["--device-pixel-ratio", dpi]
- self.command = build_servo_command(test,
- self.test_url,
- self.browser,
- self.binary,
- False,
- self.debug_info,
- extra_args,
- debug_opts)
-
- env = os.environ.copy()
- env["HOST_FILE"] = self.hosts_path
- env["RUST_BACKTRACE"] = "1"
+ self.command = self.build_servo_command(test, extra_args, debug_opts)
if not self.interactive:
self.proc = ProcessHandler(self.command,
processOutputLine=[self.on_output],
- env=env)
-
+ env=self.env_for_tests)
try:
self.proc.run()
@@ -248,8 +234,7 @@
self.proc.kill()
raise
else:
- self.proc = subprocess.Popen(self.command,
- env=env)
+ self.proc = subprocess.Popen(self.command, env=self.env_for_tests)
try:
rv = self.proc.wait()
except KeyboardInterrupt:
@@ -276,20 +261,11 @@
return self.convert_result(test, result)
- def on_output(self, line):
- line = line.decode("utf8", "replace")
- if self.interactive:
- print(line)
- else:
- self.logger.process_output(self.proc.pid,
- line,
- " ".join(self.command), self.test.url)
-
class ServoTimedRunner(TimedRunner):
def run_func(self):
try:
- self.result = True, self.func(self.protocol, self.url, self.timeout)
+ self.result = (True, self.func(self.protocol, self.url, self.timeout))
except Exception as e:
message = getattr(e, "message", "")
if message:
@@ -303,23 +279,22 @@
pass
-class ServoCrashtestExecutor(ProcessTestExecutor):
+class ServoCrashtestExecutor(ServoExecutor):
convert_result = crashtest_result_converter
def __init__(self, logger, browser, server_config, binary=None, timeout_multiplier=1,
screenshot_cache=None, debug_info=None, pause_after_test=False,
**kwargs):
- ProcessTestExecutor.__init__(self,
- logger,
- browser,
- server_config,
- timeout_multiplier=timeout_multiplier,
- debug_info=debug_info)
+ ServoExecutor.__init__(self,
+ logger,
+ browser,
+ server_config,
+ timeout_multiplier=timeout_multiplier,
+ debug_info=debug_info,
+ pause_after_test=pause_after_test)
self.pause_after_test = pause_after_test
self.protocol = ConnectionlessProtocol(self, browser)
- self.tempdir = tempfile.mkdtemp()
- self.hosts_path = write_hosts_file(server_config)
def do_test(self, test):
timeout = (test.timeout * self.timeout_multiplier if self.debug_info is None
@@ -340,36 +315,19 @@
return (test.result_cls(*data), [])
def do_crashtest(self, protocol, url, timeout):
- env = os.environ.copy()
- env["HOST_FILE"] = self.hosts_path
- env["RUST_BACKTRACE"] = "1"
-
- self.command = build_servo_command(self.test,
- self.test_url,
- self.browser,
- self.binary,
- False,
- self.debug_info,
- extra_args=["-x"])
+ self.command = self.build_servo_command(self.test, extra_args=["-x"])
if not self.interactive:
self.proc = ProcessHandler(self.command,
- env=env,
+ env=self.env_for_tests,
processOutputLine=[self.on_output],
storeOutput=False)
self.proc.run()
else:
- self.proc = subprocess.Popen(self.command, env=env)
+ self.proc = subprocess.Popen(self.command, env=self.env_for_tests)
self.proc.wait()
if self.proc.poll() >= 0:
return {"status": "PASS", "message": None}
-
return {"status": "CRASH", "message": None}
-
- def on_output(self, line):
- line = line.decode("utf8", "replace")
- self.logger.process_output(self.proc.pid,
- line,
- " ".join(self.command), self.test.url)
