eliott/servo
1
0
Fork 0
mirror of https://github.com/servo/servo synced 2026-04-25 17:15:48 +02:00
Code Issues Packages Projects Releases Wiki Activity

wpt: Use Webdriver for all WPT runs (#41511)

Browse Source 
All other browsers use a single configuration for their browser
invocations on WPT. Servo historically had two: servo and servodriver.
Now that we run WPT on Servo GitHub CI with Webdriver using the
servodriver, we can align our configuration with theirs.

The existing "servo" configuration is renamed to "servo_legacy" and
"servodriver" is then renamed to "servo". This way, we preserve the
"servo" product name as defined on wpt.fyi, but we do use its webdriver
configuration.

Since webdriver is not fully working yet for debugging purposes, we keep
the "servo_legacy" configuration now. In the future, once the debugging
story has improved, we can remove "servo_legacy".

All in all, this ensures that both on local, Servo GitHub CI and on
wpt.fyi we all use the exact same configuration. I tested this locally
by running the following test:

```
./mach test-wpt tests/wpt/tests/css/css-overflow/scrollbar-gutter-dynamic-004.html
```

This does times out with the servo binary and works with the servodriver
binary.

Running the servo_legacy configuration is done via the `--servo-legacy`
flag:

```
./mach test-wpt tests/wpt/mozilla/tests/mozilla/caption.html --servo-legacy
```

Fixes #40751

---------

Signed-off-by: Tim van der Lippe <tvanderlippe@gmail.com>
This commit is contained in:
Tim van der Lippe
2026-01-29 21:47:03 +01:00
committed by GitHub
parent 2c770c162c
commit 1133eb229a
14 changed files with 797 additions and 778 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
python/wpt/__init__.py
View File

@@ -58,6 +58,9 @@ def create_parser() -> ArgumentParser:
help="Raw structured log messages for stable unexpected results."
" '--log-raw' Must also be passed in order to use this.",
)
parser.add_argument(
"--legacy", default=False, action="store_true", help="Run WPT with the legacy Servo configuration"
)
return parser

10
python/wpt/run.py
View File

@@ -49,7 +49,12 @@ def run_tests(default_binary_path: str, multiprocess: bool, **kwargs: Any) -> in
# makes CI logs unreadable.
github_context = os.environ.pop("GITHUB_CONTEXT", None)
set_if_none(kwargs, "product", "servodriver")
# Allow to run with the legacy Servo WPT configuration. This is required
# until necessary improvements are made to the debugging experience with
# servodriver. See https://github.com/servo/servo/issues/40751
product = "servo_legacy" if kwargs.get("legacy") else "servo"
set_if_none(kwargs, "product", product)
set_if_none(kwargs, "config", os.path.join(WPT_PATH, "config.ini"))
set_if_none(kwargs, "include_manifest", os.path.join(WPT_PATH, "include.ini"))
set_if_none(kwargs, "manifest_update", False)
@@ -88,9 +93,8 @@ def run_tests(default_binary_path: str, multiprocess: bool, **kwargs: Any) -> in
if not kwargs.get("no_default_test_types"):
test_types = {
"servo": ["testharness", "reftest", "wdspec", "crashtest"],
"servodriver": ["testharness", "reftest", "wdspec", "crashtest"],
"servo_legacy": ["testharness", "reftest", "wdspec", "crashtest"],
}
product = kwargs.get("product") or "servo"
kwargs["test_types"] = test_types[product]
filter_intermittents_output = kwargs.pop("filter_intermittents", None)

2
tests/wpt/config.ini vendored
View File

@@ -1,6 +1,6 @@
[products]
servo =
servodriver =
servo_legacy =
firefox =
[web-platform-tests]

20
tests/wpt/meta/MANIFEST.json vendored
View File

