[py] Implement High Level Logging API with BiDi

.gitignore

@@ -76,6 +76,7 @@ py/selenium/webdriver/remote/isDisplayed.js
 py/docs/build/
 py/build/
 py/LICENSE
+py/pytestdebug.log
 selenium.egg-info/
 third_party/java/jetty/jetty-repacked.jar
 *.user

py/BUILD.bazel

@@ -214,6 +214,7 @@ py_package(
  "py.selenium.webdriver.chrome",
  "py.selenium.webdriver.chromium",
  "py.selenium.webdriver.common",
+ "py.selenium.webdriver.common.bidi",
  "py.selenium.webdriver.common.devtools",
  "py.selenium.webdriver.edge",
  "py.selenium.webdriver.firefox",
@@ -380,10 +381,39 @@ py_library(
  deps = [],
 )
 
+BIDI_TESTS = glob(["test/selenium/webdriver/common/**/*bidi*_tests.py"])
+
 [
  py_test_suite(
  name = "common-%s" % browser,
  size = "large",
+ srcs = glob(
+ [
+ "test/selenium/webdriver/common/**/*.py",
+ "test/selenium/webdriver/support/**/*.py",
+ ],
+ exclude = BIDI_TESTS + ["test/selenium/webdriver/common/print_pdf_tests.py"],
+ ),
+ args = [
+ "--instafail",
+ "--bidi=false",
+ ] + BROWSERS[browser]["args"],
+ data = BROWSERS[browser]["data"],
+ env_inherit = ["DISPLAY"],
+ tags = ["no-sandbox"] + BROWSERS[browser]["tags"],
+ deps = [
+ ":init-tree",
+ ":selenium",
+ ":webserver",
+ ] + TEST_DEPS,
+ )
+ for browser in BROWSERS.keys()
+]
+
+[
+ py_test_suite(
+ name = "common-%s-bidi" % browser,
+ size = "large",
  srcs = glob(
  [
  "test/selenium/webdriver/common/**/*.py",
@@ -393,12 +423,11 @@ py_library(
  ),
  args = [
  "--instafail",
+ "--bidi=true",
  ] + BROWSERS[browser]["args"],
  data = BROWSERS[browser]["data"],
  env_inherit = ["DISPLAY"],
- tags = [
- "no-sandbox",
- ] + BROWSERS[browser]["tags"],
+ tags = ["no-sandbox"] + BROWSERS[browser]["tags"],
  deps = [
  ":init-tree",
  ":selenium",
@@ -504,10 +533,13 @@ py_test_suite(
 py_test_suite(
  name = "test-remote",
  size = "large",
- srcs = glob([
- "test/selenium/webdriver/common/**/*.py",
- "test/selenium/webdriver/support/**/*.py",
- ]),
+ srcs = glob(
+ [
+ "test/selenium/webdriver/common/**/*.py",
+ "test/selenium/webdriver/support/**/*.py",
+ ],
+ exclude = BIDI_TESTS,
+ ),
  args = [
  "--instafail",
  "--driver=remote",

py/conftest.py

@@ -79,6 +79,13 @@ def pytest_addoption(parser):
  dest="use_lan_ip",
  help="Whether to start test server with lan ip instead of localhost",
  )
+ parser.addoption(
+ "--bidi",
+ action="store",
+ dest="bidi",
+ metavar="BIDI",
+ help="Whether to enable BiDi support",
+ )
 
 
 def pytest_ignore_collect(path, config):
@@ -158,6 +165,20 @@ def fin():
  driver_instance = getattr(webdriver, driver_class)(**kwargs)
  yield driver_instance
 
+ # Close the browser after BiDi tests. Those make event subscriptions
+ # and doesn't seems to be stable enough, causing the flakiness of the
+ # subsequent tests.
+ # Remove this when BiDi implementation and API is stable.
+ if bool(request.config.option.bidi):
+
+ def fin():
+ global driver_instance
+ if driver_instance is not None:
+ driver_instance.quit()
+ driver_instance = None
+
+ request.addfinalizer(fin)
+
  if request.node.get_closest_marker("no_driver_after_test"):
  driver_instance = None
 
@@ -166,6 +187,7 @@ def get_options(driver_class, config):
  browser_path = config.option.binary
  browser_args = config.option.args
  headless = bool(config.option.headless)
+ bidi = bool(config.option.bidi)
  options = None
 
  if browser_path or browser_args:
@@ -187,6 +209,13 @@ def get_options(driver_class, config):
  options.add_argument("--headless=new")
  if driver_class == "Firefox":
  options.add_argument("-headless")
+
+ if bidi:
+ if not options:
+ options = getattr(webdriver, f"{driver_class}Options")()
+
+ options.web_socket_url = True
+
  return options
 
 

py/selenium/webdriver/common/bidi/script.py

@@ -0,0 +1,111 @@
+# Licensed to the Software Freedom Conservancy (SFC) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The SFC licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import typing
+from dataclasses import dataclass
+
+from .session import session_subscribe
+from .session import session_unsubscribe
+
+
+class Script:
+ def __init__(self, conn):
+ self.conn = conn
+ self.log_entry_subscribed = False
+
+ def add_console_message_handler(self, handler):
+ self._subscribe_to_log_entries()
+ return self.conn.add_callback(LogEntryAdded, self._handle_log_entry("console", handler))
+
+ def add_javascript_error_handler(self, handler):
+ self._subscribe_to_log_entries()
+ return self.conn.add_callback(LogEntryAdded, self._handle_log_entry("javascript", handler))
+
+ def remove_console_message_handler(self, id):
+ self.conn.remove_callback(LogEntryAdded, id)
+ self._unsubscribe_from_log_entries()
+
+ remove_javascript_error_handler = remove_console_message_handler
+
+ def _subscribe_to_log_entries(self):
+ if not self.log_entry_subscribed:
+ self.conn.execute(session_subscribe(LogEntryAdded.event_class))
+ self.log_entry_subscribed = True
+
+ def _unsubscribe_from_log_entries(self):
+ if self.log_entry_subscribed and LogEntryAdded.event_class not in self.conn.callbacks:
+ self.conn.execute(session_unsubscribe(LogEntryAdded.event_class))
+ self.log_entry_subscribed = False
+
+ def _handle_log_entry(self, type, handler):
+ def _handle_log_entry(log_entry):
+ if log_entry.type_ == type:
+ handler(log_entry)
+
+ return _handle_log_entry
+
+
+class LogEntryAdded:
+ event_class = "log.entryAdded"
+
+ @classmethod
+ def from_json(cls, json):
+ print(json)
+ if json["type"] == "console":
+ return ConsoleLogEntry.from_json(json)
+ elif json["type"] == "javascript":
+ return JavaScriptLogEntry.from_json(json)
+
+
+@dataclass
+class ConsoleLogEntry:
+ level: str
+ text: str
+ timestamp: str
+ method: str
+ args: typing.List[dict]
+ type_: str
+
+ @classmethod
+ def from_json(cls, json):
+ return cls(
+ level=json["level"],
+ text=json["text"],
+ timestamp=json["timestamp"],
+ method=json["method"],
+ args=json["args"],
+ type_=json["type"],
+ )
+
+
+@dataclass
+class JavaScriptLogEntry:
+ level: str
+ text: str
+ timestamp: str
+ stacktrace: dict
+ type_: str
+
+ @classmethod
+ def from_json(cls, json):
+ return cls(
+ level=json["level"],
+ text=json["text"],
+ timestamp=json["timestamp"],
+ stacktrace=json["stackTrace"],
+ type_=json["type"],
+ )

py/selenium/webdriver/common/bidi/session.py

@@ -0,0 +1,42 @@
+# Licensed to the Software Freedom Conservancy (SFC) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The SFC licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+
+def session_subscribe(*events, browsing_contexts=[]):
+ cmd_dict = {
+ "method": "session.subscribe",
+ "params": {
+ "events": events,
+ },
+ }
+ if browsing_contexts:
+ cmd_dict["params"]["browsingContexts"] = browsing_contexts
+ _ = yield cmd_dict
+ return None
+
+
+def session_unsubscribe(*events, browsing_contexts=[]):
+ cmd_dict = {
+ "method": "session.unsubscribe",
+ "params": {
+ "events": events,
+ },
+ }
+ if browsing_contexts:
+ cmd_dict["params"]["browsingContexts"] = browsing_contexts
+ _ = yield cmd_dict
+ return None

py/selenium/webdriver/common/options.py

@@ -416,6 +416,28 @@ class BaseOptions(metaclass=ABCMeta):
  - `None`
  """
 
+ web_socket_url = _BaseOptionsDescriptor("webSocketUrl")
+ """Gets and Sets WebSocket URL.
+
+ Usage
+ -----
+ - Get
+ - `self.web_socket_url`
+ - Set
+ - `self.web_socket_url` = `value`
+
+ Parameters
+ ----------
+ `value`: `bool`
+
+ Returns
+ -------
+ - Get
+ - `bool`
+ - Set
+ - `None`
+ """
+
  def __init__(self) -> None:
  super().__init__()
  self._caps = self.default_capabilities

py/selenium/webdriver/remote/webdriver.py

@@ -41,6 +41,7 @@
 from selenium.common.exceptions import NoSuchCookieException
 from selenium.common.exceptions import NoSuchElementException
 from selenium.common.exceptions import WebDriverException
+from selenium.webdriver.common.bidi.script import Script
 from selenium.webdriver.common.by import By
 from selenium.webdriver.common.options import BaseOptions
 from selenium.webdriver.common.print_page_options import PrintOptions
@@ -209,7 +210,9 @@ def __init__(
  self._authenticator_id = None
  self.start_client()
  self.start_session(capabilities)
+
  self._websocket_connection = None
+ self._script = None
 
  def __repr__(self):
  return f'<{type(self).__module__}.{type(self).__name__} (session="{self.session_id}")>'
@@ -1067,6 +1070,24 @@ async def bidi_connection(self):
  async with conn.open_session(target_id) as session:
  yield BidiConnection(session, cdp, devtools)
 
+ @property
+ def script(self):
+ if not self._websocket_connection:
+ self._start_bidi()
+
+ if not self._script:
+ self._script = Script(self._websocket_connection)
+
+ return self._script
+
+ def _start_bidi(self):
+ if self.caps.get("webSocketUrl"):
+ ws_url = self.caps.get("webSocketUrl")
+ else:
+ raise WebDriverException("Unable to find url to connect to from capabilities")
+
+ self._websocket_connection = WebSocketConnection(ws_url)
+
  def _get_cdp_details(self):
  import json