[py] BiDi Network implementation of Intercepts and Auth in Python

[shbenzer]

User description

Add implementations for BiDi Network's Auth and Interception to Python

Description

Added network.py:

• Network class
• Event class
• Request class
• add_intercept()
• remove_intercept()
• add_request_handler()
• remove_request_handler()
• continue_request()
• add_auth_handler()
• remove_auth_handler()
• continue_with_auth()
• clear_request_handlers()

Updated websocket_connection.py
Added bidi_network_tests.py

Motivation and Context

Issue #13993

Types of changes

• Bug fix (non-breaking change which fixes an issue)
• New feature (non-breaking change which adds functionality)
• Breaking change (fix or feature that would cause existing functionality to change)

Checklist

• I have read the contributing document.
• My change requires a change to the documentation.
• I have updated the documentation accordingly.
• I have added tests to cover my changes.
• All new and existing tests passed.

---

PR Type

Enhancement, Tests

---

Description

• Added BiDi Network implementation for network interception and authentication.

• Introduced Network, Request, and NetworkEvent classes for handling network events.

• Extended WebDriver to include a network property for BiDi capabilities.

• Added comprehensive tests for BiDi Network functionality, including intercepts and authentication.

---

Changes walkthrough 📝

	Relevant files
Enhancement	network.py Implement BiDi Network functionality for intercepts and authentication py/selenium/webdriver/common/bidi/network.py Implemented Network class for network interception and authentication. Added Request and NetworkEvent classes for handling network requests and events. Provided methods for adding/removing intercepts, request handlers, and authentication handlers. Introduced support for continuing requests and handling authentication. +328/-0  webdriver.py Extend WebDriver with BiDi Network integration                      py/selenium/webdriver/remote/webdriver.py Integrated Network class into WebDriver via a new network property. Ensured BiDi WebSocket connection is initialized for network operations. +12/-0
Tests	bidi_network_tests.py Add tests for BiDi Network functionality                                  py/test/selenium/webdriver/common/bidi_network_tests.py Added tests for BiDi Network initialization and intercept functionality. Verified request handling, continuation, and authentication mechanisms. Marked tests as expected to fail on Safari. +76/-0

---

Need help?

Type /help how to ... in the comments thread for any questions about Qodo Merge usage.

Check out the documentation for more information.

[qodo-merge-pro]

PR Reviewer Guide 🔍

(Review updated until commit 00f76a5)

Here are some key observations to aid the review process:

⏱️ Estimated effort to review: 3 🔵🔵🔵⚪⚪

🧪 PR contains tests

🔒 No security concerns identified

⚡ Recommended focus areas for review Potential Bug In the remove_intercept method, when removing all intercepts, modifying the list while iterating over it can cause issues. Consider using a copy of the list for iteration. for intercept_id in self.intercepts: # remove all intercepts self.conn.execute(self.command_builder("network.removeIntercept", {"intercept": intercept_id})) self.intercepts.remove(intercept_id) Incorrect Unsubscription When removing a request handler, the code calls session.subscribe instead of session.unsubscribe, which would keep the subscription active instead of removing it. params["events"] = [event_name] self.conn.execute(self.command_builder("session.subscribe", params)) Missing Parameters The _on_request callback function doesn't handle all possible parameters from the event data, potentially missing important request information like method and headers. request = Request( network=self, request_id=event_data.params["request"].get("request", None), body_size=event_data.params["request"].get("bodySize", None), cookies=event_data.params["request"].get("cookies", None), resource_type=event_data.params["request"].get("goog:resourceType", None), headers_size=event_data.params["request"].get("headersSize", None), timings=event_data.params["request"].get("timings", None), url=event_data.params["request"].get("url", None), ) callback(request)

[qodo-merge-pro]

PR Code Suggestions ✨

Latest suggestions up to 00f76a5

Explore these optional code suggestions:

