From be40b5c85350b2f5cf83194cce4cb1ab23e13172 Mon Sep 17 00:00:00 2001 From: Navin Chandra <98466550+navin772@users.noreply.github.com> Date: Mon, 2 Sep 2024 19:52:29 +0530 Subject: [PATCH] fix type errors for `service.py`, `cdp.py`, `webelement.py` and `remote_connection.py` (#14448) * fix type errors for `service.py`, `cdp.py`, `webelement.py` and `remote_connection.py` * remove raise error --------- Co-authored-by: Sri Harsha <12621691+harsha509@users.noreply.github.com> --- py/selenium/webdriver/common/bidi/cdp.py | 2 ++ py/selenium/webdriver/common/service.py | 23 +++++++++++-------- .../webdriver/remote/remote_connection.py | 6 ++--- py/selenium/webdriver/remote/webelement.py | 2 +- py/selenium/webdriver/webkitgtk/service.py | 6 ++--- 5 files changed, 23 insertions(+), 16 deletions(-) diff --git a/py/selenium/webdriver/common/bidi/cdp.py b/py/selenium/webdriver/common/bidi/cdp.py index b2b3ae5f7368f..c4cb0feeedf40 100644 --- a/py/selenium/webdriver/common/bidi/cdp.py +++ b/py/selenium/webdriver/common/bidi/cdp.py @@ -237,6 +237,8 @@ async def wait_for(self, event_type: typing.Type[T], buffer_size=10) -> typing.A an async with block. The block will not exit until the indicated event is received. """ + sender: trio.MemorySendChannel + receiver: trio.MemoryReceiveChannel sender, receiver = trio.open_memory_channel(buffer_size) self.channels[event_type].add(sender) proxy = CmEventProxy() diff --git a/py/selenium/webdriver/common/service.py b/py/selenium/webdriver/common/service.py index 3287b77a5ac60..829e4f43ad967 100644 --- a/py/selenium/webdriver/common/service.py +++ b/py/selenium/webdriver/common/service.py @@ -25,6 +25,7 @@ from platform import system from subprocess import PIPE from time import sleep +from typing import cast from urllib import request from urllib.error import URLError @@ -55,11 +56,11 @@ def __init__( **kwargs, ) -> None: if isinstance(log_output, str): - self.log_output = open(log_output, "a+", encoding="utf-8") + self.log_output = cast(IOBase, open(log_output, "a+", encoding="utf-8")) elif log_output == subprocess.STDOUT: - self.log_output = None + self.log_output = cast(typing.Optional[typing.Union[int, IOBase]], None) elif log_output is None or log_output == subprocess.DEVNULL: - self.log_output = subprocess.DEVNULL + self.log_output = cast(typing.Optional[typing.Union[int, IOBase]], subprocess.DEVNULL) else: self.log_output = log_output @@ -82,7 +83,7 @@ def command_line_args(self) -> typing.List[str]: @property def path(self) -> str: - return self._path + return self._path or "" @path.setter def path(self, value: str) -> None: @@ -95,6 +96,8 @@ def start(self) -> None: - WebDriverException : Raised either when it can't start the service or when it can't connect to the service """ + if self._path is None: + raise WebDriverException("Service path cannot be None.") self._start_process(self._path) count = 0 @@ -201,16 +204,16 @@ def _start_process(self, path: str) -> None: try: start_info = None if system() == "Windows": - start_info = subprocess.STARTUPINFO() - start_info.dwFlags = subprocess.CREATE_NEW_CONSOLE | subprocess.STARTF_USESHOWWINDOW - start_info.wShowWindow = subprocess.SW_HIDE + start_info = subprocess.STARTUPINFO() # type: ignore[attr-defined] + start_info.dwFlags = subprocess.CREATE_NEW_CONSOLE | subprocess.STARTF_USESHOWWINDOW # type: ignore[attr-defined] + start_info.wShowWindow = subprocess.SW_HIDE # type: ignore[attr-defined] self.process = subprocess.Popen( cmd, env=self.env, close_fds=close_file_descriptors, - stdout=self.log_output, - stderr=self.log_output, + stdout=cast(typing.Optional[typing.Union[int, typing.IO[typing.Any]]], self.log_output), + stderr=cast(typing.Optional[typing.Union[int, typing.IO[typing.Any]]], self.log_output), stdin=PIPE, creationflags=self.creation_flags, startupinfo=start_info, @@ -227,6 +230,8 @@ def _start_process(self, path: str) -> None: raise except OSError as err: if err.errno == errno.EACCES: + if self._path is None: + raise WebDriverException("Service path cannot be None.") raise WebDriverException( f"'{os.path.basename(self._path)}' executable may have wrong permissions." ) from err diff --git a/py/selenium/webdriver/remote/remote_connection.py b/py/selenium/webdriver/remote/remote_connection.py index f09937945b90f..c3c28eca0cd59 100644 --- a/py/selenium/webdriver/remote/remote_connection.py +++ b/py/selenium/webdriver/remote/remote_connection.py @@ -137,9 +137,9 @@ class RemoteConnection: browser_name = None _timeout = ( - float(os.getenv("GLOBAL_DEFAULT_TIMEOUT")) - if "GLOBAL_DEFAULT_TIMEOUT" in os.environ - else socket._GLOBAL_DEFAULT_TIMEOUT + float(os.getenv("GLOBAL_DEFAULT_TIMEOUT", str(socket.getdefaulttimeout()))) + if os.getenv("GLOBAL_DEFAULT_TIMEOUT") is not None + else socket.getdefaulttimeout() ) _ca_certs = os.getenv("REQUESTS_CA_BUNDLE") if "REQUESTS_CA_BUNDLE" in os.environ else certifi.where() diff --git a/py/selenium/webdriver/remote/webelement.py b/py/selenium/webdriver/remote/webelement.py index 8f26b26a4b0ee..ef60757294caa 100644 --- a/py/selenium/webdriver/remote/webelement.py +++ b/py/selenium/webdriver/remote/webelement.py @@ -226,7 +226,7 @@ def send_keys(self, *value: str) -> None: remote_files = [] for file in local_files: remote_files.append(self._upload(file)) - value = "\n".join(remote_files) + value = tuple("\n".join(remote_files)) self._execute( Command.SEND_KEYS_TO_ELEMENT, {"text": "".join(keys_to_typing(value)), "value": keys_to_typing(value)} diff --git a/py/selenium/webdriver/webkitgtk/service.py b/py/selenium/webdriver/webkitgtk/service.py index 978aa3cadfe1a..92cea26c535f3 100644 --- a/py/selenium/webdriver/webkitgtk/service.py +++ b/py/selenium/webdriver/webkitgtk/service.py @@ -18,7 +18,7 @@ from selenium.webdriver.common import service -DEFAULT_EXECUTABLE_PATH = "WebKitWebDriver" +DEFAULT_EXECUTABLE_PATH: str = "WebKitWebDriver" class Service(service.Service): @@ -40,7 +40,7 @@ def __init__( service_args: typing.Optional[typing.List[str]] = None, env: typing.Optional[typing.Mapping[str, str]] = None, **kwargs, - ): + ) -> None: self.service_args = service_args or [] log_file = open(log_path, "wb") if log_path else None super().__init__( @@ -49,7 +49,7 @@ def __init__( log_file=log_file, env=env, **kwargs, - ) # type: ignore + ) def command_line_args(self) -> typing.List[str]: return ["-p", f"{self.port}"] + self.service_args