Add non-blocking navigation

Author: jman-sketch

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

@@ -0,0 +1,55 @@
+import re
+import typing
+from dataclasses import dataclass, fields, is_dataclass
+
+from selenium.webdriver.common.bidi.cdp import import_devtools
+
+devtools = import_devtools("")
+event_class = devtools.util.event_class
+
+@dataclass
+class NavigateParameters:
+    context: str
+    url: str
+    wait: str = "complete"
+
+    def to_json(self):
+        json = {}
+        for field in fields(self):
+            key = field.name
+            value = getattr(self, key)
+            if not value:
+                continue
+            if is_dataclass(value):
+                value = value.to_json()
+            json = json | {re.sub(r"^_", "", key): value}
+        return json
+
+    @classmethod
+    def from_json(cls, json):
+        return cls(**json)
+
+@dataclass
+class Navigate:
+    params: NavigateParameters
+    method: typing.Literal["browsingContext.navigate"] = "browsingContext.navigate"
+
+    def to_json(self):
+        json = {}
+        for field in fields(self):
+            key = field.name
+            value = getattr(self, key)
+            if not value:
+                continue
+            if is_dataclass(value):
+                value = value.to_json()
+            json = json | {re.sub(r"^_", "", key): value}
+        return json
+
+    @classmethod
+    def from_json(cls, json):
+        return cls(**json)
+
+    def cmd(self):
+        result = yield self.to_json()
+        return result

py/selenium/webdriver/remote/network.py

@@ -1,6 +1,7 @@
 from dataclasses import fields
 
 from selenium.webdriver.common.bidi import network
+from selenium.webdriver.common.bidi.browsing_context import Navigate, NavigateParameters
 from selenium.webdriver.common.bidi.network import (
     AddInterceptParameters,
     BeforeRequestSent,
@@ -29,6 +30,10 @@ async def add_request_handler(self, request_filter=lambda _: True, handler=defau
         await self.network.add_listener(event=BeforeRequestSent, callback=callback)
         return intercept
 
+    async def get(self, url, conn):
+        params = NavigateParameters(context=self.driver.current_window_handle, url=url)
+        await conn.execute(Navigate(params).cmd())
+
     def _callback(self, request_filter, handler):
         async def callback(request):
             if request_filter(request):