Category	Suggestion	Impact
Possible issue	✅ Fix incorrect event unsubscription Suggestion Impact: The commit implemented the exact change suggested, replacing session.subscribe with session.unsubscribe when unsubscribing from events with no remaining callbacks code diff: - self.conn.execute(self.command_builder("session.subscribe", params)) + self.conn.execute(self.command_builder("session.unsubscribe", params)) When unsubscribing from an event with no remaining callbacks, you should use session.unsubscribe instead of session.subscribe. The current implementation incorrectly resubscribes to events when trying to unsubscribe. py/selenium/webdriver/common/bidi/network.py [200-223] def remove_request_handler(self, event, callback_id): """Remove a request handler from the network. Parameters: ---------- event_name (str): The event to unsubscribe from. callback_id (int): The callback id to remove. """ try: event_name = self.EVENTS[event] except KeyError: raise Exception(f"Event {event} not found") net_event = NetworkEvent(event_name) self.conn.remove_callback(net_event, callback_id) self.remove_intercept(self.callbacks[callback_id]) del self.callbacks[callback_id] self.subscriptions[event_name].remove(callback_id) if len(self.subscriptions[event_name]) == 0: params = {} params["events"] = [event_name] - self.conn.execute(self.command_builder("session.subscribe", params)) + self.conn.execute(self.command_builder("session.unsubscribe", params)) del self.subscriptions[event_name] [Suggestion has been applied] Suggestion importance[1-10]: 10 __ Why: The code incorrectly calls "session.subscribe" when trying to unsubscribe from events with no remaining callbacks. This is a critical bug that would cause the opposite of the intended behavior, keeping subscriptions active instead of removing them.	High
	✅ Fix list modification during iteration Suggestion Impact: The commit implemented the exact suggestion by changing the code to create a copy of the self.intercepts list before iterating over it, preventing potential issues with modifying a list during iteration code diff: - for intercept_id in self.intercepts: # remove all intercepts + intercepts_to_remove = self.intercepts.copy() # create a copy before iterating + for intercept_id in intercepts_to_remove: # remove all intercepts The current implementation modifies the self.intercepts list while iterating over it, which can cause items to be skipped during removal. Create a copy of the list before iterating to avoid this issue. py/selenium/webdriver/common/bidi/network.py [97-116] def remove_intercept(self, intercept=None): """Remove a specific intercept, or all intercepts. Parameters: ---------- intercept (str, optional): The intercept to remove. Default is None. Raises: ------ Exception: If intercept is not found. Notes: ----- If intercept is None, all intercepts will be removed. """ if intercept is None: - for intercept_id in self.intercepts: # remove all intercepts + 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.intercepts.remove(intercept_id) [Suggestion has been applied] Suggestion importance[1-10]: 9 __ Why: Modifying a list while iterating over it can cause elements to be skipped, leading to a potential bug where not all intercepts are removed. Creating a copy before iteration is a critical fix to ensure proper cleanup.	High
	Add missing request fields The _on_request method doesn't extract the method and headers fields from the request data, but these fields are used in the continue_request method. Add these missing fields to ensure proper request continuation. py/selenium/webdriver/common/bidi/network.py [124-151] def _on_request(self, event_name, callback): """Set a callback function to subscribe to a network event. Parameters: ---------- event_name (str): The event to subscribe to. callback (function): The callback function to execute on event. Takes Request object as argument. Returns: ------- int : callback id """ event = NetworkEvent(event_name) def _callback(event_data): request = Request( network=self, request_id=event_data.params["request"].get("request", None), body_size=event_data.params["request"].get("bodySize", None), cookies=event_data.params["request"].get("cookies", None), resource_type=event_data.params["request"].get("goog:resourceType", None), + headers=event_data.params["request"].get("headers", None), headers_size=event_data.params["request"].get("headersSize", None), + method=event_data.params["request"].get("method", None), timings=event_data.params["request"].get("timings", None), url=event_data.params["request"].get("url", None), ) callback(request) Apply this suggestion Suggestion importance[1-10]: 8 __ Why: The method doesn't extract 'method' and 'headers' fields from request data, but these fields are used in the continue_request method. This omission would cause issues when trying to continue intercepted requests with the original parameters.	Medium
