[py] fix type errors for log.py, chromium/options.py, websocket_connection.py and chrome/options.py (#14392)

* fix type errors for `log.py`

* fix type errors for `chromium/options.py`

Signed-off-by: Navin Chandra <[email protected]>

* ignore mypy stub type error for `websocket_connection.py`

Signed-off-by: Navin Chandra <[email protected]>

* fix type errors for `chrome/options.py`

Signed-off-by: Navin Chandra <[email protected]>

* fix failing RBE tests due to `log.py`

Signed-off-by: Navin Chandra <[email protected]>

---------

Signed-off-by: Navin Chandra <[email protected]>
Co-authored-by: Sri Harsha <[email protected]>

Author: navin772

py/selenium/webdriver/chrome/options.py

@@ -28,7 +28,7 @@ def default_capabilities(self) -> dict:
 
     def enable_mobile(
         self,
-        android_package: str = "com.android.chrome",
+        android_package: Optional[str] = "com.android.chrome",
         android_activity: Optional[str] = None,
         device_serial: Optional[str] = None,
     ) -> None:

py/selenium/webdriver/chromium/options.py

@@ -18,7 +18,9 @@
 import base64
 import os
 from typing import BinaryIO
+from typing import Dict
 from typing import List
+from typing import Optional
 from typing import Union
 
 from selenium.webdriver.common.desired_capabilities import DesiredCapabilities
@@ -30,11 +32,11 @@ class ChromiumOptions(ArgOptions):
 
     def __init__(self) -> None:
         super().__init__()
-        self._binary_location = ""
-        self._extension_files = []
-        self._extensions = []
-        self._experimental_options = {}
-        self._debugger_address = None
+        self._binary_location: str = ""
+        self._extension_files: List[str] = []
+        self._extensions: List[str] = []
+        self._experimental_options: Dict[str, Union[str, int, dict, List[str]]] = {}
+        self._debugger_address: Optional[str] = None
 
     @property
     def binary_location(self) -> str:
@@ -53,7 +55,7 @@ def binary_location(self, value: str) -> None:
         self._binary_location = value
 
     @property
-    def debugger_address(self) -> str:
+    def debugger_address(self) -> Optional[str]:
         """:Returns: The address of the remote devtools instance."""
         return self._debugger_address
 

py/selenium/webdriver/common/log.py

@@ -19,6 +19,10 @@
 import pkgutil
 from contextlib import asynccontextmanager
 from importlib import import_module
+from typing import Any
+from typing import AsyncGenerator
+from typing import Dict
+from typing import Optional
 
 from selenium.webdriver.common.by import By
 
@@ -45,10 +49,14 @@ def __init__(self, driver, bidi_session) -> None:
         self.cdp = bidi_session.cdp
         self.devtools = bidi_session.devtools
         _pkg = ".".join(__name__.split(".")[:-1])
-        self._mutation_listener_js = pkgutil.get_data(_pkg, "mutation-listener.js").decode("utf8").strip()
+        # Ensure _mutation_listener_js is not None before decoding
+        _mutation_listener_js_bytes: Optional[bytes] = pkgutil.get_data(_pkg, "mutation-listener.js")
+        if _mutation_listener_js_bytes is None:
+            raise ValueError("Failed to load mutation-listener.js")
+        self._mutation_listener_js = _mutation_listener_js_bytes.decode("utf8").strip()
 
     @asynccontextmanager
-    async def mutation_events(self) -> dict:
+    async def mutation_events(self) -> AsyncGenerator[Dict[str, Any], None]:
         """Listen for mutation events and emit them as they are found.
 
         :Usage:
@@ -76,12 +84,13 @@ async def mutation_events(self) -> dict:
         )
         self.driver.pin_script(self._mutation_listener_js, script_key)
         self.driver.execute_script(f"return {self._mutation_listener_js}")
-        event = {}
+
+        event: Dict[str, Any] = {}
         async with runtime.wait_for(self.devtools.runtime.BindingCalled) as evnt:
             yield event
 
         payload = json.loads(evnt.value.payload)
-        elements: list = self.driver.find_elements(By.CSS_SELECTOR, f"*[data-__webdriver_id={payload['target']}")
+        elements: list = self.driver.find_elements(By.CSS_SELECTOR, f"*[data-__webdriver_id={payload['target']}]")
         if not elements:
             elements.append(None)
         event["element"] = elements[0]
@@ -90,7 +99,7 @@ async def mutation_events(self) -> dict:
         event["old_value"] = payload["oldValue"]
 
     @asynccontextmanager
-    async def add_js_error_listener(self):
+    async def add_js_error_listener(self) -> AsyncGenerator[Dict[str, Any], None]:
         """Listen for JS errors and when the contextmanager exits check if
         there were JS Errors.
 
@@ -114,7 +123,7 @@ async def add_js_error_listener(self):
         js_exception.exception_details = exception.value.exception_details
 
     @asynccontextmanager
-    async def add_listener(self, event_type) -> dict:
+    async def add_listener(self, event_type) -> AsyncGenerator[Dict[str, Any], None]:
         """Listen for certain events that are passed in.
 
         :Args:
@@ -134,7 +143,7 @@ async def add_listener(self, event_type) -> dict:
         await session.execute(self.devtools.page.enable())
         session = self.cdp.get_session_context("runtime.enable")
         await session.execute(self.devtools.runtime.enable())
-        console = {"message": None, "level": None}
+        console: Dict[str, Any] = {"message": None, "level": None}
         async with session.wait_for(self.devtools.runtime.ConsoleAPICalled) as messages:
             yield console
 

py/selenium/webdriver/remote/websocket_connection.py

@@ -20,7 +20,7 @@
 from threading import Thread
 from time import sleep
 
-from websocket import WebSocketApp
+from websocket import WebSocketApp  # type: ignore
 
 logger = logging.getLogger(__name__)