move command_builder to common.py to avoid function duplication

Author: navin772

py/selenium/webdriver/common/bidi/browser.py

@@ -18,6 +18,8 @@
 from typing import Dict
 from typing import List
 
+from selenium.webdriver.common.bidi.common import command_builder
+
 
 class ClientWindowState:
     """Represents a window state."""
@@ -143,29 +145,14 @@ class Browser:
     def __init__(self, conn):
         self.conn = conn
 
-    def command_builder(self, method: str, params: Dict = None) -> Dict:
-        """Build a command iterator to send to the browser.
-
-        Parameters:
-        -----------
-            method: The method to execute.
-            params: The parameters to pass to the method. Default is None.
-        """
-        if params is None:
-            params = {}
-
-        command = {"method": method, "params": params}
-        cmd = yield command
-        return cmd
-
     def create_user_context(self) -> str:
         """Creates a new user context.
 
         Returns:
         -------
             str: The ID of the created user context.
         """
-        result = self.conn.execute(self.command_builder("browser.createUserContext", {}))
+        result = self.conn.execute(command_builder("browser.createUserContext", {}))
         return result["userContext"]
 
     def get_user_contexts(self) -> List[str]:
@@ -175,7 +162,7 @@ def get_user_contexts(self) -> List[str]:
         -------
             List[str]: A list of user context IDs.
         """
-        result = self.conn.execute(self.command_builder("browser.getUserContexts", {}))
+        result = self.conn.execute(command_builder("browser.getUserContexts", {}))
         return [context_info["userContext"] for context_info in result["userContexts"]]
 
     def remove_user_context(self, user_context_id: str) -> None:
@@ -190,7 +177,7 @@ def remove_user_context(self, user_context_id: str) -> None:
             Exception: If the user context ID is "default" or does not exist.
         """
         params = {"userContext": user_context_id}
-        self.conn.execute(self.command_builder("browser.removeUserContext", params))
+        self.conn.execute(command_builder("browser.removeUserContext", params))
 
     def get_client_windows(self) -> List[ClientWindowInfo]:
         """Gets all client windows.
@@ -199,5 +186,5 @@ def get_client_windows(self) -> List[ClientWindowInfo]:
         -------
             List[ClientWindowInfo]: A list of client window information.
         """
-        result = self.conn.execute(self.command_builder("browser.getClientWindows", {}))
+        result = self.conn.execute(command_builder("browser.getClientWindows", {}))
         return [ClientWindowInfo.from_dict(window) for window in result["clientWindows"]]

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

@@ -0,0 +1,38 @@
+# 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.
+
+from typing import Dict
+
+
+def command_builder(method: str, params: Dict = None) -> Dict:
+    """Build a command iterator to send to the BiDi protocol.
+
+    Parameters:
+    -----------
+        method: The method to execute.
+        params: The parameters to pass to the method. Default is None.
+
+    Returns:
+    --------
+        The response from the command execution.
+    """
+    if params is None:
+        params = {}
+
+    command = {"method": method, "params": params}
+    cmd = yield command
+    return cmd

py/selenium/webdriver/common/bidi/network.py

@@ -15,6 +15,8 @@
 # specific language governing permissions and limitations
 # under the License.
 
+from selenium.webdriver.common.bidi.common import command_builder
+
 
 class NetworkEvent:
     """Represents a network event."""
@@ -51,18 +53,6 @@ def __init__(self, conn):
         self.callbacks = {}
         self.subscriptions = {}
 