Learned best practice	Add null validation for required parameters before accessing their attributes to prevent potential NoneType errors The method doesn't validate that the request parameter is not None before accessing its request_id attribute. This could lead to a NoneType attribute error if a null request is passed. Add a null check at the beginning of the method to validate the required parameter. py/selenium/webdriver/common/bidi/network.py [244-259] def _continue_with_auth(self, request, username=None, password=None): """Continue with authentication. Parameters: ---------- request (Request): The request to continue with. username (str): The username to authenticate with. password (str): The password to authenticate with. Notes: ----- If username or password is None, it attempts auth with no credentials """ + if request is None: + raise ValueError("Request cannot be None") params = {} params["request"] = request.request_id Apply this suggestion Suggestion importance[1-10]: 6	Low
Update

---

Previous suggestions

✅ Suggestions up to commit 2b3bb2a

Category	Suggestion	Score
Possible issue	✅ Add missing import statement for a module used in the code Suggestion Impact: The uuid import statement was added at the top of the file as suggested code diff: +import uuid Import the missing 'uuid' module at the top of the file, as it's used in request and response handlers for generating unique IDs. py/selenium/webdriver/common/bidi/network.py [150] +import uuid +... request_id = data['request'].get('requestId', uuid.uuid4()) Suggestion importance[1-10]: 8 Why: The code uses uuid.uuid4() but the uuid module is not imported, which would cause a NameError at runtime. This is a critical issue that would break the request/response handling functionality.	8
General	✅ Remove inconsistent subscription handling to prevent potential state issues Suggestion Impact: The commit implemented a more comprehensive solution by completely refactoring the subscription handling logic. Instead of just uncommenting the session_subscribe line, it created a new implementation where both subscribe and unsubscribe are properly handled in the new code structure. code diff: + params = {} + params["events"] = [event_name] + self.conn.execute(self.command_builder("session.subscribe", params)) + self.subscriptions[event_name] = [callback_id] + + self.callbacks[callback_id] = result["intercept"] + return callback_id + + def remove_request_handler(self, event, callback_id): + """Remove a request handler from the network. + + Parameters: + ---------- + event_name (str): The event to unsubscribe from. + callback_id (int): The callback id to remove. + """ + try: + event_name = self.EVENTS[event] + except KeyError: + raise Exception(f"Event {event} not found") + + net_event = NetworkEvent(event_name) + + self.conn.remove_callback(net_event, callback_id) + self.remove_intercept(self.callbacks[callback_id]) + del self.callbacks[callback_id] + self.subscriptions[event_name].remove(callback_id) + if len(self.subscriptions[event_name]) == 0: + params = {} + params["events"] = [event_name] + self.conn.execute(self.command_builder("session.subscribe", params)) + del self.subscriptions[event_name] The session_subscribe call is commented out but the session_unsubscribe is still active, which could lead to inconsistent subscription state - either uncomment the subscribe call or remove both. py/selenium/webdriver/common/bidi/network.py [113-114] -# session_subscribe(self.conn, event, self.__handle_event) return self.conn.add_callback(event, self.__handle_event) [Suggestion has been applied] Suggestion importance[1-10]: 7 Why: The inconsistency between commented subscribe and active unsubscribe calls could lead to subscription state issues. This needs to be resolved to ensure proper event handling.	7
Security	✅ Add validation for authentication credentials to prevent silent failures Suggestion Impact: The commit implemented credential validation in the _continue_with_auth method, but in a different way than suggested. The code was completely refactored, and the validation was implemented in the new structure. code diff: + def _continue_with_auth(self, request, username, password): """Continue with authentication.""" - command = {'method': 'network.continueWithAuth', 'params': - { - 'request': request_id, - 'action': 'provideCredentials', - 'credentials': { - 'type': 'password', - 'username': username, - 'password': password - } - } - } - self.conn.execute(self.command_iterator(command)) - - def __on(self, event, callback): - """Set a callback function to subscribe to a network event.""" - event = self.EVENTS.get(event, event) - self.callbacks[event] = callback - if len(self.subscriptions[event]) == 0: - # session_subscribe(self.conn, event, self.__handle_event) - return self.conn.add_callback(event, self.__handle_event) - - def __handle_event(self, event, data): - """Perform callback function on event.""" - if event in self.callbacks: - self.callbacks[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] - - 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']) - - 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. - - Parameters: - callback (function): A function to be executed when url is matched by a URL pattern - The callback function receives a `Response` object as its argument. - url_pattern (str, optional): A substring to match against the response URL. - Default is an empty string, which matches all URLs. - - Returns: - str: The request ID of the intercepted response. - """ - self.__add_intercept(phases=[self.PHASES['before_request']]) - def callback_on_url_match(data): - 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 = Request(request_id, url, method, headers, body, self) - callback(request) - 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] - else: - 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) - del self.callbacks[request_id] - if len(self.subscriptions['before_request']) == 0: - session_unsubscribe(self.conn, self.EVENTS['before_request']) - - 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. - - Parameters: - callback (function): A function to be executed when url is matched by a url_pattern - The callback function receives a `Response` object as its argument. - url_pattern (str, optional): A substring to match against the response URL. - Default is an empty string, which matches all URLs. - - Returns: - str: The request ID of the intercepted response. - """ - 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', {}) - response = Response(request_id, url, status_code, headers, body, self) - callback(response) - 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] - else: - 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) - del self.callbacks[response_id] - if len(self.subscriptions['response_started']) == 0: - session_unsubscribe(self.conn, self.EVENTS['response_started']) + + params = {} + params["request"] = request.request_id + params["action"] = "provideCredentials" + params["credentials"] = {"type": "password", "username": username, "password": password} + + self.conn.execute(self.command_builder("network.continueWithAuth", params)) Add error handling for invalid credentials in the authentication handler to prevent silent failures when authentication fails. py/selenium/webdriver/common/bidi/network.py [93-106] def __continue_with_auth(self, request_id, username, password): + if not username or not password: + raise ValueError("Both username and password must be provided") command = {'method': 'network.continueWithAuth', 'params': { 'request': request_id, 'action': 'provideCredentials', 'credentials': { 'type': 'password', 'username': username, 'password': password } } } self.conn.execute(self.command_iterator(command)) [Suggestion has been applied] Suggestion importance[1-10]: 6 Why: Adding input validation for authentication credentials would prevent potential security issues and improve error handling, though the current implementation would still work for valid credentials.	6

