format.sh

Author: shbenzer

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

@@ -16,35 +16,35 @@
 # under the License.
 
 import uuid
+
 from .session import session_subscribe
 from .session import session_unsubscribe
 
 
 class Network:
     EVENTS = {
-        'before_request': 'network.beforeRequestSent',
-        'response_started': 'network.responseStarted',
-        'response_completed': 'network.responseCompleted',
-        'auth_required': 'network.authRequired',
-        'fetch_error': 'network.fetchError'
+        "before_request": "network.beforeRequestSent",
+        "response_started": "network.responseStarted",
+        "response_completed": "network.responseCompleted",
+        "auth_required": "network.authRequired",
+        "fetch_error": "network.fetchError",
     }
 
     PHASES = {
-        'before_request': 'beforeRequestSent',
-        'response_started': 'responseStarted',
-        'auth_required': 'authRequired'
+        "before_request": "beforeRequestSent",
+        "response_started": "responseStarted",
+        "auth_required": "authRequired",
     }
 
     def __init__(self, conn):
         self.conn = conn
         self.callbacks = {}
         self.subscriptions = {
-            'network.responseStarted': [],
-            'network.beforeRequestSent': [],
-            'network.authRequired': []
+            "network.responseStarted": [],
+            "network.beforeRequestSent": [],
+            "network.authRequired": [],
         }
 
-
     def command_iterator(self, command):
         """Generator to yield command."""
         yield command.to_json()
@@ -59,50 +59,37 @@ def __add_intercept(self, phases=None, contexts=None, url_patterns=None):
             phases = []
         if contexts is None and url_patterns is None:
             params = {
-                'phases': phases,
+                "phases": phases,
             }
         elif contexts is None:
-            params = {
-                'phases': phases,
-                'urlPatterns': url_patterns
-            }
+            params = {"phases": phases, "urlPatterns": url_patterns}
         elif url_patterns is None:
-            params = {
-                'phases': phases,
-                'contexts': contexts
-            }
+            params = {"phases": phases, "contexts": contexts}
         else:
-            params = {
-                'phases': phases,
-                'contexts': contexts,
-                'urlPatterns': url_patterns
-            }
-        command = {'method': 'network.addIntercept', 'params': params}
+            params = {"phases": phases, "contexts": contexts, "urlPatterns": url_patterns}
+        command = {"method": "network.addIntercept", "params": params}
         self.conn.execute(self.command_iterator(command))
 
     def __remove_intercept(self, intercept=None, request_id=None):
         """Remove an intercept from the network."""
         if request_id is not None:
-            command = {'method': 'network.removeIntercept', 'requestId': request_id}
+            command = {"method": "network.removeIntercept", "requestId": request_id}
             self.conn.execute(self.command_iterator(command))
         elif intercept is not None:
-            command = {'method': 'network.removeIntercept', 'intercept': intercept}
+            command = {"method": "network.removeIntercept", "intercept": intercept}
             self.conn.execute(self.command_iterator(command))
         else:
-            raise ValueError('Either requestId or intercept must be specified')
+            raise ValueError("Either requestId or intercept must be specified")
 
     def __continue_with_auth(self, request_id, username, password):
         """Continue with authentication."""
