marked add/remove_intercept() private, added clear_request_handlers()

Author: shbenzer

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

@@ -63,7 +63,7 @@ def command_builder(self, method, params):
         cmd = yield command
         return cmd
 
-    def add_intercept(self, phases=[], contexts=None, url_patterns=None):
+    def _add_intercept(self, phases=[], contexts=None, url_patterns=None):
         """Add an intercept to the network.
 
         Parameters:
@@ -94,7 +94,7 @@ def add_intercept(self, phases=[], contexts=None, url_patterns=None):
         self.intercepts.append(result["intercept"])
         return result
 
-    def remove_intercept(self, intercept=None):
+    def _remove_intercept(self, intercept=None):
         """Remove a specific intercept, or all intercepts.
 
         Parameters:
@@ -183,7 +183,7 @@ def add_request_handler(self, event, callback, url_patterns=None, contexts=None)
         except KeyError:
             raise Exception(f"Event {event} not found")
 
-        result = self.add_intercept(phases=[phase_name], url_patterns=url_patterns, contexts=contexts)
+        result = self._add_intercept(phases=[phase_name], url_patterns=url_patterns, contexts=contexts)
         callback_id = self._on_request(event_name, callback)
 
         if event_name in self.subscriptions:
@@ -213,7 +213,7 @@ def remove_request_handler(self, event, callback_id):
         net_event = NetworkEvent(event_name)
 
         self.conn.remove_callback(net_event, callback_id)
-        self.remove_intercept(self.callbacks[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:
@@ -222,6 +222,20 @@ def remove_request_handler(self, event, callback_id):
             self.conn.execute(self.command_builder("session.unsubscribe", params))
             del self.subscriptions[event_name]
 
+    def clear_request_handlers(self):
+        """Clear all request handlers from the network."""
+
+        for event_name in self.subscriptions:
+            net_event = NetworkEvent(event_name)
+            for callback_id in self.subscriptions[event_name]:
+                self.conn.remove_callback(net_event, callback_id)
+                self._remove_intercept(self.callbacks[callback_id])
+                del self.callbacks[callback_id]
+            params = {}
+            params["events"] = [event_name]
+            self.conn.execute(self.command_builder("session.unsubscribe", params))
+        self.subscriptions = {}
+
     def add_auth_handler(self, username, password):
         """Add an authentication handler to the network.
 

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

@@ -28,14 +28,14 @@ def test_network_initialized(driver):
 
 @pytest.mark.xfail_safari
 def test_add_intercept(driver, pages):
-    result = driver.network.add_intercept()
+    result = driver.network._add_intercept()
     assert result is not None, "Intercept not added"
 
 
 @pytest.mark.xfail_safari
 def test_remove_intercept(driver):
-    result = driver.network.add_intercept()
-    driver.network.remove_intercept(result["intercept"])
+    result = driver.network._add_intercept()
+    driver.network._remove_intercept(result["intercept"])
     assert driver.network.intercepts == [], "Intercept not removed"
 
 
@@ -55,6 +55,25 @@ def callback(request: Request):
     assert driver.find_element(By.NAME, "login").is_displayed(), "Request not continued"
 
 
+@pytest.mark.xfail_safari
+def test_clear_request_handlers(driver, pages):
+    requests = []
+
+    def callback(request: Request):
+        requests.append(request)
+
+    callback_id_1 = driver.network.add_request_handler("before_request", callback)
+    assert callback_id_1 is not None, "Request handler not added"
+    callback_id_2 = driver.network.add_request_handler("before_request", callback)
+    assert callback_id_2 is not None, "Request handler not added"
+
+    driver.network.clear_request_handlers()
+
+    pages.load("formPage.html")
+    assert not requests, "Requests intercepted"
+    assert driver.find_element(By.NAME, "login").is_displayed(), "Request not continued"
+
+
 @pytest.mark.xfail_safari
 def test_continue_request(driver, pages):