@@ -542308,7 +542308,7 @@
[]
],
"browser.py": [
"8d8e9703648fb4d27fe172a935ebf8ad32615876",
"4265ad06f9b02022da199a79e4879c3c079f648d",
[]
],
"commands.json": [
@@ -542356,7 +542356,7 @@
[]
],
"run.py": [
"00e8cf9197e12f34067f79f6e7311f5684232d60",
"5cbeca8f5617f35ac327a22cc23afa9ab7d01461",
[]
],
"testfiles.py": [
@@ -542554,7 +542554,7 @@
],
"browsers": {
"__init__.py": [
"c27ae7281dc075226c0166b2d038c019b987373c",
"e103b126bb9018ec5c8356d1a093d2bf42da5173",
[]
],
"android_webview.py": [
@@ -542628,11 +542628,11 @@
]
},
"servo.py": [
"7a95b08e7ad9ef88be26dfdaf4547b198a0a1439",
"c135311b1e137a9de74363922debc08326d0cb82",
[]
],
"servodriver.py": [
"c2ba9e4c818f94ba0c5a133dfe23aaa78d04e710",
"servo_legacy.py": [
"47b3b504f58750d681ce7a5e04598e30e093d06d",
[]
],
"webkit.py": [
@@ -542698,11 +542698,11 @@
[]
],
"executorservo.py": [
"9c1a762703aaee875554fc6d15e60f0d806b44ee",
"ef25248c03097fcbfb49a86637a355d1edc0fa38",
[]
],
"executorservodriver.py": [
"37dc8862b782d523481a476eaa294dc45e6fa12a",
"executorservolegacy.py": [
"a03c2d4ee6fff5e606a420789840c596476f8fd4",
[]
],
"executorwebdriver.py": [
@@ -542862,7 +542862,7 @@
[]
],
"base.py": [
"8e71aba812de83b1dccd4dfe58de2ca6d57e2ba4",
"817fc8f9d9729d8bb9ef577ff895b9572ac15469",
[]
],
"browsers": {

6
tests/wpt/tests/tools/wpt/browser.py vendored
View File

@@ -2339,9 +2339,9 @@ class Servo(Browser):
if m:
return m.group(0)
class ServoWebDriver(Servo):
product = "servodriver"
# Uses same configuration as Servo
class ServoLegacy(Servo):
product = "servo_legacy"
class Sauce(Browser):

20
tests/wpt/tests/tools/wpt/run.py vendored
View File

@@ -723,9 +723,21 @@ class Servo(BrowserSetup):
kwargs["binary"] = binary
class ServoWebDriver(Servo):
name = "servodriver"
browser_cls = browser.ServoWebDriver
class ServoLegacy(Servo):
name = "servo_legacy"
browser_cls = browser.ServoLegacy
def install(self, channel=None):
if self.prompt_install(self.name):
return self.browser.install(self.venv.path)
def setup_kwargs(self, kwargs):
if kwargs["binary"] is None:
binary = self.browser.find_binary(self.venv.path, None)
if binary is None:
raise WptrunError("Unable to find servo binary in PATH")
kwargs["binary"] = binary
class WebKit(BrowserSetup):
@@ -840,7 +852,7 @@ product_setup = {
"headless_shell": HeadlessShell,
"safari": Safari,
"servo": Servo,
"servodriver": ServoWebDriver,
"servo_legacy": ServoLegacy,
"sauce": Sauce,
"opera": Opera,
"webkit": WebKit,

2
tests/wpt/tests/tools/wptrunner/wptrunner/browsers/__init__.py vendored
View File

@@ -34,7 +34,7 @@ product_list = ["android_webview",
"safari",
"sauce",
"servo",
"servodriver",
"servo_legacy",
"opera",
"webkit",
"webkitgtk_minibrowser",

194
tests/wpt/tests/tools/wptrunner/wptrunner/browsers/servo.py vendored
View File

@@ -1,27 +1,31 @@
# mypy: allow-untyped-defs
import os
import requests
import tempfile
import time
from .base import ExecutorBrowser, NullBrowser, WebDriverBrowser, require_arg
from tools.serve.serve import make_hosts_file
from .base import (WebDriverBrowser,
require_arg)
from .base import get_timeout_multiplier # noqa: F401
from ..executors import executor_kwargs as base_executor_kwargs
from ..executors.base import WdspecExecutor # noqa: F401
from ..executors.executorservo import (ServoCrashtestExecutor, # noqa: F401
ServoTestharnessExecutor, # noqa: F401
ServoRefTestExecutor) # noqa: F401
from ..executors.executorservo import (ServoTestharnessExecutor, # noqa: F401
ServoRefTestExecutor, # noqa: F401
ServoCrashtestExecutor) # noqa: F401
here = os.path.dirname(__file__)
__wptrunner__ = {
"product": "servo",
"check_args": "check_args",
"browser": {None: "ServoBrowser",
"wdspec": "ServoWdspecBrowser"},
"browser": "ServoBrowser",
"executor": {
"crashtest": "ServoCrashtestExecutor",
"testharness": "ServoTestharnessExecutor",
"reftest": "ServoRefTestExecutor",
"crashtest": "ServoCrashtestExecutor",
"wdspec": "WdspecExecutor",
},
"browser_kwargs": "browser_kwargs",
@@ -40,21 +44,18 @@ def check_args(**kwargs):
def browser_kwargs(logger, test_type, run_info_data, config, subsuite, **kwargs):
return {
"binary": kwargs["binary"],
"debug_info": kwargs["debug_info"],
"binary_args": kwargs["binary_args"] + subsuite.config.get("binary_args", []),
"headless": kwargs["headless"],
"debug_info": kwargs["debug_info"],
"server_config": config,
"user_stylesheets": kwargs.get("user_stylesheets"),
"ca_certificate_path": config.ssl_config["ca_cert_path"],
"headless": kwargs.get("headless"),
"capabilities": kwargs.get("capabilities"),
}
def executor_kwargs(logger, test_type, test_environment, run_info_data,
**kwargs):
def executor_kwargs(logger, test_type, test_environment, run_info_data, **kwargs):
rv = base_executor_kwargs(test_type, test_environment, run_info_data, **kwargs)
rv["pause_after_test"] = kwargs["pause_after_test"]
rv["headless"] = kwargs.get("headless", False)
if test_type == "wdspec":
rv["capabilities"] = {}
rv['capabilities'] = {}
return rv
@@ -64,68 +65,125 @@ def env_extras(**kwargs):
def env_options():
return {"server_host": "127.0.0.1",
"bind_address": False,
"testharnessreport": "testharnessreport-servo.js",
"supports_debugger": True}
def update_properties():
return ["debug", "os", "processor", "subsuite"], {"os": ["version"], "processor": ["bits"]}
return (["debug", "os", "processor", "subsuite"], {"os": ["version"], "processor": ["bits"]})
class ServoBrowser(NullBrowser):
def __init__(self, logger, binary, debug_info=None, binary_args=None,
user_stylesheets=None, ca_certificate_path=None, **kwargs):
NullBrowser.__init__(self, logger, **kwargs)
self.binary = binary
self.debug_info = debug_info
self.binary_args = binary_args or []
self.user_stylesheets = user_stylesheets or []
self.ca_certificate_path = ca_certificate_path
def executor_browser(self):
return ExecutorBrowser, {
"binary": self.binary,
"debug_info": self.debug_info,
"binary_args": self.binary_args,
"user_stylesheets": self.user_stylesheets,
"ca_certificate_path": self.ca_certificate_path,
}
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
class ServoWdspecBrowser(WebDriverBrowser):
# TODO: could share an implemenation with servodriver.py, perhaps
def __init__(self, logger, binary="servo", webdriver_binary="servo",
binary_args=None, webdriver_args=None, env=None, port=None,
headless=None,
**kwargs):
class ServoBrowser(WebDriverBrowser):
init_timeout = 300 # Large timeout for cases where we're booting an Android emulator
shutdown_retry_attempts = 3
env = os.environ.copy() if env is None else env
def __init__(self, logger, binary, debug_info=None, webdriver_host="127.0.0.1",
server_config=None, binary_args=None,
user_stylesheets=None, headless=None, **kwargs):
hosts_path = write_hosts_file(server_config)
env = os.environ.copy()
env["HOST_FILE"] = hosts_path
env["RUST_BACKTRACE"] = "1"
super().__init__(logger,
binary=binary,
webdriver_binary=webdriver_binary,
webdriver_args=webdriver_args,
port=port,
env=env,
**kwargs)
self.binary_args = binary_args
self.headless = ["--headless"] if headless else None
if debug_info:
env["DELAY_AFTER_ACCEPT"] = env.get("DELAY_SECS", "15")
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",
"--enable-experimental-web-platform-features",
"--window-size", "800x600",
"about:blank",
]
ca_cert_path = server_config.ssl_config["ca_cert_path"]
if ca_cert_path:
args += ["--certificate-path", ca_cert_path]
if binary_args:
args += binary_args
if user_stylesheets:
for stylesheet in user_stylesheets:
args += ["--user-stylesheet", stylesheet]
if headless:
args += ["--headless"]
# Add the shared `wpt-prefs.json` file to the list of arguments.
args += ["--prefs-file", self.find_wpt_prefs(logger)]
super().__init__(logger, binary=binary, webdriver_binary=binary,
webdriver_args=args, host=webdriver_host, env=env,
supports_pac=False, **kwargs)
self.hosts_path = hosts_path
def make_command(self):
command = [self.binary,
f"--webdriver={self.port}",
"--hard-fail",
# 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",
"--window-size",
"800x600",
"about:blank"] + self.webdriver_args
if self.binary_args:
command += self.binary_args
if self.headless:
command += self.headless
return command
return [self.webdriver_binary, f"--webdriver={self.port}"] + self.webdriver_args
def cleanup(self):
os.remove(self.hosts_path)
def is_alive(self):
# This is broken. It is always True.
if not super().is_alive():
return False
try:
requests.get(f"http://{self.host}:{self.port}/status", timeout=3)
except requests.exceptions.Timeout:
# FIXME: This indicates a hanged browser. Reasons need to be investigated further.
# It happens with ~0.1% probability in our CI runs.
self.logger.debug("Servo webdriver status request timed out.")
return True
except Exception as exception:
self.logger.debug(f"Servo has shut down normally. {exception}")
return False
return True
def stop(self, force=False):
retry_cnt = 0
while self.is_alive():
self.logger.info("Trying to shut down gracefully by extension command")
try:
requests.delete(
f"http://{self.host}:{self.port}/session/dummy-session-id/servo/shutdown",
timeout=3
)
except requests.exceptions.ConnectionError:
self.logger.debug("Browser already shut down (connection refused)")
break
except requests.exceptions.RequestException as exeception:
self.logger.debug(f"Request exception: {exeception}")
break
except requests.exceptions.Timeout:
self.logger.debug("Request timed out")
break
retry_cnt += 1
if retry_cnt >= self.shutdown_retry_attempts:
self.logger.warning("Max retry exceeded to normally shut down. Killing instead.")
break
time.sleep(1)
super().stop(force)
def find_wpt_prefs(self, logger):
default_path = os.path.join("resources", "wpt-prefs.json")
# The cwd is the servo repo for `./mach test-wpt`, but on WPT runners
# it is the WPT repo. The nightly tar is extracted inside the Python
# virtual environment within the repo. This means that on WPT runners,
# the cwd has the `_venv3/servo` directory inside which we find the
# binary and the 'resources' directory.
for dir in [".", "./_venv3/servo"]:
candidate = os.path.abspath(os.path.join(dir, default_path))
if os.path.isfile(candidate):
return candidate
logger.error("Unable to find wpt-prefs.json")
return default_path

131
tests/wpt/tests/tools/wptrunner/wptrunner/browsers/servo_legacy.py vendored Normal file
View File

@@ -0,0 +1,131 @@
# mypy: allow-untyped-defs
import os
from .base import ExecutorBrowser, NullBrowser, WebDriverBrowser, require_arg
from .base import get_timeout_multiplier # noqa: F401
from ..executors import executor_kwargs as base_executor_kwargs
from ..executors.base import WdspecExecutor # noqa: F401
from ..executors.executorservolegacy import (ServoLegacyCrashtestExecutor, # noqa: F401
ServoLegacyTestharnessExecutor, # noqa: F401
ServoLegacyRefTestExecutor) # noqa: F401
here = os.path.dirname(__file__)
__wptrunner__ = {
"product": "servo_legacy",
"check_args": "check_args",
"browser": {None: "ServoLegacyBrowser",
"wdspec": "ServoLegacyWdspecBrowser"},
"executor": {
"crashtest": "ServoLegacyCrashtestExecutor",
"testharness": "ServoLegacyTestharnessExecutor",
"reftest": "ServoLegacyRefTestExecutor",
"wdspec": "WdspecExecutor",
},
"browser_kwargs": "browser_kwargs",
"executor_kwargs": "executor_kwargs",
"env_extras": "env_extras",
"env_options": "env_options",
"timeout_multiplier": "get_timeout_multiplier",
"update_properties": "update_properties",
}
def check_args(**kwargs):
require_arg(kwargs, "binary")
def browser_kwargs(logger, test_type, run_info_data, config, subsuite, **kwargs):
return {
"binary": kwargs["binary"],
"debug_info": kwargs["debug_info"],
"binary_args": kwargs["binary_args"] + subsuite.config.get("binary_args", []),
"headless": kwargs["headless"],
"user_stylesheets": kwargs.get("user_stylesheets"),
"ca_certificate_path": config.ssl_config["ca_cert_path"],
}
def executor_kwargs(logger, test_type, test_environment, run_info_data,
**kwargs):
rv = base_executor_kwargs(test_type, test_environment, run_info_data, **kwargs)
rv["pause_after_test"] = kwargs["pause_after_test"]
rv["headless"] = kwargs.get("headless", False)
if test_type == "wdspec":
rv["capabilities"] = {}
return rv
def env_extras(**kwargs):
return []
def env_options():
return {"server_host": "127.0.0.1",
"bind_address": False,
"testharnessreport": "testharnessreport-servo.js",
"supports_debugger": True}
def update_properties():
return ["debug", "os", "processor", "subsuite"], {"os": ["version"], "processor": ["bits"]}
class ServoLegacyBrowser(NullBrowser):
def __init__(self, logger, binary, debug_info=None, binary_args=None,
user_stylesheets=None, ca_certificate_path=None, **kwargs):
NullBrowser.__init__(self, logger, **kwargs)
self.binary = binary
self.debug_info = debug_info
self.binary_args = binary_args or []
self.user_stylesheets = user_stylesheets or []
self.ca_certificate_path = ca_certificate_path
def executor_browser(self):
return ExecutorBrowser, {
"binary": self.binary,
"debug_info": self.debug_info,
"binary_args": self.binary_args,
"user_stylesheets": self.user_stylesheets,
"ca_certificate_path": self.ca_certificate_path,
}
class ServoLegacyWdspecBrowser(WebDriverBrowser):
# TODO: could share an implemenation with servodriver.py, perhaps
def __init__(self, logger, binary="servo", webdriver_binary="servo",
binary_args=None, webdriver_args=None, env=None, port=None,
headless=None,
**kwargs):
env = os.environ.copy() if env is None else env
env["RUST_BACKTRACE"] = "1"
super().__init__(logger,
binary=binary,
webdriver_binary=webdriver_binary,
webdriver_args=webdriver_args,
port=port,
env=env,
**kwargs)
self.binary_args = binary_args
self.headless = ["--headless"] if headless else None
def make_command(self):
command = [self.binary,
f"--webdriver={self.port}",
"--hard-fail",
# 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",
"--window-size",
"800x600",
"about:blank"] + self.webdriver_args
if self.binary_args:
command += self.binary_args
if self.headless:
command += self.headless
return command

189
tests/wpt/tests/tools/wptrunner/wptrunner/browsers/servodriver.py vendored
View File

@@ -1,189 +0,0 @@
# mypy: allow-untyped-defs
import os
import requests
import tempfile
import time
from tools.serve.serve import make_hosts_file
from .base import (WebDriverBrowser,
require_arg)
from .base import get_timeout_multiplier # noqa: F401
from ..executors import executor_kwargs as base_executor_kwargs
from ..executors.base import WdspecExecutor # noqa: F401
from ..executors.executorservodriver import (ServoWebDriverTestharnessExecutor, # noqa: F401
ServoWebDriverRefTestExecutor, # noqa: F401
ServoWebDriverCrashtestExecutor) # noqa: F401
here = os.path.dirname(__file__)
__wptrunner__ = {
"product": "servodriver",
"check_args": "check_args",
"browser": "ServoWebDriverBrowser",
"executor": {
"testharness": "ServoWebDriverTestharnessExecutor",
"reftest": "ServoWebDriverRefTestExecutor",
"crashtest": "ServoWebDriverCrashtestExecutor",
"wdspec": "WdspecExecutor",
},
"browser_kwargs": "browser_kwargs",
"executor_kwargs": "executor_kwargs",
"env_extras": "env_extras",
"env_options": "env_options",
"timeout_multiplier": "get_timeout_multiplier",
"update_properties": "update_properties",
}
def check_args(**kwargs):
require_arg(kwargs, "binary")
def browser_kwargs(logger, test_type, run_info_data, config, subsuite, **kwargs):
return {
"binary": kwargs["binary"],
"binary_args": kwargs["binary_args"] + subsuite.config.get("binary_args", []),
"debug_info": kwargs["debug_info"],
"server_config": config,
"user_stylesheets": kwargs.get("user_stylesheets"),
"headless": kwargs.get("headless"),
"capabilities": kwargs.get("capabilities"),
}
def executor_kwargs(logger, test_type, test_environment, run_info_data, **kwargs):
rv = base_executor_kwargs(test_type, test_environment, run_info_data, **kwargs)
rv['capabilities'] = {}
return rv
def env_extras(**kwargs):
return []
def env_options():
return {"server_host": "127.0.0.1",
"supports_debugger": True}
def update_properties():
return (["debug", "os", "processor", "subsuite"], {"os": ["version"], "processor": ["bits"]})
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
class ServoWebDriverBrowser(WebDriverBrowser):
init_timeout = 300 # Large timeout for cases where we're booting an Android emulator
shutdown_retry_attempts = 3
def __init__(self, logger, binary, debug_info=None, webdriver_host="127.0.0.1",
server_config=None, binary_args=None,
user_stylesheets=None, headless=None, **kwargs):
hosts_path = write_hosts_file(server_config)
env = os.environ.copy()
env["HOST_FILE"] = hosts_path
env["RUST_BACKTRACE"] = "1"
if debug_info:
env["DELAY_AFTER_ACCEPT"] = env.get("DELAY_SECS", "15")
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",
"--enable-experimental-web-platform-features",
"--window-size", "800x600",
"about:blank",
]
ca_cert_path = server_config.ssl_config["ca_cert_path"]
if ca_cert_path:
args += ["--certificate-path", ca_cert_path]
if binary_args:
args += binary_args
if user_stylesheets:
for stylesheet in user_stylesheets:
args += ["--user-stylesheet", stylesheet]
if headless:
args += ["--headless"]
# Add the shared `wpt-prefs.json` file to the list of arguments.
args += ["--prefs-file", self.find_wpt_prefs(logger)]
super().__init__(logger, binary=binary, webdriver_binary=binary,
webdriver_args=args, host=webdriver_host, env=env,
supports_pac=False, **kwargs)
self.hosts_path = hosts_path
def make_command(self):
return [self.webdriver_binary, f"--webdriver={self.port}"] + self.webdriver_args
def cleanup(self):
os.remove(self.hosts_path)
def is_alive(self):
# This is broken. It is always True.
if not super().is_alive():
return False
try:
requests.get(f"http://{self.host}:{self.port}/status", timeout=3)
except requests.exceptions.Timeout:
# FIXME: This indicates a hanged browser. Reasons need to be investigated further.
# It happens with ~0.1% probability in our CI runs.
self.logger.debug("Servo webdriver status request timed out.")
return True
except Exception as exception:
self.logger.debug(f"Servo has shut down normally. {exception}")
return False
return True
def stop(self, force=False):
retry_cnt = 0
while self.is_alive():
self.logger.info("Trying to shut down gracefully by extension command")
try:
requests.delete(
f"http://{self.host}:{self.port}/session/dummy-session-id/servo/shutdown",
timeout=3
)
except requests.exceptions.ConnectionError:
self.logger.debug("Browser already shut down (connection refused)")
break
except requests.exceptions.RequestException as exeception:
self.logger.debug(f"Request exception: {exeception}")
break
except requests.exceptions.Timeout:
self.logger.debug("Request timed out")
break
retry_cnt += 1
if retry_cnt >= self.shutdown_retry_attempts:
self.logger.warning("Max retry exceeded to normally shut down. Killing instead.")
break
time.sleep(1)
super().stop(force)
def find_wpt_prefs(self, logger):
default_path = os.path.join("resources", "wpt-prefs.json")
# The cwd is the servo repo for `./mach test-wpt`, but on WPT runners
# it is the WPT repo. The nightly tar is extracted inside the Python
# virtual environment within the repo. This means that on WPT runners,
# the cwd has the `_venv3/servo` directory inside which we find the
# binary and the 'resources' directory.
for dir in [".", "./_venv3/servo"]:
candidate = os.path.abspath(os.path.join(dir, default_path))
if os.path.isfile(candidate):
return candidate
logger.error("Unable to find wpt-prefs.json")
return default_path

474
tests/wpt/tests/tools/wptrunner/wptrunner/executors/executorservo.py vendored
View File

@@ -1,369 +1,153 @@
# mypy: allow-untyped-defs
import base64
import json
import os
import subprocess
import tempfile
import threading
import traceback
import uuid
from mozprocess import ProcessHandler
from .executorwebdriver import (
WebDriverProtocol,
WebDriverTestharnessExecutor,
WebDriverTestharnessProtocolPart,
WebDriverRefTestExecutor,
WebDriverCrashtestExecutor,
)
from tools.serve.serve import make_hosts_file
from .base import (RefTestExecutor, RefTestImplementation, TestExecutor,
crashtest_result_converter,
testharness_result_converter,
reftest_result_converter,
TimedRunner)
from .protocol import ConnectionlessProtocol
from ..browsers.base import browser_command
pytestrunner = None
webdriver = None
ServoCommandExtensions = None
here = os.path.dirname(__file__)
# A mixin class that includes functionality common to all Servo
# executors that work by spawing a new process. This is intended to
# be used along with either the `TestExecutor` class or its children
# and must be the first in the inheritance list to allow `super`
# to forward the calls to correct base class.
class ServoExecutorMixin:
def __init__(self, logger, browser, server_config, headless,
timeout_multiplier, debug_info,
pause_after_test, reftest_screenshot="unexpected"):
super().__init__(logger, browser, server_config,
timeout_multiplier=timeout_multiplier,
debug_info=debug_info,
reftest_screenshot=reftest_screenshot)
self.binary = self.browser.binary
self.interactive = (False if self.debug_info is None
else self.debug_info.interactive)
self.pause_after_test = pause_after_test
self.environment = {}
self.protocol = ConnectionlessProtocol(self, browser)
self.headless = headless
def do_delayed_imports():
global webdriver
import webdriver
self.wpt_prefs_path = self.find_wpt_prefs()
global ServoCommandExtensions
hosts_fd, self.hosts_path = tempfile.mkstemp()
with os.fdopen(hosts_fd, "w") as f:
f.write(make_hosts_file(self.server_config, "127.0.0.1"))
class ServoCommandExtensions:
def __init__(self, session):
self.session = session
self.env_for_tests = os.environ.copy()
self.env_for_tests["HOST_FILE"] = self.hosts_path
self.env_for_tests["RUST_BACKTRACE"] = "1"
def get_prefs(self, *prefs):
body = {"prefs": list(prefs)}
return self.session.send_session_command("GET", "servo/prefs/get", body)
def setup(self, runner, protocol=None):
self.runner = runner
self.runner.send_message("init_succeeded")
def set_prefs(self, prefs):
body = {"prefs": prefs}
return self.session.send_session_command("POST", "servo/prefs/set", body)
def reset_prefs(self, *prefs):
body = {"prefs": list(prefs)}
return self.session.send_session_command("POST", "servo/prefs/reset", body)
def shutdown(self):
body = {}
return self.session.send_session_command("DELETE", "servo/shutdown", body)
# Clear all cookies for all origins.
def reset_all_cookies(self):
body = {}
return self.session.send_session_command("POST", "servo/cookies/reset", body)
def change_prefs(self, old_prefs, new_prefs):
# Servo interprets reset with an empty list as reset everything
if old_prefs:
self.reset_prefs(*old_prefs.keys())
self.set_prefs({k: parse_pref_value(v) for k, v in new_prefs.items()})
# See parse_pref_from_command_line() in components/config/opts.rs
def parse_pref_value(value):
if value == "true":
return True
if value == "false":
return False
try:
return float(value)
except ValueError:
return value
def teardown(self):
try:
os.unlink(self.hosts_path)
except OSError:
pass
super().teardown()
class ServoDriverTestharnessProtocolPart(WebDriverTestharnessProtocolPart):
def reset_browser_state(self):
self.parent.webdriver.extension.reset_all_cookies()
class ServoProtocol(WebDriverProtocol):
implements = [
ServoDriverTestharnessProtocolPart,
]
for base_part in WebDriverProtocol.implements:
if base_part.name not in {part.name for part in implements}:
implements.append(base_part)
def __init__(self, executor, browser, capabilities, **kwargs):
do_delayed_imports()
self.implements = list(ServoProtocol.implements)
super().__init__(executor, browser, capabilities, **kwargs)
def connect(self):
"""Connect to browser via WebDriver and crete a WebDriver session."""
self.logger.debug("Connecting to WebDriver on URL: %s" % self.url)
host, port = self.url.split(":")[1].strip("/"), self.url.split(':')[-1].strip("/")
capabilities = {"alwaysMatch": self.capabilities}
self.webdriver = webdriver.Session(host, port,
capabilities=capabilities,
enable_bidi=self.enable_bidi,
extension=ServoCommandExtensions)
self.webdriver.start()
class ServoTestharnessExecutor(WebDriverTestharnessExecutor):
supports_testdriver = True
protocol_cls = ServoProtocol
def __init__(self, logger, browser, server_config, timeout_multiplier=1,
close_after_done=True, capabilities=None, debug_info=None,
**kwargs):
WebDriverTestharnessExecutor.__init__(self, logger, browser, server_config,
timeout_multiplier, capabilities=capabilities,
debug_info=debug_info, close_after_done=close_after_done,
cleanup_after_test=False)
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 find_wpt_prefs(self):
default_path = os.path.join("resources", "wpt-prefs.json")
# The cwd is the servo repo for `./mach test-wpt`, but on WPT runners
# it is the WPT repo. The nightly tar is extracted inside the python
# virtual environment within the repo. This means that on WPT runners,
# the cwd has the `_venv3/servo` directory inside which we find the
# binary and the 'resources' directory.
for dir in [".", "./_venv3/servo"]:
candidate = os.path.abspath(os.path.join(dir, default_path))
if os.path.isfile(candidate):
return candidate
self.logger.error("Unable to find wpt-prefs.json")
return default_path
def build_servo_command(self, test, extra_args=None):
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",
"--enable-experimental-web-platform-features",
self.test_url(test),
]
if self.headless:
args += ["-z"]
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}"]
args += ["--prefs-file", self.wpt_prefs_path]
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)
return debug_args + command
self.protocol.webdriver.extension.change_prefs(
self.last_environment.get("prefs", {}),
new_environment.get("prefs", {})
)
class ServoTestharnessExecutor(ServoExecutorMixin, TestExecutor):
convert_result = testharness_result_converter
class ServoRefTestExecutor(WebDriverRefTestExecutor):
protocol_cls = ServoProtocol
def __init__(self, logger, browser, server_config, headless,
timeout_multiplier=1, debug_info=None,
pause_after_test=False, **kwargs):
super().__init__(logger, browser, server_config,
headless,
timeout_multiplier=timeout_multiplier,
debug_info=debug_info,
pause_after_test=pause_after_test)
self.result_data = None
self.result_flag = None
def __init__(self, logger, browser, server_config, timeout_multiplier=1,
screenshot_cache=None, capabilities=None, debug_info=None,
**kwargs):
WebDriverRefTestExecutor.__init__(self, logger, browser, server_config,
timeout_multiplier, screenshot_cache,
capabilities=capabilities,
debug_info=debug_info)
def do_test(self, test):
self.test = test
self.result_data = None
self.result_flag = threading.Event()
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=self.env_for_tests,
storeOutput=False)
self.proc.run()
else:
self.proc = subprocess.Popen(self.command, env=self.env_for_tests)
try:
timeout = test.timeout * self.timeout_multiplier
# Now wait to get the output we expect, or until we reach the timeout
if not self.interactive and not self.pause_after_test:
wait_timeout = timeout + 5
self.result_flag.wait(wait_timeout)
else:
wait_timeout = None
self.proc.wait()
proc_is_running = True
if self.result_flag.is_set():
if self.result_data is not None:
result = self.convert_result(test, self.result_data)
else:
self.proc.wait()
result = (test.make_result("CRASH", None), [])
proc_is_running = False
else:
result = (test.make_result("TIMEOUT", None), [])
if proc_is_running:
if self.pause_after_test:
self.logger.info("Pausing until the browser exits")
self.proc.wait()
else:
self.proc.kill()
except: # noqa
self.proc.kill()
raise
return result
def on_output(self, line):
prefix = "ALERT: RESULT: "
decoded_line = line.decode("utf8", "replace")
if decoded_line.startswith(prefix):
try:
self.result_data = json.loads(decoded_line[len(prefix):])
except json.JSONDecodeError as error:
self.logger.error(f"Could not process test output JSON: {error}")
self.result_flag.set()
else:
super().on_output(line)
def on_finish(self):
self.result_flag.set()
def on_environment_change(self, new_environment):
self.protocol.webdriver.extension.change_prefs(
self.last_environment.get("prefs", {}),
new_environment.get("prefs", {})
)
class TempFilename:
def __init__(self, directory):
self.directory = directory
self.path = None
class ServoCrashtestExecutor(WebDriverCrashtestExecutor):
protocol_cls = ServoProtocol
def __enter__(self):
self.path = os.path.join(self.directory, str(uuid.uuid4()))
return self.path
def __init__(self, logger, browser, server_config, timeout_multiplier=1,
screenshot_cache=None, capabilities=None, debug_info=None,
**kwargs):
WebDriverCrashtestExecutor.__init__(self, logger, browser, server_config,
timeout_multiplier, screenshot_cache,
capabilities=capabilities,
debug_info=debug_info)
def __exit__(self, *args, **kwargs):
try:
os.unlink(self.path)
except OSError:
pass
class ServoRefTestExecutor(ServoExecutorMixin, RefTestExecutor):
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):
super().__init__(logger,
browser,
server_config,
headless=True,
timeout_multiplier=timeout_multiplier,
debug_info=debug_info,
reftest_screenshot=reftest_screenshot,
pause_after_test=pause_after_test)
self.screenshot_cache = screenshot_cache
self.reftest_screenshot = reftest_screenshot
self.implementation = RefTestImplementation(self)
self.tempdir = tempfile.mkdtemp()
def reset(self):
self.implementation.reset()
def teardown(self):
os.rmdir(self.tempdir)
super().teardown()
def screenshot(self, test, viewport_size, dpi, page_ranges):
with TempFilename(self.tempdir) as output_path:
extra_args = ["--exit",
"--output=%s" % output_path,
"--window-size", viewport_size or "800x600"]
if dpi:
extra_args += ["--device-pixel-ratio", str(dpi)]
self.command = self.build_servo_command(test, extra_args)
if not self.interactive:
self.proc = ProcessHandler(self.command,
processOutputLine=[self.on_output],
env=self.env_for_tests)
try:
self.proc.run()
timeout = test.timeout * self.timeout_multiplier + 5
rv = self.proc.wait(timeout=timeout)
except KeyboardInterrupt:
self.proc.kill()
raise
else:
self.proc = subprocess.Popen(self.command, env=self.env_for_tests)
try:
rv = self.proc.wait()
except KeyboardInterrupt:
self.proc.kill()
raise
if rv is None:
self.proc.kill()
return False, ("EXTERNAL-TIMEOUT", None)
if rv != 0 or not os.path.exists(output_path):
return False, ("CRASH", None)
with open(output_path, "rb") as f:
# Might need to strip variable headers or something here
data = f.read()
# Returning the screenshot as a string could potentially be avoided,
# see https://github.com/web-platform-tests/wpt/issues/28929.
return True, [base64.b64encode(data).decode()]
def do_test(self, test):
# FIXME: This is a temporary fix until Servo syncs with upstream WPT.
# Once that happens, we can patch the `RefTestImplementation.get_screenshot_list`
# method to cast dpi to integer when using it in arithmetic expressions.
if test.dpi is not None:
test.dpi = int(test.dpi)
self.test = test
result = self.implementation.run_test(test)
return self.convert_result(test, result)
class ServoTimedRunner(TimedRunner):
def run_func(self):
try:
self.result = (True, self.func(self.protocol, self.url, self.timeout))
except Exception as e:
message = getattr(e, "message", "")
if message:
message += "\n"
message += traceback.format_exc(e)
self.result = False, ("INTERNAL-ERROR", message)
finally:
self.result_flag.set()
def set_timeout(self):
pass
class ServoCrashtestExecutor(ServoExecutorMixin, TestExecutor):
convert_result = crashtest_result_converter
def __init__(self, logger, browser, server_config, headless,
binary=None, timeout_multiplier=1, screenshot_cache=None,
debug_info=None, pause_after_test=False, **kwargs):
super().__init__(logger,
browser,
server_config,
headless,
timeout_multiplier=timeout_multiplier,
debug_info=debug_info,
pause_after_test=pause_after_test)
def do_test(self, test):
timeout = (test.timeout * self.timeout_multiplier if self.debug_info is None
else None)
test_url = self.test_url(test)
# We want to pass the full test object into build_servo_command,
# so stash it in the class
self.test = test
success, data = ServoTimedRunner(self.logger, self.do_crashtest, self.protocol,
test_url, timeout, self.extra_timeout).run()
# Ensure that no processes hang around if they timeout.
self.proc.kill()
if success:
return self.convert_result(test, data)
return (test.make_result(*data), [])
def do_crashtest(self, protocol, url, timeout):
self.command = self.build_servo_command(self.test, extra_args=["-x"])
if not self.interactive:
self.proc = ProcessHandler(self.command,
env=self.env_for_tests,
processOutputLine=[self.on_output],
storeOutput=False)
self.proc.run()
else:
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_environment_change(self, new_environment):
self.protocol.webdriver.extension.change_prefs(
self.last_environment.get("prefs", {}),
new_environment.get("prefs", {})
)