-    def command_builder(self, method, params):
-        """Build a command iterator to send to the network.
-
-        Parameters:
-        ----------
-            method (str): The method to execute.
-            params (dict): The parameters to pass to the method.
-        """
-        command = {"method": method, "params": params}
-        cmd = yield command
-        return cmd
-
     def _add_intercept(self, phases=[], contexts=None, url_patterns=None):
         """Add an intercept to the network.
 
@@ -88,7 +78,7 @@ def _add_intercept(self, phases=[], contexts=None, url_patterns=None):
             params["phases"] = phases
         else:
             params["phases"] = ["beforeRequestSent"]
-        cmd = self.command_builder("network.addIntercept", params)
+        cmd = command_builder("network.addIntercept", params)
 
         result = self.conn.execute(cmd)
         self.intercepts.append(result["intercept"])
@@ -113,11 +103,11 @@ def _remove_intercept(self, intercept=None):
         if intercept is None:
             intercepts_to_remove = self.intercepts.copy()  # create a copy before iterating
             for intercept_id in intercepts_to_remove:  # remove all intercepts
-                self.conn.execute(self.command_builder("network.removeIntercept", {"intercept": intercept_id}))
+                self.conn.execute(command_builder("network.removeIntercept", {"intercept": intercept_id}))
                 self.intercepts.remove(intercept_id)
         else:
             try:
-                self.conn.execute(self.command_builder("network.removeIntercept", {"intercept": intercept}))
+                self.conn.execute(command_builder("network.removeIntercept", {"intercept": intercept}))
                 self.intercepts.remove(intercept)
             except Exception as e:
                 raise Exception(f"Exception: {e}")
@@ -192,7 +182,7 @@ def add_request_handler(self, event, callback, url_patterns=None, contexts=None)
         else:
             params = {}
             params["events"] = [event_name]
-            self.conn.execute(self.command_builder("session.subscribe", params))
+            self.conn.execute(command_builder("session.subscribe", params))
             self.subscriptions[event_name] = [callback_id]
 
         self.callbacks[callback_id] = result["intercept"]
@@ -220,7 +210,7 @@ def remove_request_handler(self, event, callback_id):
         if len(self.subscriptions[event_name]) == 0:
             params = {}
             params["events"] = [event_name]
-            self.conn.execute(self.command_builder("session.unsubscribe", params))
+            self.conn.execute(command_builder("session.unsubscribe", params))
             del self.subscriptions[event_name]
 
     def clear_request_handlers(self):
@@ -234,7 +224,7 @@ def clear_request_handlers(self):
                 del self.callbacks[callback_id]
             params = {}
             params["events"] = [event_name]
-            self.conn.execute(self.command_builder("session.unsubscribe", params))
+            self.conn.execute(command_builder("session.unsubscribe", params))
         self.subscriptions = {}
 
     def add_auth_handler(self, username, password):
@@ -294,26 +284,14 @@ def __init__(
         self.timings = timings
         self.url = url
 
-    def command_builder(self, method, params):
-        """Build a command iterator to send to the network.
-
-        Parameters:
-        ----------
-            method (str): The method to execute.
-            params (dict): The parameters to pass to the method.
-        """
-        command = {"method": method, "params": params}
-        cmd = yield command
-        return cmd
-
     def fail_request(self):
         """Fail this request."""
 
         if not self.request_id:
             raise ValueError("Request not found.")
 
         params = {"request": self.request_id}
-        self.network.conn.execute(self.command_builder("network.failRequest", params))
+        self.network.conn.execute(command_builder("network.failRequest", params))
 
     def continue_request(self, body=None, method=None, headers=None, cookies=None, url=None):
         """Continue after intercepting this request."""
@@ -333,7 +311,7 @@ def continue_request(self, body=None, method=None, headers=None, cookies=None, u
         if url is not None:
             params["url"] = url
 
-        self.network.conn.execute(self.command_builder("network.continueRequest", params))
+        self.network.conn.execute(command_builder("network.continueRequest", params))
 
     def _continue_with_auth(self, username=None, password=None):
         """Continue with authentication.
@@ -358,4 +336,4 @@ def _continue_with_auth(self, username=None, password=None):
             params["action"] = "provideCredentials"
             params["credentials"] = {"type": "password", "username": username, "password": password}
 
-        self.network.conn.execute(self.command_builder("network.continueWithAuth", params))
+        self.network.conn.execute(command_builder("network.continueWithAuth", params))