[py][bidi]: add bidi command session.status and encapsulate in Session class (#15615)

* add bidi command session status
* use Session class for session module
* make webdriver bidi session private

Author: navin772

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):

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)

py/selenium/webdriver/remote/webdriver.py

@@ -46,6 +46,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
@@ -262,6 +263,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}")>'
@@ -1277,6 +1279,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
 

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