diff --git a/py/selenium/webdriver/common/bidi/script.py b/py/selenium/webdriver/common/bidi/script.py index 1dc8d101d670e..cfefee96fce75 100644 --- a/py/selenium/webdriver/common/bidi/script.py +++ b/py/selenium/webdriver/common/bidi/script.py @@ -18,8 +18,7 @@ from dataclasses import dataclass from typing import List -from .session import session_subscribe -from .session import session_unsubscribe +from .session import Session class Script: @@ -43,12 +42,14 @@ def remove_console_message_handler(self, id): def _subscribe_to_log_entries(self): if not self.log_entry_subscribed: - self.conn.execute(session_subscribe(LogEntryAdded.event_class)) + session = Session(self.conn) + 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)) + session = Session(self.conn) + self.conn.execute(session.unsubscribe(LogEntryAdded.event_class)) self.log_entry_subscribed = False def _handle_log_entry(self, type, handler): diff --git a/py/selenium/webdriver/common/bidi/session.py b/py/selenium/webdriver/common/bidi/session.py index dbe5d26644a87..693f2b4931ba4 100644 --- a/py/selenium/webdriver/common/bidi/session.py +++ b/py/selenium/webdriver/common/bidi/session.py @@ -15,32 +15,43 @@ # specific language governing permissions and limitations # under the License. +from selenium.webdriver.common.bidi.common import command_builder -def session_subscribe(*events, browsing_contexts=None): - cmd_dict = { - "method": "session.subscribe", - "params": { + +class Session: + + def __init__(self, conn): + self.conn = conn + + def subscribe(self, *events, browsing_contexts=None): + params = { "events": events, - }, - } - if browsing_contexts is None: - browsing_contexts = [] - if browsing_contexts: - cmd_dict["params"]["browsingContexts"] = browsing_contexts - _ = yield cmd_dict - return None - - -def session_unsubscribe(*events, browsing_contexts=None): - cmd_dict = { - "method": "session.unsubscribe", - "params": { + } + if browsing_contexts is None: + browsing_contexts = [] + if browsing_contexts: + params["browsingContexts"] = browsing_contexts + return command_builder("session.subscribe", params) + + def unsubscribe(self, *events, browsing_contexts=None): + params = { "events": events, - }, - } - if browsing_contexts is None: - browsing_contexts = [] - if browsing_contexts: - cmd_dict["params"]["browsingContexts"] = browsing_contexts - _ = yield cmd_dict - return None + } + if browsing_contexts is None: + browsing_contexts = [] + if browsing_contexts: + params["browsingContexts"] = browsing_contexts + return command_builder("session.unsubscribe", params) + + def status(self): + """ + The session.status command returns information about the remote end's readiness + to create new sessions and may include implementation-specific metadata. + + Returns + ------- + dict + Dictionary containing the ready state (bool), message (str) and metadata + """ + cmd = command_builder("session.status", {}) + return self.conn.execute(cmd) diff --git a/py/selenium/webdriver/remote/webdriver.py b/py/selenium/webdriver/remote/webdriver.py index 2c192673a7089..0fe824c62f444 100644 --- a/py/selenium/webdriver/remote/webdriver.py +++ b/py/selenium/webdriver/remote/webdriver.py @@ -44,6 +44,7 @@ from selenium.webdriver.common.bidi.browser import Browser from selenium.webdriver.common.bidi.network import Network from selenium.webdriver.common.bidi.script import Script +from selenium.webdriver.common.bidi.session import Session from selenium.webdriver.common.by import By from selenium.webdriver.common.options import ArgOptions from selenium.webdriver.common.options import BaseOptions @@ -256,6 +257,7 @@ def __init__( self._script = None self._network = None self._browser = None + self._bidi_session = None def __repr__(self): return f'<{type(self).__module__}.{type(self).__name__} (session="{self.session_id}")>' @@ -1294,6 +1296,19 @@ def browser(self): return self._browser + @property + def _session(self): + """ + Returns the BiDi session object for the current WebDriver session. + """ + if not self._websocket_connection: + self._start_bidi() + + if self._bidi_session is None: + self._bidi_session = Session(self._websocket_connection) + + return self._bidi_session + def _get_cdp_details(self): import json diff --git a/py/test/selenium/webdriver/common/bidi_session_tests.py b/py/test/selenium/webdriver/common/bidi_session_tests.py new file mode 100644 index 0000000000000..8bb59a1751c30 --- /dev/null +++ b/py/test/selenium/webdriver/common/bidi_session_tests.py @@ -0,0 +1,50 @@ +# 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 pytest + +from selenium.webdriver.common.window import WindowTypes + + +@pytest.mark.xfail_safari +def test_session_status(driver): + result = driver._session.status() + assert result is not None + assert "ready" in result + assert "message" in result + assert isinstance(result["ready"], bool) + assert isinstance(result["message"], str) + + +@pytest.mark.xfail_safari +def test_session_status_not_closed_with_one_window(driver): + # initial session status + initial_status = driver._session.status() + assert initial_status is not None + + # Open new window and tab + driver.switch_to.new_window(WindowTypes.WINDOW) + driver.switch_to.new_window(WindowTypes.TAB) + + # Close one window + driver.close() + + # Session should still be active + status_after_closing = driver._session.status() + assert status_after_closing is not None + assert "ready" in status_after_closing + assert "message" in status_after_closing