✅ Suggestions up to commit 01f6aaf

Category	Suggestion	Score
Best practice	✅ Add input validation to network operation methods to prevent issues caused by invalid parameters Suggestion Impact: The commit implemented input validation in the continue_request method (lines 348-351) by checking if request_id exists before proceeding. This is a partial implementation of the suggested input validation, though not as comprehensive as the original suggestion which included type checking and value validation for multiple parameters. code diff: + if not self.request_id: + raise ValueError("Request not found.") + + params = {"request": self.request_id} Consider adding input validation for the parameters in methods like continue_response, continue_request, and add_intercept. This can help prevent issues caused by invalid input and improve the overall robustness of the code. py/selenium/webdriver/common/bidi/network.py [39-48] def continue_response(self, request_id, status_code, headers=None, body=None): + if not isinstance(request_id, str) or not request_id: + raise ValueError("request_id must be a non-empty string") + if not isinstance(status_code, int) or status_code < 100 or status_code >= 600: + raise ValueError("status_code must be a valid HTTP status code") params = { 'requestId': request_id, 'status': status_code } if headers is not None: + if not isinstance(headers, dict): + raise ValueError("headers must be a dictionary") params['headers'] = headers if body is not None: + if not isinstance(body, (str, bytes)): + raise ValueError("body must be a string or bytes") params['body'] = body self.conn.execute('network.continueResponse', params) Suggestion importance[1-10]: 8 Why: Input validation is crucial for preventing errors and ensuring that methods receive valid data. This suggestion enhances the robustness and reliability of the code by checking parameter types and values before proceeding with network operations.	8
Enhancement	✅ Implement robust error handling for network operations to improve reliability and debugging Suggestion Impact: The commit implemented error handling in the remove_intercept method (lines 177-181) which catches exceptions from conn.execute() calls and provides informative error messages, similar to what was suggested. code diff: + try: + self.conn.execute(self.command_builder("network.removeIntercept", {"intercept": intercept})) + self.intercepts.remove(intercept) + except Exception as e: + raise Exception(f"Exception: {e}") Consider using a more robust error handling mechanism for the network operations. Wrap the self.conn.execute() calls in try-except blocks to catch and handle potential exceptions, providing more informative error messages or logging. py/selenium/webdriver/common/bidi/network.py [48] -self.conn.execute('network.continueResponse', params) +try: + self.conn.execute('network.continueResponse', params) +except Exception as e: + logging.error(f"Failed to continue response: {str(e)}") + raise Suggestion importance[1-10]: 7 Why: Adding error handling around network operations can improve the robustness of the code by catching and logging exceptions, which aids in debugging and ensures that the application can handle unexpected issues gracefully.	7
	✅ Add a method to remove event listeners, improving flexibility in event management Suggestion Impact: The commit introduced methods to remove event handlers, such as `remove_authentication_handler`, `remove_request_handler`, and `remove_response_handler`, which align with the suggestion to provide more flexibility in managing event subscriptions. code diff: + def remove_authentication_handler(self, username,): + """Removes an authentication handler.""" + 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=''): + """ + Adds a request handler that executes a callback function when a request matches the given URL pattern. + + Parameters: + callback (function): A function to be executed when url is matched by a URL pattern + The callback function receives a `Response` object as its argument. + url_pattern (str, optional): A substring to match against the response URL. + Default is an empty string, which matches all URLs. + + Returns: + str: The request ID of the intercepted response. + """ + self.__add_intercept(phases=[self.PHASES['before_request']]) + def callback_on_url_match(data): + if url_pattern in data['request']['url']: + # create request object to pass to callback + request_id = data['request'].get('requestId') + url = data['request'].get('url') + method = data['request'].get('method') + headers = data['request'].get('headers', {}) + body = data['request'].get('postData', None) + request = Request(request_id, url, method, headers, body, self) + callback(request) + 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] + else: + 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) + del self.callbacks[request_id] + if len(self.subscriptions['before_request']) == 0: + session_unsubscribe(self.conn, self.EVENTS['before_request']) + + 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. + + Parameters: + callback (function): A function to be executed when url is matched by a url_pattern + The callback function receives a `Response` object as its argument. + url_pattern (str, optional): A substring to match against the response URL. + Default is an empty string, which matches all URLs. + + Returns: + str: The request ID of the intercepted response. + """ + 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') + 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) + 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] + else: + 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) + del self.callbacks[response_id] + if len(self.subscriptions['response_started']) == 0: + session_unsubscribe(self.conn, self.EVENTS['response_started']) Consider implementing a method to remove all event listeners for a specific event or all events. This would provide more flexibility in managing event subscriptions and prevent potential memory leaks. py/selenium/webdriver/common/bidi/network.py [91-94] def on(self, event, callback): event = self.EVENTS.get(event, event) self.callbacks[event] = callback session_subscribe(self.conn, event, self.handle_event) +def off(self, event=None): + if event: + event = self.EVENTS.get(event, event) + if event in self.callbacks: + del self.callbacks[event] + session_unsubscribe(self.conn, event, self.handle_event) + else: + for event in list(self.callbacks.keys()): + session_unsubscribe(self.conn, event, self.handle_event) + self.callbacks.clear() + Suggestion importance[1-10]: 7 Why: Implementing a method to remove event listeners provides better control over event management and can prevent potential memory leaks. This enhancement increases the flexibility and maintainability of the code.	7
	✅ Expand test coverage to include various scenarios and edge cases for network operations Suggestion Impact: The commit implements expanded test coverage by adding multiple new test cases for network operations, including request handlers, response handlers, and authentication scenarios. While not exactly matching the suggested parametrized test, it achieves the same goal of comprehensive testing. code diff: +@pytest.mark.xfail_safari +def test_network_initialized(driver): + assert driver.network is not None -def test_add_intercept(driver, network): - network.add_intercept(phases=[Network.PHASES['before_request']]) +@pytest.mark.xfail_safari +def test_add_response_handler(driver, pages): + passed = [False] -def test_remove_intercept(driver, network): - intercept = network.add_intercept(phases=[Network.PHASES['before_request']]) - network.remove_intercept(intercept) + def callback(response): + passed[0] = True + response.continue_response() -def test_continue_response(driver, network): - network.add_intercept(phases=[Network.PHASES['before_request']]) - network.on('before_request', lambda event: network.continue_response(event['requestId'], 200)) - - driver.get(url_for("basicAuth")) - assert driver.find_element_by_tag_name('h1').text == 'authorized' + driver.network.add_response_handler(callback) + pages.load("basicAuth") + assert passed[0], "Callback was NOT successful" -def test_continue_request(driver, network): - network.add_intercept(phases=[Network.PHASES['before_request']]) - network.on('before_request', lambda event: network.continue_request(event['requestId'], url=url_for("basicAuth"))) - - driver.get(url_for("basicAuth")) - assert driver.find_element_by_tag_name('h1').text == 'authorized' +@pytest.mark.xfail_safari +def test_remove_response_handler(driver, pages): + passed = [False] -def test_continue_with_auth(driver, network): - username = 'your_username' - password = 'your_password' - - network.add_intercept(phases=[Network.PHASES['auth_required']]) - network.on('auth_required', lambda event: network.continue_with_auth(event['requestId'], username, password)) - - driver.get(url_for("basicAuth")) - assert driver.find_element_by_tag_name('h1').text == 'authorized' + def callback(response): + passed[0] = True + response.continue_response() + + test_response_id = driver.network.add_response_handler(callback) + driver.network.remove_response_handler(response_id=test_response_id) + 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] + + def callback(request): + passed[0] = True + request.continue_request() + + driver.network.add_request_handler(callback) + pages.load("basicAuth") + assert passed[0], "Callback was NOT successful" + +@pytest.mark.xfail_safari +def test_remove_request_handler(driver, pages): + passed = [False] + + def callback(request): + passed[0] = True + request.continue_request() + + test_request_id = driver.network.add_request_handler(callback) + driver.network.remove_request_handler(request_id=test_request_id) + 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') + pages.load("basicAuth") + 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.remove_authentication_handler() + pages.load("basicAuth") + assert driver.find_element(By.TAG_NAME, 'h1').text != 'authorized', "Authentication was successful" Enhance the test coverage by adding more comprehensive test cases for different scenarios, including edge cases and error conditions. This will help ensure the robustness of the network functionality. py/test/selenium/webdriver/common/bidi_network_tests.py [27-32] -def test_continue_response(driver, network): +@pytest.mark.parametrize("status_code,expected_text", [ + (200, "authorized"), + (401, "unauthorized"), + (500, "server error") +]) +def test_continue_response(driver, network, status_code, expected_text): network.add_intercept(phases=[Network.PHASES['before_request']]) - network.on('before_request', lambda event: network.continue_response(event['requestId'], 200)) + network.on('before_request', lambda event: network.continue_response(event['requestId'], status_code)) driver.get(url_for("basicAuth")) - assert driver.find_element_by_tag_name('h1').text == 'authorized' + assert expected_text in driver.page_source.lower() Suggestion importance[1-10]: 6 Why: Enhancing test coverage by including different scenarios and edge cases can significantly improve the reliability of the network functionality. This suggestion is valuable for ensuring that the code behaves correctly under various conditions.	6

