From 51674f365be9af1315f3779eecc39912b8d176ce Mon Sep 17 00:00:00 2001 From: Jonathan Lee Date: Thu, 7 Nov 2024 18:17:35 -0500 Subject: [PATCH] [wptrunner] Extract executor testdriver logic into mixin This no-op refactor will allow other WebDriver executors, not just the testharness executor, to perform testdriver actions. --- .../wptrunner/executors/executorwebdriver.py | 184 ++++++++++-------- 1 file changed, 101 insertions(+), 83 deletions(-) diff --git a/tools/wptrunner/wptrunner/executors/executorwebdriver.py b/tools/wptrunner/wptrunner/executors/executorwebdriver.py index ca7b531227b699..c155a29659f11d 100644 --- a/tools/wptrunner/wptrunner/executors/executorwebdriver.py +++ b/tools/wptrunner/wptrunner/executors/executorwebdriver.py @@ -746,72 +746,19 @@ def run_func(self): self.result_flag.set() -class WebDriverTestharnessExecutor(TestharnessExecutor): - supports_testdriver = True - protocol_cls = WebDriverProtocol - _get_next_message = None - - def __init__(self, logger, browser, server_config, timeout_multiplier=1, - close_after_done=True, capabilities=None, debug_info=None, - cleanup_after_test=True, **kwargs): - """WebDriver-based executor for testharness.js tests""" - TestharnessExecutor.__init__(self, logger, browser, server_config, - timeout_multiplier=timeout_multiplier, - debug_info=debug_info) - self.protocol = self.protocol_cls(self, browser, capabilities) - with open(os.path.join(here, "testharness_webdriver_resume.js")) as f: - self.script_resume = f.read() - with open(os.path.join(here, "window-loaded.js")) as f: - self.window_loaded_script = f.read() - - if hasattr(self.protocol, 'bidi_script'): - # If `bidi_script` is available, the messages can be handled via BiDi. - self._get_next_message = self._get_next_message_bidi - else: - self._get_next_message = self._get_next_message_classic - - self.close_after_done = close_after_done - self.cleanup_after_test = cleanup_after_test - - def is_alive(self): - return self.protocol.is_alive() - - def on_environment_change(self, new_environment): - if new_environment["protocol"] != self.last_environment["protocol"]: - self.protocol.testharness.load_runner(new_environment["protocol"]) - - def do_test(self, test): - url = self.test_url(test) - - success, data = WebDriverRun(self.logger, - self.do_testharness, - self.protocol, - url, - test.timeout * self.timeout_multiplier, - self.extra_timeout).run() - - if success: - data, extra = data - return self.convert_result(test, data, extra=extra) - - return (test.make_result(*data), []) - - 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. - protocol.testharness.close_old_windows() +# TODO(web-platform-tests/wpt#13183): Add testdriver support to the other +# executors. +class TestDriverExecutorMixin: + def __init__(self, script_resume: str): + self.script_resume = script_resume + def run_testdriver(self, protocol, url, timeout): # 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 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) - # Exceptions occurred outside the main loop. unexpected_exceptions = [] @@ -896,36 +843,14 @@ async def process_bidi_event(method, params): # Use protocol loop to run the async cleanup. protocol.loop.run_until_complete(protocol.bidi_events.unsubscribe_all()) - extra = {} - if leak_part := getattr(protocol, "leak", None): - testharness_window = protocol.base.current_window - extra_windows = set(protocol.base.window_handles()) - extra_windows -= {protocol.testharness.runner_handle, testharness_window} - protocol.testharness.close_windows(extra_windows) - try: - protocol.base.set_window(testharness_window) - if counters := leak_part.check(): - extra["leak_counters"] = counters - except webdriver_error.NoSuchWindowException: - pass - finally: - protocol.base.set_window(protocol.testharness.runner_handle) - - # Attempt to clean up any leftover windows, if allowed. This is - # preferable as it will blame the correct test if something goes wrong - # closing windows, but if the user wants to see the test results we - # have to leave the window(s) open. - if self.cleanup_after_test: - protocol.testharness.close_old_windows() - if len(unexpected_exceptions) > 0: # TODO: what to do if there are more then 1 unexpected exceptions? raise unexpected_exceptions[0] - return rv, extra + return rv def get_or_create_test_window(self, protocol): - return protocol.base.create_window() + return protocol.base.current_window def _get_next_message_classic(self, protocol, url, _): """ @@ -969,6 +894,99 @@ def _get_next_message_bidi(self, protocol, url, test_window): return deserialized_message +class WebDriverTestharnessExecutor(TestharnessExecutor, TestDriverExecutorMixin): + supports_testdriver = True + protocol_cls = WebDriverProtocol + _get_next_message = None + + def __init__(self, logger, browser, server_config, timeout_multiplier=1, + close_after_done=True, capabilities=None, debug_info=None, + cleanup_after_test=True, **kwargs): + """WebDriver-based executor for testharness.js tests""" + TestharnessExecutor.__init__(self, logger, browser, server_config, + timeout_multiplier=timeout_multiplier, + debug_info=debug_info) + self.protocol = self.protocol_cls(self, browser, capabilities) + with open(os.path.join(here, "testharness_webdriver_resume.js")) as f: + script_resume = f.read() + TestDriverExecutorMixin.__init__(self, script_resume) + with open(os.path.join(here, "window-loaded.js")) as f: + self.window_loaded_script = f.read() + + if hasattr(self.protocol, 'bidi_script'): + # If `bidi_script` is available, the messages can be handled via BiDi. + self._get_next_message = self._get_next_message_bidi + else: + self._get_next_message = self._get_next_message_classic + + self.close_after_done = close_after_done + self.cleanup_after_test = cleanup_after_test + + def is_alive(self): + return self.protocol.is_alive() + + def on_environment_change(self, new_environment): + if new_environment["protocol"] != self.last_environment["protocol"]: + self.protocol.testharness.load_runner(new_environment["protocol"]) + + def do_test(self, test): + url = self.test_url(test) + + success, data = WebDriverRun(self.logger, + self.do_testharness, + self.protocol, + url, + test.timeout * self.timeout_multiplier, + self.extra_timeout).run() + + if success: + data, extra = data + return self.convert_result(test, data, extra=extra) + + return (test.make_result(*data), []) + + def do_testharness(self, protocol, url, timeout): + try: + # 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. + protocol.testharness.close_old_windows() + raw_results = self.run_testdriver(protocol, url, timeout) + extra = {} + if counters := self._check_for_leaks(protocol): + extra["leak_counters"] = counters + return raw_results, extra + finally: + # Attempt to clean up any leftover windows, if allowed. This is + # preferable as it will blame the correct test if something goes + # wrong closing windows, but if the user wants to see the test + # results we have to leave the window(s) open. + if self.cleanup_after_test: + protocol.testharness.close_old_windows() + + def _check_for_leaks(self, protocol): + leak_part = getattr(protocol, "leak", None) + if not leak_part: + return None + testharness_window = protocol.base.current_window + extra_windows = set(protocol.base.window_handles()) + extra_windows -= {protocol.testharness.runner_handle, testharness_window} + protocol.testharness.close_windows(extra_windows) + try: + protocol.base.set_window(testharness_window) + return leak_part.check() + except webdriver_error.NoSuchWindowException: + pass + finally: + protocol.base.set_window(protocol.testharness.runner_handle) + + def get_or_create_test_window(self, protocol): + test_window = protocol.base.create_window() + protocol.base.set_window(test_window) + # Wait until about:blank has been loaded + protocol.base.execute_script(self.window_loaded_script, asynchronous=True) + return test_window + + class WebDriverRefTestExecutor(RefTestExecutor): protocol_cls = WebDriverProtocol