[wptrunner] Reset permissions between tests for Chrome (#48610)

Similar to #48106, permission changes may leak between tests that use
the same WebDriver session (i.e., browser process). For now, add a
Chromium-specific `Browser.resetPermissions` call to reset permission
defaults.

To consolidate calls that reset state, refactor the WebDriver executor's
testharness window management to align with #45735.

Author: jonathan-j-lee

tools/wptrunner/wptrunner/executors/executorchrome.py

@@ -80,66 +80,28 @@ class ChromeDriverTestharnessProtocolPart(WebDriverTestharnessProtocolPart):
     The main difference from the default WebDriver testharness implementation is
     that the test window can be reused between tests for better performance.
     """
-
-    def setup(self):
-        super().setup()
-        # Handle (an alphanumeric string) that may be set if window reuse is
-        # enabled. This state allows the protocol to distinguish the test
-        # window from other windows a test itself may create that the "Get
-        # Window Handles" command also returns.
-        #
-        # Because test window persistence is a Chrome-only feature, it's not
-        # exposed to the base WebDriver testharness executor.
-        self.test_window = None
-        self.reuse_window = self.parent.reuse_window
-
-    def close_test_window(self):
-        if self.test_window:
-            self._close_window(self.test_window)
-            self.test_window = None
-
-    def close_old_windows(self):
-        self.webdriver.actions.release()
-        for handle in self.webdriver.handles:
-            if handle not in {self.runner_handle, self.test_window}:
-                self._close_window(handle)
-        if not self.reuse_window:
-            self.close_test_window()
-        self.webdriver.window_handle = self.runner_handle
-        # TODO(web-platform-tests/wpt#48078): Find a cross-platform way to clear
+    def reset_browser_state(self):
+        # TODO(web-platform-tests/wpt#48078): Find a cross-vendor way to clear
         # cookies for all domains.
         self.parent.cdp.execute_cdp_command("Network.clearBrowserCookies")
-        return self.runner_handle
-
-    def open_test_window(self, window_id):
-        if self.test_window:
-            # Try to reuse the existing test window by emulating the `about:blank`
-            # page with no history you would get with a new window.
-            try:
-                self.webdriver.window_handle = self.test_window
-                # Reset navigation history with Chrome DevTools Protocol:
-                # https://chromedevtools.github.io/devtools-protocol/tot/Page/#method-resetNavigationHistory
-                self.parent.cdp.execute_cdp_command("Page.resetNavigationHistory")
-                self.webdriver.url = "about:blank"
-                return
-            except error.NoSuchWindowException:
-                self.test_window = None
-        super().open_test_window(window_id)
-
-    def get_test_window(self, window_id, parent, timeout=5):
-        if self.test_window:
-            return self.test_window
-        # Poll the handles endpoint for the test window like the base WebDriver
-        # protocol part, but don't bother checking for the serialized
-        # WindowProxy (not supported by Chrome currently).
-        deadline = time.time() + timeout
-        while time.time() < deadline:
-            self.test_window = self._poll_handles_for_test_window(parent)
-            if self.test_window is not None:
-                assert self.test_window != parent
-                return self.test_window
-            time.sleep(0.03)
-        raise Exception("unable to find test window")
+        # Reset default permissions that `test_driver.set_permission(...)` may
+        # have altered.
+        self.parent.cdp.execute_cdp_command("Browser.resetPermissions")
+        # Chromium requires the `background-sync` permission for reporting APIs
+        # to work. Not all embedders (notably, `chrome --headless=old`) grant
+        # `background-sync` by default, so this CDP call ensures the permission
+        # is granted for all origins, in line with the background sync spec's
+        # recommendation [0].
+        #
+        # WebDriver's "Set Permission" command can only act on the test's
+        # origin, which may be too limited.
+        #
+        # [0]: https://wicg.github.io/background-sync/spec/#permission
+        params = {
+            "permission": {"name": "background-sync"},
+            "setting": "granted",
+        }
+        self.parent.cdp.execute_cdp_command("Browser.setPermission", params)
 
 
 class ChromeDriverFedCMProtocolPart(WebDriverFedCMProtocolPart):
@@ -177,7 +139,6 @@ class ChromeDriverProtocol(WebDriverProtocol):
         if base_part.name not in {part.name for part in implements}:
             implements.append(base_part)
 
-    reuse_window = False
     # Prefix to apply to vendor-specific WebDriver extension commands.
     vendor_prefix = "goog"
 
@@ -225,25 +186,24 @@ class ChromeDriverTestharnessExecutor(WebDriverTestharnessExecutor, _SanitizerMi
 
     def __init__(self, *args, reuse_window=False, **kwargs):
         super().__init__(*args, **kwargs)
-        self.protocol.reuse_window = reuse_window
+        self.reuse_window = reuse_window
 
-    def setup(self, runner, protocol=None):
-        super().setup(runner, protocol)
-        # Chromium requires the `background-sync` permission for reporting APIs
-        # to work. Not all embedders (notably, `chrome --headless=old`) grant
-        # `background-sync` by default, so this CDP call ensures the permission
-        # is granted for all origins, in line with the background sync spec's
-        # recommendation [0].
-        #
-        # WebDriver's "Set Permission" command can only act on the test's
-        # origin, which may be too limited.
-        #
-        # [0]: https://wicg.github.io/background-sync/spec/#permission
-        params = {
-            "permission": {"name": "background-sync"},
-            "setting": "granted",
-        }
-        self.protocol.cdp.execute_cdp_command("Browser.setPermission", params)
+    def get_or_create_test_window(self, protocol):
+        test_window = self.protocol.testharness.persistent_test_window
+        if test_window:
+            try:
+                # Mimic the "new window" WebDriver command by loading `about:blank`
+                # with no other browsing history.
+                protocol.base.set_window(test_window)
+                protocol.base.load("about:blank")
+                protocol.cdp.execute_cdp_command("Page.resetNavigationHistory")
+            except error.NoSuchWindowException:
+                test_window = self.protocol.testharness.persistent_test_window = None
+        if not test_window:
+            test_window = super().get_or_create_test_window(protocol)
+        if self.reuse_window:
+            self.protocol.testharness.persistent_test_window = test_window
+        return test_window
 
     def _get_next_message_classic(self, protocol, url, test_window):
         try:

tools/wptrunner/wptrunner/executors/executoredge.py

@@ -30,10 +30,6 @@ class EdgeDriverRefTestExecutor(WebDriverRefTestExecutor, _SanitizerMixin):  # t
 class EdgeDriverTestharnessExecutor(WebDriverTestharnessExecutor, _SanitizerMixin):  # type: ignore
     protocol_cls = EdgeDriverProtocol
 
-    def __init__(self, *args, reuse_window=False, **kwargs):
-        super().__init__(*args, **kwargs)
-        self.protocol.reuse_window = reuse_window
-
 
 class EdgeDriverPrintRefTestExecutor(EdgeDriverRefTestExecutor):
     protocol_cls = EdgeDriverProtocol

tools/wptrunner/wptrunner/executors/executorwebdriver.py

@@ -5,9 +5,7 @@
 import os
 import socket
 import threading
-import time
 import traceback
-import uuid
 from urllib.parse import urljoin
 
 from .base import (AsyncCallbackHandler,
@@ -82,6 +80,9 @@ def set_timeout(self, timeout):
             body = {"type": "script", "ms": timeout * 1000}
             self.webdriver.send_session_command("POST", "timeouts", body)
 
+    def create_window(self, type="tab", **kwargs):
+        return self.webdriver.new_window(type_hint=type)
+
     @property
     def current_window(self):
         return self.webdriver.window_handle
@@ -226,10 +227,9 @@ class WebDriverTestharnessProtocolPart(TestharnessProtocolPart):
     def setup(self):
         self.webdriver = self.parent.webdriver
         self.runner_handle = None
+        self.persistent_test_window = None
         with open(os.path.join(here, "runner.js")) as f:
             self.runner_script = f.read()
-        with open(os.path.join(here, "window-loaded.js")) as f:
-            self.window_loaded_script = f.read()
 
     def load_runner(self, url_protocol):
         if self.runner_handle:
@@ -245,10 +245,11 @@ def load_runner(self, url_protocol):
 
     def close_old_windows(self):
         self.webdriver.actions.release()
-        handles = [item for item in self.webdriver.handles if item != self.runner_handle]
-        for handle in handles:
-            self._close_window(handle)
+        for handle in self.webdriver.handles:
+            if handle not in {self.runner_handle, self.persistent_test_window}:
+                self._close_window(handle)
         self.webdriver.window_handle = self.runner_handle
+        self.reset_browser_state()
         return self.runner_handle
 
     def _close_window(self, window_handle):
@@ -258,65 +259,8 @@ def _close_window(self, window_handle):
         except webdriver_error.NoSuchWindowException:
             pass
 
-    def open_test_window(self, window_id):
-        self.webdriver.execute_script(
-            "window.open('about:blank', '%s', 'noopener')" % window_id)
-
-    def get_test_window(self, window_id, parent, timeout=5):
-        """Find the test window amongst all the open windows.
-        This is assumed to be either the named window or the one after the parent in the list of
-        window handles
-
-        :param window_id: The DOM name of the Window
-        :param parent: The handle of the runner window
-        :param timeout: The time in seconds to wait for the window to appear. This is because in
-                        some implementations there's a race between calling window.open and the
-                        window being added to the list of WebDriver accessible windows."""
-        test_window = None
-        end_time = time.time() + timeout
-        while time.time() < end_time:
-            try:
-                # Try using the JSON serialization of the WindowProxy object,
-                # it's in Level 1 but nothing supports it yet
-                win_s = self.webdriver.execute_script("return window['%s'];" % window_id)
-                win_obj = json.loads(win_s)
-                test_window = win_obj["window-fcc6-11e5-b4f8-330a88ab9d7f"]
-            except Exception:
-                pass
-
-            if test_window is None:
-                test_window = self._poll_handles_for_test_window(parent)
-
-            if test_window is not None:
-                assert test_window != parent
-                return test_window
-
-            time.sleep(0.1)
-
-        raise Exception("unable to find test window")
-
-    def _poll_handles_for_test_window(self, parent):
-        test_window = None
-        after = self.webdriver.handles
-        if len(after) == 2:
-            test_window = next(iter(set(after) - {parent}))
-        elif after[0] == parent and len(after) > 2:
-            # Hope the first one here is the test window
-            test_window = after[1]
-        return test_window
-
-    def test_window_loaded(self):
-        """Wait until the page in the new window has been loaded.
-
-        Hereby ignore Javascript execptions that are thrown when
-        the document has been unloaded due to a process change.
-        """
-        while True:
-            try:
-                self.webdriver.execute_script(self.window_loaded_script, asynchronous=True)
-                break
-            except webdriver_error.JavascriptErrorException:
-                pass
+    def reset_browser_state(self):
+        """Reset browser-wide state that normally persists between tests."""
 
 
 class WebDriverPrintProtocolPart(PrintProtocolPart):
@@ -828,7 +772,6 @@ def __init__(self, logger, browser, server_config, timeout_multiplier=1,
             self._get_next_message = self._get_next_message_classic
 
         self.close_after_done = close_after_done
-        self.window_id = str(uuid.uuid4())
         self.cleanup_after_test = cleanup_after_test
 
     def is_alive(self):
@@ -857,20 +800,16 @@ def do_test(self, test):
     def do_testharness(self, protocol, url, timeout):
         # The previous test may not have closed its old windows (if something
         # went wrong or if cleanup_after_test was False), so clean up here.
-        parent_window = protocol.testharness.close_old_windows()
+        protocol.testharness.close_old_windows()
 
         # If protocol implements `bidi_events`, remove all the existing subscriptions.
         if hasattr(protocol, 'bidi_events'):
             # Use protocol loop to run the async cleanup.
             protocol.loop.run_until_complete(protocol.bidi_events.unsubscribe_all())
 
         # Now start the test harness
-        protocol.testharness.open_test_window(self.window_id)
-        test_window = protocol.testharness.get_test_window(self.window_id,
-                                                           parent_window,
-                                                           timeout=5*self.timeout_multiplier)
+        test_window = self.get_or_create_test_window(protocol)
         self.protocol.base.set_window(test_window)
-
         # Wait until about:blank has been loaded
         protocol.base.execute_script(self.window_loaded_script, asynchronous=True)
 
@@ -975,6 +914,9 @@ async def process_bidi_event(method, params):
 
         return rv, extra
 
+    def get_or_create_test_window(self, protocol):
+        return protocol.base.create_window()
+
     def _get_next_message_classic(self, protocol, url, _):
         """
         Get the next message from the test_driver using the classic WebDriver async script execution. This will block

tools/wptrunner/wptrunner/executors/protocol.py

@@ -215,6 +215,7 @@ def close_old_windows(self, url_protocol):
         contains the initial runner page.
 
         :param str url_protocol: "https" or "http" depending on the test metadata.
+        :returns: A browser-specific handle to the runner page.
         """
         pass