✅ Suggestions up to commit 14bf971

Category	Suggestion	Score
Possible issue	Validate input parameters to prevent potential issues with invalid data being passed to the BiDi protocol Consider implementing input validation for the add_intercept method to ensure that the provided phases are valid according to the PHASES dictionary. This can prevent potential issues with invalid phase names being passed to the BiDi protocol. py/selenium/webdriver/common/bidi/network.py [68-76] def add_intercept(self, phases=None, contexts=None, url_patterns=None): if phases is None: phases = [] + else: + valid_phases = set(self.PHASES.values()) + if not all(phase in valid_phases for phase in phases): + raise ValueError(f"Invalid phase(s) provided. Valid phases are: {', '.join(valid_phases)}") params = { 'phases': phases, 'contexts': contexts, 'urlPatterns': url_patterns } self.conn.execute('network.addIntercept', params) Suggestion importance[1-10]: 8 Why: Implementing input validation for the add_intercept method is a valuable enhancement that prevents invalid phase names from causing issues. This suggestion addresses a potential bug and improves the method's robustness.	8
Enhancement	✅ Implement error handling for network operations to improve robustness and prevent unexpected crashes Suggestion Impact: The commit implemented error handling in the remove_intercept method (lines 191-195) which catches exceptions when executing network operations, similar to what was suggested. While not implementing the exact try-except pattern in continue_response, it added error handling to network operations. code diff: + try: + self.conn.execute(self.command_builder("network.removeIntercept", {"intercept": intercept})) + self.intercepts.remove(intercept) + except Exception as e: + raise Exception(f"Exception: {e}") Consider using a more robust error handling mechanism for the network operations. Wrap the self.conn.execute() calls in try-except blocks to catch and handle potential exceptions, ensuring graceful error handling and preventing crashes. py/selenium/webdriver/common/bidi/network.py [43-52] def continue_response(self, request_id, status_code, headers=None, body=None): params = { 'requestId': request_id, 'status': status_code } if headers is not None: params['headers'] = headers if body is not None: params['body'] = body - self.conn.execute('network.continueResponse', params) + try: + self.conn.execute('network.continueResponse', params) + except Exception as e: + logging.error(f"Error in continue_response: {str(e)}") + raise Suggestion importance[1-10]: 7 Why: Adding error handling to network operations can improve the robustness of the code by preventing crashes due to unhandled exceptions. This suggestion is relevant and enhances the reliability of the continue_response method.	7
	✅ Enhance test assertions to provide more comprehensive verification of network interception behavior Suggestion Impact: The commit added more comprehensive assertions to verify network interception behavior, such as checking the status of responses and the method of requests, which aligns with the suggestion to enhance test assertions. code diff: +def test_add_response_handler(response): + passed = [False] -def test_continue_request(driver, network): - network.add_intercept(phases=[Network.PHASES['before_request']]) - network.on('before_request', lambda event: network.continue_request(event['requestId'], url=url_for("basicAuth"))) - - driver.get(url_for("basicAuth")) - assert driver.find_element_by_tag_name('h1').text == 'authorized' + def callback(event): + if event['response']['status'] == 200: + passed[0] = True -def test_continue_with_auth(driver, network): - username = 'your_username' - password = 'your_password' - - network.add_intercept(phases=[Network.PHASES['auth_required']]) - network.on('auth_required', lambda event: network.continue_with_auth(event['requestId'], username, password)) - - driver.get(url_for("basicAuth")) - assert driver.find_element_by_tag_name('h1').text == 'authorized'+ network_response.add_response_handler(callback) + pages.load("basicAuth") + assert passed[0] == True, "Callback was NOT successful" + +def test_remove_response_handler(response): + passed = [False] + + def callback(event): + if event['response']['status'] == 200: + passed[0] = True + + network_response.add_response_handler(callback) + network_response.remove_response_handler(callback) + pages.load("basicAuth") + assert passed[0] == False, "Callback was successful" + +def test_add_request_handler(request): + passed = [False] + + def callback(event): + if event['request']['method'] == 'GET': + passed[0] = True + + network_request.add_request_handler(callback) + pages.load("basicAuth") + assert passed[0] == True, "Callback was NOT successful" + +def test_remove_request_handler(request): + passed = [False] + + def callback(event): + if event['request']['method'] == 'GET': + passed[0] = True + + network_request.add_request_handler(callback) + network_request.remove_request_handler(callback) + pages.load("basicAuth") + assert passed[0] == False, "Callback was successful" + +def test_add_authentication_handler(network): + network.add_authentication_handler('test','test') + pages.load("basicAuth") + assert driver.find_element_by_tag_name('h1').text == 'authorized', "Authentication was NOT successful" + +def test_remove_authentication_handler(network): + network.add_authentication_handler('test', 'test') + network.remove_authentication_handler() + pages.load("basicAuth") Consider adding more comprehensive assertions to verify the behavior of network interceptions. This could include checking the content of intercepted requests or responses, verifying headers, or ensuring that the correct number of interceptions occur. py/test/selenium/webdriver/common/bidi_network_tests.py [29-34] def test_continue_response(driver, network): + intercepted_requests = [] network.add_intercept(phases=[Network.PHASES['before_request']]) + network.on('before_request', lambda event: intercepted_requests.append(event)) network.on('before_request', lambda event: network.continue_response(event['requestId'], 200)) driver.get(url_for("basicAuth")) assert driver.find_element_by_tag_name('h1').text == 'authorized' + assert len(intercepted_requests) > 0, "No requests were intercepted" + assert intercepted_requests[0]['request']['url'] == url_for("basicAuth"), "Unexpected URL intercepted" Suggestion importance[1-10]: 7 Why: Adding more comprehensive assertions in tests improves the verification of network interception behavior, ensuring the tests are more thorough and reliable. This suggestion enhances the quality of the test suite.	7