153
tests/wpt/tests/tools/wptrunner/wptrunner/executors/executorservodriver.py vendored
View File

@@ -1,153 +0,0 @@
# mypy: allow-untyped-defs
import os
from .executorwebdriver import (
WebDriverProtocol,
WebDriverTestharnessExecutor,
WebDriverTestharnessProtocolPart,
WebDriverRefTestExecutor,
WebDriverCrashtestExecutor,
)
webdriver = None
ServoCommandExtensions = None
here = os.path.dirname(__file__)
def do_delayed_imports():
global webdriver
import webdriver
global ServoCommandExtensions
class ServoCommandExtensions:
def __init__(self, session):
self.session = session
def get_prefs(self, *prefs):
body = {"prefs": list(prefs)}
return self.session.send_session_command("GET", "servo/prefs/get", body)
def set_prefs(self, prefs):
body = {"prefs": prefs}
return self.session.send_session_command("POST", "servo/prefs/set", body)
def reset_prefs(self, *prefs):
body = {"prefs": list(prefs)}
return self.session.send_session_command("POST", "servo/prefs/reset", body)
def shutdown(self):
body = {}
return self.session.send_session_command("DELETE", "servo/shutdown", body)
# Clear all cookies for all origins.
def reset_all_cookies(self):
body = {}
return self.session.send_session_command("POST", "servo/cookies/reset", body)
def change_prefs(self, old_prefs, new_prefs):
# Servo interprets reset with an empty list as reset everything
if old_prefs:
self.reset_prefs(*old_prefs.keys())
self.set_prefs({k: parse_pref_value(v) for k, v in new_prefs.items()})
# See parse_pref_from_command_line() in components/config/opts.rs
def parse_pref_value(value):
if value == "true":
return True
if value == "false":
return False
try:
return float(value)
except ValueError:
return value
class ServoDriverTestharnessProtocolPart(WebDriverTestharnessProtocolPart):
def reset_browser_state(self):
self.parent.webdriver.extension.reset_all_cookies()
class ServoWebDriverProtocol(WebDriverProtocol):
implements = [
ServoDriverTestharnessProtocolPart,
]
for base_part in WebDriverProtocol.implements:
if base_part.name not in {part.name for part in implements}:
implements.append(base_part)
def __init__(self, executor, browser, capabilities, **kwargs):
do_delayed_imports()
self.implements = list(ServoWebDriverProtocol.implements)
super().__init__(executor, browser, capabilities, **kwargs)
def connect(self):
"""Connect to browser via WebDriver and crete a WebDriver session."""
self.logger.debug("Connecting to WebDriver on URL: %s" % self.url)
host, port = self.url.split(":")[1].strip("/"), self.url.split(':')[-1].strip("/")
capabilities = {"alwaysMatch": self.capabilities}
self.webdriver = webdriver.Session(host, port,
capabilities=capabilities,
enable_bidi=self.enable_bidi,
extension=ServoCommandExtensions)
self.webdriver.start()
class ServoWebDriverTestharnessExecutor(WebDriverTestharnessExecutor):
supports_testdriver = True
protocol_cls = ServoWebDriverProtocol
def __init__(self, logger, browser, server_config, timeout_multiplier=1,
close_after_done=True, capabilities=None, debug_info=None,
**kwargs):
WebDriverTestharnessExecutor.__init__(self, logger, browser, server_config,
timeout_multiplier, capabilities=capabilities,
debug_info=debug_info, close_after_done=close_after_done,
cleanup_after_test=False)
def on_environment_change(self, new_environment):
self.protocol.webdriver.extension.change_prefs(
self.last_environment.get("prefs", {}),
new_environment.get("prefs", {})
)
class ServoWebDriverRefTestExecutor(WebDriverRefTestExecutor):
protocol_cls = ServoWebDriverProtocol
def __init__(self, logger, browser, server_config, timeout_multiplier=1,
screenshot_cache=None, capabilities=None, debug_info=None,
**kwargs):
WebDriverRefTestExecutor.__init__(self, logger, browser, server_config,
timeout_multiplier, screenshot_cache,
capabilities=capabilities,
debug_info=debug_info)
def on_environment_change(self, new_environment):
self.protocol.webdriver.extension.change_prefs(
self.last_environment.get("prefs", {}),
new_environment.get("prefs", {})
)
class ServoWebDriverCrashtestExecutor(WebDriverCrashtestExecutor):
protocol_cls = ServoWebDriverProtocol
def __init__(self, logger, browser, server_config, timeout_multiplier=1,
screenshot_cache=None, capabilities=None, debug_info=None,
**kwargs):
WebDriverCrashtestExecutor.__init__(self, logger, browser, server_config,
timeout_multiplier, screenshot_cache,
capabilities=capabilities,
debug_info=debug_info)
def on_environment_change(self, new_environment):
self.protocol.webdriver.extension.change_prefs(
self.last_environment.get("prefs", {}),
new_environment.get("prefs", {})
)

