Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

fix type errors for service.py, cdp.py, webelement.py and remote_connection.py #14448

Merged
merged 3 commits into from
Sep 2, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
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
Loading