Best practice	Use a context manager for WebDriver instances to ensure proper resource cleanup Consider using a context manager for the WebDriver instance to ensure proper cleanup, even if an exception occurs during the test. This can help prevent resource leaks and improve test reliability. py/test/selenium/webdriver/common/bidi_network_tests.py [11-16] @pytest.fixture def driver(): options = webdriver.ChromeOptions() options.add_argument('--remote-debugging-port=9222') - driver = webdriver.Chrome(options=options) - yield driver - driver.quit() + with webdriver.Chrome(options=options) as driver: + yield driver Suggestion importance[1-10]: 6 Why: Using a context manager for the WebDriver instance ensures proper cleanup and resource management, even if exceptions occur. This suggestion follows best practices for resource handling in tests.	6

[shbenzer]

was adding this to BUILD.bazel for easy testing:

py_test(
    name = "bidi_network_tests",
    srcs = ["test/selenium/webdriver/common/bidi_network_tests.py"],
    deps = [
        ":init-tree",
        ":selenium",
        ":webserver",
    ] + TEST_DEPS,
)

so i could run bazel test //py:bidi_network_tests

[shbenzer]

Think it's ready now... @AutomatedTester could you give it a look?

[qodo-merge-pro]

Persistent review updated to latest commit 01f6aaf

[shbenzer]

Any thoughts @p0deje @AutomatedTester ?

[navin772]

@shbenzer this seems like a known issue with Chromium BiDi - #15553. We had a discussion on slack TLC channel regarding this (pinged you there).

[shbenzer]

//py:common-firefox-bidi-test/selenium/webdriver/common/driver_element_finding_tests.py TIMEOUT in 2 out of 2 in 600.2s

Unrelated failure due to flaky test

[AutomatedTester]

LGTM! Thanks!

[cgoldberg]

congrats on landing this!

[shbenzer]

congrats on landing this!

Thank you!

[gabriel-swapcard]

How do I use the url_patterns for add_request_handler? I'm trying different formats, but I always get an invalid argument exception

[shbenzer]

How do I use the url_patterns for add_request_handler? I'm trying different formats, but I always get an invalid argument exception

It should be a list of these objects:

network.UrlPatternString = {
    type: "string",
    pattern: text,
}

See: https://w3c.github.io/webdriver-bidi/#type-network-UrlPattern and https://w3c.github.io/webdriver-bidi/#command-network-addIntercept

If you find a problem (which is certainly possible given the newness of our implementations, as well as the browsers') please create an issue with a reproducible example for us to diagnose.