369
tests/wpt/tests/tools/wptrunner/wptrunner/executors/executorservolegacy.py vendored Normal file
View File

@@ -0,0 +1,369 @@
# mypy: allow-untyped-defs
import base64
import json
import os
import subprocess
import tempfile
import threading
import traceback
import uuid
from mozprocess import ProcessHandler
from tools.serve.serve import make_hosts_file
from .base import (RefTestExecutor, RefTestImplementation, TestExecutor,
crashtest_result_converter,
testharness_result_converter,
reftest_result_converter,
TimedRunner)
from .protocol import ConnectionlessProtocol
from ..browsers.base import browser_command
pytestrunner = None
webdriver = None
# A mixin class that includes functionality common to all Servo
# executors that work by spawing a new process. This is intended to
# be used along with either the `TestExecutor` class or its children
# and must be the first in the inheritance list to allow `super`
# to forward the calls to correct base class.
class ServoExecutorMixin:
def __init__(self, logger, browser, server_config, headless,
timeout_multiplier, debug_info,
pause_after_test, reftest_screenshot="unexpected"):
super().__init__(logger, browser, server_config,
timeout_multiplier=timeout_multiplier,
debug_info=debug_info,
reftest_screenshot=reftest_screenshot)
self.binary = self.browser.binary
self.interactive = (False if self.debug_info is None
else self.debug_info.interactive)
self.pause_after_test = pause_after_test
self.environment = {}
self.protocol = ConnectionlessProtocol(self, browser)
self.headless = headless
self.wpt_prefs_path = self.find_wpt_prefs()
hosts_fd, self.hosts_path = tempfile.mkstemp()
with os.fdopen(hosts_fd, "w") as f:
f.write(make_hosts_file(self.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 setup(self, runner, protocol=None):
self.runner = runner
self.runner.send_message("init_succeeded")
return True
def teardown(self):
try:
os.unlink(self.hosts_path)
except OSError:
pass
super().teardown()
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 find_wpt_prefs(self):
default_path = os.path.join("resources", "wpt-prefs.json")
# The cwd is the servo repo for `./mach test-wpt`, but on WPT runners
# it is the WPT repo. The nightly tar is extracted inside the python
# virtual environment within the repo. This means that on WPT runners,
# the cwd has the `_venv3/servo` directory inside which we find the
# binary and the 'resources' directory.
for dir in [".", "./_venv3/servo"]:
candidate = os.path.abspath(os.path.join(dir, default_path))
if os.path.isfile(candidate):
return candidate
self.logger.error("Unable to find wpt-prefs.json")
return default_path
def build_servo_command(self, test, extra_args=None):
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",
"--enable-experimental-web-platform-features",
self.test_url(test),
]
if self.headless:
args += ["-z"]
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}"]
args += ["--prefs-file", self.wpt_prefs_path]
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)
return debug_args + command
class ServoLegacyTestharnessExecutor(ServoExecutorMixin, TestExecutor):
convert_result = testharness_result_converter
def __init__(self, logger, browser, server_config, headless,
timeout_multiplier=1, debug_info=None,
pause_after_test=False, **kwargs):
super().__init__(logger, browser, server_config,
headless,
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 = self.build_servo_command(test)
if not self.interactive:
self.proc = ProcessHandler(self.command,
processOutputLine=[self.on_output],
onFinish=self.on_finish,
env=self.env_for_tests,
storeOutput=False)
self.proc.run()
else:
self.proc = subprocess.Popen(self.command, env=self.env_for_tests)
try:
timeout = test.timeout * self.timeout_multiplier
# Now wait to get the output we expect, or until we reach the timeout
if not self.interactive and not self.pause_after_test:
wait_timeout = timeout + 5
self.result_flag.wait(wait_timeout)
else:
wait_timeout = None
self.proc.wait()
proc_is_running = True
if self.result_flag.is_set():
if self.result_data is not None:
result = self.convert_result(test, self.result_data)
else:
self.proc.wait()
result = (test.make_result("CRASH", None), [])
proc_is_running = False
else:
result = (test.make_result("TIMEOUT", None), [])
if proc_is_running:
if self.pause_after_test:
self.logger.info("Pausing until the browser exits")
self.proc.wait()
else:
self.proc.kill()
except: # noqa
self.proc.kill()
raise
return result
def on_output(self, line):
prefix = "ALERT: RESULT: "
decoded_line = line.decode("utf8", "replace")
if decoded_line.startswith(prefix):
try:
self.result_data = json.loads(decoded_line[len(prefix):])
except json.JSONDecodeError as error:
self.logger.error(f"Could not process test output JSON: {error}")
self.result_flag.set()
else:
super().on_output(line)
def on_finish(self):
self.result_flag.set()
class TempFilename:
def __init__(self, directory):
self.directory = directory
self.path = None
def __enter__(self):
self.path = os.path.join(self.directory, str(uuid.uuid4()))
return self.path
def __exit__(self, *args, **kwargs):
try:
os.unlink(self.path)
except OSError:
pass
class ServoLegacyRefTestExecutor(ServoExecutorMixin, RefTestExecutor):
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):
super().__init__(logger,
browser,
server_config,
headless=True,
timeout_multiplier=timeout_multiplier,
debug_info=debug_info,
reftest_screenshot=reftest_screenshot,
pause_after_test=pause_after_test)
self.screenshot_cache = screenshot_cache
self.reftest_screenshot = reftest_screenshot
self.implementation = RefTestImplementation(self)
self.tempdir = tempfile.mkdtemp()
def reset(self):
self.implementation.reset()
def teardown(self):
os.rmdir(self.tempdir)
super().teardown()
def screenshot(self, test, viewport_size, dpi, page_ranges):
with TempFilename(self.tempdir) as output_path:
extra_args = ["--exit",
"--output=%s" % output_path,
"--window-size", viewport_size or "800x600"]
if dpi:
extra_args += ["--device-pixel-ratio", str(dpi)]
self.command = self.build_servo_command(test, extra_args)
if not self.interactive:
self.proc = ProcessHandler(self.command,
processOutputLine=[self.on_output],
env=self.env_for_tests)
try:
self.proc.run()
timeout = test.timeout * self.timeout_multiplier + 5
rv = self.proc.wait(timeout=timeout)
except KeyboardInterrupt:
self.proc.kill()
raise
else:
self.proc = subprocess.Popen(self.command, env=self.env_for_tests)
try:
rv = self.proc.wait()
except KeyboardInterrupt:
self.proc.kill()
raise
if rv is None:
self.proc.kill()
return False, ("EXTERNAL-TIMEOUT", None)
if rv != 0 or not os.path.exists(output_path):
return False, ("CRASH", None)
with open(output_path, "rb") as f:
# Might need to strip variable headers or something here
data = f.read()
# Returning the screenshot as a string could potentially be avoided,
# see https://github.com/web-platform-tests/wpt/issues/28929.
return True, [base64.b64encode(data).decode()]
def do_test(self, test):
# FIXME: This is a temporary fix until Servo syncs with upstream WPT.
# Once that happens, we can patch the `RefTestImplementation.get_screenshot_list`
# method to cast dpi to integer when using it in arithmetic expressions.
if test.dpi is not None:
test.dpi = int(test.dpi)
self.test = test
result = self.implementation.run_test(test)
return self.convert_result(test, result)
class ServoTimedRunner(TimedRunner):
def run_func(self):
try:
self.result = (True, self.func(self.protocol, self.url, self.timeout))
except Exception as e:
message = getattr(e, "message", "")
if message:
message += "\n"
message += traceback.format_exc(e)
self.result = False, ("INTERNAL-ERROR", message)
finally:
self.result_flag.set()
def set_timeout(self):
pass
class ServoLegacyCrashtestExecutor(ServoExecutorMixin, TestExecutor):
convert_result = crashtest_result_converter
def __init__(self, logger, browser, server_config, headless,
binary=None, timeout_multiplier=1, screenshot_cache=None,
debug_info=None, pause_after_test=False, **kwargs):
super().__init__(logger,
browser,
server_config,
headless,
timeout_multiplier=timeout_multiplier,
debug_info=debug_info,
pause_after_test=pause_after_test)
def do_test(self, test):
timeout = (test.timeout * self.timeout_multiplier if self.debug_info is None
else None)
test_url = self.test_url(test)
# We want to pass the full test object into build_servo_command,
# so stash it in the class
self.test = test
success, data = ServoTimedRunner(self.logger, self.do_crashtest, self.protocol,
test_url, timeout, self.extra_timeout).run()
# Ensure that no processes hang around if they timeout.
self.proc.kill()
if success:
return self.convert_result(test, data)
return (test.make_result(*data), [])
def do_crashtest(self, protocol, url, timeout):
self.command = self.build_servo_command(self.test, extra_args=["-x"])
if not self.interactive:
self.proc = ProcessHandler(self.command,
env=self.env_for_tests,
processOutputLine=[self.on_output],
storeOutput=False)
self.proc.run()
else:
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}

2
tests/wpt/tests/tools/wptrunner/wptrunner/tests/base.py vendored
View File

@@ -20,7 +20,7 @@ if "CURRENT_TOX_ENV" in os.environ:
tox_env_extra_browsers = {
"chrome": {"chrome_android"},
"servo": {"servodriver"},
"servo": {"servo_legacy"},
}
_active_products = set(_products) & set(current_tox_env_split)