-        command = {'method': 'network.continueWithAuth', 'params': 
-                    {
-                        'request': request_id,
-                        'action': 'provideCredentials',
-                        'credentials': {
-                            'type': 'password',
-                            'username': username,
-                            'password': password
-                        }
-                    }
+        command = {
+            "method": "network.continueWithAuth",
+            "params": {
+                "request": request_id,
+                "action": "provideCredentials",
+                "credentials": {"type": "password", "username": username, "password": password},
+            },
         }
         self.conn.execute(self.command_iterator(command))
 
@@ -122,17 +109,19 @@ def __handle_event(self, event, data):
 
     def add_authentication_handler(self, username, password):
         """Adds an authentication handler."""
-        self.__add_intercept(phases=[self.PHASES['auth_required']])
-        self.__on('auth_required', lambda data: self.__continue_with_auth(data['request']['request'], username, password))
-        self.subscriptions['auth_required'] = [username, password]
+        self.__add_intercept(phases=[self.PHASES["auth_required"]])
+        self.__on(
+            "auth_required", lambda data: self.__continue_with_auth(data["request"]["request"], username, password)
+        )
+        self.subscriptions["auth_required"] = [username, password]
 
     def remove_authentication_handler(self):
         """Removes an authentication handler."""
-        self.__remove_intercept(intercept='auth_required')
-        del self.subscriptions['auth_required']
-        session_unsubscribe(self.conn, self.EVENTS['auth_required'])
+        self.__remove_intercept(intercept="auth_required")
+        del self.subscriptions["auth_required"]
+        session_unsubscribe(self.conn, self.EVENTS["auth_required"])
 
-    def add_request_handler(self, callback, url_pattern=''):
+    def add_request_handler(self, callback, url_pattern=""):
         """Adds a request handler that executes a callback function when a
         request matches the given URL pattern.
 
@@ -145,34 +134,36 @@ def add_request_handler(self, callback, url_pattern=''):
         Returns:
             str: The request ID of the intercepted response.
         """
-        self.__add_intercept(phases=[self.PHASES['before_request']])
+        self.__add_intercept(phases=[self.PHASES["before_request"]])
+
         def callback_on_url_match(data):
-            if url_pattern in data['request']['url']:
+            if url_pattern in data["request"]["url"]:
                 # create request object to pass to callback
-                request_id = data['request'].get('requestId', uuid.uuid4())
-                url = data['request'].get('url')
-                method = data['request'].get('method')
-                headers = data['request'].get('headers', {})
-                body = data['request'].get('body', None)
+                request_id = data["request"].get("requestId", uuid.uuid4())
+                url = data["request"].get("url")
+                method = data["request"].get("method")
+                headers = data["request"].get("headers", {})
+                body = data["request"].get("body", None)
                 request = Request(request_id, url, method, headers, body, self)
                 callback(request)
-        request_id = self.__on('before_request', callback_on_url_match)
+
+        request_id = self.__on("before_request", callback_on_url_match)
         self.callbacks[request_id] = callback
-        if 'before_request' not in self.subscriptions or not self.subscriptions.get('before_request'):
-            self.subscriptions['before_request'] = [request_id]
+        if "before_request" not in self.subscriptions or not self.subscriptions.get("before_request"):
+            self.subscriptions["before_request"] = [request_id]
         else:
-            self.subscriptions['before_request'].append(request_id)
+            self.subscriptions["before_request"].append(request_id)
         return request_id
 
     def remove_request_handler(self, request_id):
         """Removes a request handler."""
         self.__remove_intercept(request_id=request_id)
-        self.subscriptions['before_request'].remove(request_id)
+        self.subscriptions["before_request"].remove(request_id)
         del self.callbacks[request_id]
-        if len(self.subscriptions['before_request']) == 0:
-            session_unsubscribe(self.conn, self.EVENTS['before_request']) 
+        if len(self.subscriptions["before_request"]) == 0:
+            session_unsubscribe(self.conn, self.EVENTS["before_request"])
 
-    def add_response_handler(self, callback, url_pattern=''):
+    def add_response_handler(self, callback, url_pattern=""):
         """Adds a response handler that executes a callback function when a
         response matches the given URL pattern.
 
@@ -185,32 +176,35 @@ def add_response_handler(self, callback, url_pattern=''):
         Returns:
             str: The request ID of the intercepted response.
         """
-        self.__add_intercept(phases=[self.PHASES['response_started']])
+        self.__add_intercept(phases=[self.PHASES["response_started"]])
+
         def callback_on_url_match(data):
             # create response object to pass to callback
-            if url_pattern in data['response']['url']:
-                request_id = data['request'].get('requestId', uuid.uuid4())
-                url = data['response'].get('url')
-                status_code = data['response'].get('status')
-                body = data['response'].get('body', None)
-                headers = data['response'].get('headers', {})
+            if url_pattern in data["response"]["url"]:
+                request_id = data["request"].get("requestId", uuid.uuid4())
+                url = data["response"].get("url")
+                status_code = data["response"].get("status")
+                body = data["response"].get("body", None)
+                headers = data["response"].get("headers", {})
                 response = Response(request_id, url, status_code, headers, body, self)
                 callback(response)
-        request_id = self.__on('response_started', callback_on_url_match)
+
+        request_id = self.__on("response_started", callback_on_url_match)
         self.callbacks[request_id] = callback
-        if 'response_started' not in self.subscriptions or not self.subscriptions.get('response_started'):
-            self.subscriptions['response_started'] = [request_id]
+        if "response_started" not in self.subscriptions or not self.subscriptions.get("response_started"):
+            self.subscriptions["response_started"] = [request_id]
         else:
-            self.subscriptions['response_started'].append(request_id)
+            self.subscriptions["response_started"].append(request_id)
         return request_id
 
     def remove_response_handler(self, response_id):
         """Removes a response handler."""
         self.__remove_intercept(request_id=response_id)
-        self.subscriptions['response_started'].remove(response_id)
+        self.subscriptions["response_started"].remove(response_id)
         del self.callbacks[response_id]
-        if len(self.subscriptions['response_started']) == 0:
-            session_unsubscribe(self.conn, self.EVENTS['response_started']) 
+        if len(self.subscriptions["response_started"]) == 0:
+            session_unsubscribe(self.conn, self.EVENTS["response_started"])
+
 
 class Request:
     def __init__(self, request_id, url, method, headers, body, network: Network):
@@ -223,24 +217,23 @@ def __init__(self, request_id, url, method, headers, body, network: Network):
 
     def continue_request(self):
         """Continue after sending a request."""
-        params = {
-            'requestId': self.request_id
-        }
+        params = {"requestId": self.request_id}
         if self.url is not None:
-            params['url'] = self.url
+            params["url"] = self.url
         if self.method is not None:
-            params['method'] = self.method
+            params["method"] = self.method
         if self.headers is not None:
-            params['headers'] = self.headers
+            params["headers"] = self.headers
         if self.body is not None:
-            params['body'] = self.body
-        command = {'method': 'network.continueRequest', 'params': params}
+            params["body"] = self.body
+        command = {"method": "network.continueRequest", "params": params}
         self.network.conn.execute(self.command_iterator(command))
 
     def command_iterator(self, command):
         """Generator to yield command."""
         yield command.to_json()
 
+
 class Response:
     def __init__(self, request_id, url, status_code, headers, body, network: Network):
         self.request_id = request_id
@@ -252,15 +245,12 @@ def __init__(self, request_id, url, status_code, headers, body, network: Network
 
     def continue_response(self):
         """Continue after receiving a response."""
-        params = {
-            'requestId': self.request_id,
-            'status': self.status_code
-        }
+        params = {"requestId": self.request_id, "status": self.status_code}
         if self.headers is not None:
-            params['headers'] = self.headers
+            params["headers"] = self.headers
         if self.body is not None:
-            params['body'] = self.body
-        command = {'method': 'network.continueResponse', 'params': params}
+            params["body"] = self.body
+        command = {"method": "network.continueResponse", "params": params}
         self.network.conn.execute(self.command_iterator(command))
 
     def command_iterator(self, command):

py/selenium/webdriver/remote/webdriver.py

@@ -1252,7 +1252,7 @@ def network(self):
         if not self._websocket_connection:
             self._start_bidi()
 
-        if not hasattr(self, '_network') or self._network is None:
+        if not hasattr(self, "_network") or self._network is None:
             self._network = Network(self._websocket_connection)
 
         return self._network

py/selenium/webdriver/remote/websocket_connection.py

@@ -70,7 +70,7 @@ def execute(self, command):
             return self._deserialize_result(result, command)
 
     def add_callback(self, event, callback):
-        event_name = event.event_class if hasattr(event, 'event_class') else event
+        event_name = event.event_class if hasattr(event, "event_class") else event
         if event_name not in self.callbacks:
             self.callbacks[event_name] = []
 

py/test/selenium/webdriver/common/bidi_network_tests.py

@@ -24,6 +24,7 @@
 def test_network_initialized(driver):
     assert driver.network is not None
 
+
 @pytest.mark.xfail_safari
 def test_add_response_handler(driver, pages):
     passed = [False]
@@ -36,6 +37,7 @@ def callback(response):
     pages.load("basicAuth")
     assert passed[0], "Callback was NOT successful"
 
+
 @pytest.mark.xfail_safari
 def test_remove_response_handler(driver, pages):
     passed = [False]
@@ -49,6 +51,7 @@ def callback(response):
     pages.load("basicAuth")
     assert not passed[0], "Callback should NOT be successful"
 
+
 @pytest.mark.xfail_safari
 def test_add_request_handler(driver, pages):
     passed = [False]
@@ -61,6 +64,7 @@ def callback(request):
     pages.load("basicAuth")
     assert passed[0], "Callback was NOT successful"
 
+
 @pytest.mark.xfail_safari
 def test_remove_request_handler(driver, pages):
     passed = [False]
@@ -74,15 +78,17 @@ def callback(request):
     pages.load("basicAuth")
     assert not passed[0], "Callback should NOT be successful"
 
+
 @pytest.mark.xfail_safari
 def test_add_authentication_handler(driver, pages):
-    driver.network.add_authentication_handler('test','test')
+    driver.network.add_authentication_handler("test", "test")
     pages.load("basicAuth")
-    assert driver.find_element(By.TAG_NAME, 'h1').text == 'authorized', "Authentication was NOT successful"
+    assert driver.find_element(By.TAG_NAME, "h1").text == "authorized", "Authentication was NOT successful"
+
 
 @pytest.mark.xfail_safari
 def test_remove_authentication_handler(driver, pages):
-    driver.network.add_authentication_handler('test', 'test')
+    driver.network.add_authentication_handler("test", "test")
     driver.network.remove_authentication_handler()
     pages.load("basicAuth")
-    assert driver.find_element(By.TAG_NAME, 'h1').text != 'authorized', "Authentication was successful"
+    assert driver.find_element(By.TAG_NAME, "h1").text != "authorized", "Authentication was successful"