Skip to content

Commit

Permalink
fix type errors for service.py, cdp.py, webelement.py and `remo…
Browse files Browse the repository at this point in the history
…te_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 <[email protected]>
  • Loading branch information
navin772 and harsha509 authored Sep 2, 2024
1 parent 8fc4299 commit be40b5c
Show file tree
Hide file tree
Showing 5 changed files with 23 additions and 16 deletions.
2 changes: 2 additions & 0 deletions py/selenium/webdriver/common/bidi/cdp.py
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand Down
23 changes: 14 additions & 9 deletions py/selenium/webdriver/common/service.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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

Expand All @@ -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:
Expand All @@ -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
Expand Down Expand Up @@ -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,
Expand All @@ -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
Expand Down
6 changes: 3 additions & 3 deletions py/selenium/webdriver/remote/remote_connection.py
Original file line number Diff line number Diff line change
Expand Up @@ -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()

Expand Down
2 changes: 1 addition & 1 deletion py/selenium/webdriver/remote/webelement.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)}
Expand Down
6 changes: 3 additions & 3 deletions py/selenium/webdriver/webkitgtk/service.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@

from selenium.webdriver.common import service

DEFAULT_EXECUTABLE_PATH = "WebKitWebDriver"
DEFAULT_EXECUTABLE_PATH: str = "WebKitWebDriver"


class Service(service.Service):
Expand All @@ -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__(
Expand All @@ -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

0 comments on commit be40b5c

Please sign in to comment.