From ba04acdf9ea3e53c483cc38a1ae796496e5da9c1 Mon Sep 17 00:00:00 2001 From: symonk Date: Sun, 2 Oct 2022 17:10:13 +0100 Subject: [PATCH] [py]: Base `Service` tidy up --- py/selenium/webdriver/common/service.py | 77 +++++++++++++------------ py/tox.ini | 2 +- 2 files changed, 40 insertions(+), 39 deletions(-) diff --git a/py/selenium/webdriver/common/service.py b/py/selenium/webdriver/common/service.py index 87ac85d5a59f9..fbd65b998f273 100644 --- a/py/selenium/webdriver/common/service.py +++ b/py/selenium/webdriver/common/service.py @@ -14,9 +14,11 @@ # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. +import contextlib import errno import os import subprocess +import typing from platform import system from subprocess import DEVNULL from subprocess import PIPE @@ -29,22 +31,22 @@ class Service: - - def __init__(self, executable, port=0, log_file=DEVNULL, env=None, start_error_message=""): + def __init__( + self, + executable: str, + port: int = 0, + log_file=DEVNULL, + env: typing.Optional[typing.Dict[typing.Any, typing.Any]] = None, + start_error_message: str = "", + ) -> None: self.path = executable - - self.port = port - if self.port == 0: - self.port = utils.free_port() - - if not _HAS_NATIVE_DEVNULL and log_file == DEVNULL: - log_file = open(os.devnull, 'wb') - + self.port = port or utils.free_port() + self.log_file = open(os.devnull, "wb") if not _HAS_NATIVE_DEVNULL and log_file == DEVNULL else log_file self.start_error_message = start_error_message - self.log_file = log_file # Default value for every python subprocess: subprocess.Popen(..., creationflags=0) - self.creationflags = 0 + self.creation_flags = 0 self.env = env or os.environ + self.process = None @property def service_url(self): @@ -67,31 +69,37 @@ def start(self): try: cmd = [self.path] cmd.extend(self.command_line_args()) - self.process = subprocess.Popen(cmd, env=self.env, - close_fds=system() != 'Windows', - stdout=self.log_file, - stderr=self.log_file, - stdin=PIPE, - creationflags=self.creationflags) + self.process = subprocess.Popen( + cmd, + env=self.env, + close_fds=system() != "Windows", + stdout=self.log_file, + stderr=self.log_file, + stdin=PIPE, + creationflags=self.creation_flags, + ) except TypeError: raise except OSError as err: if err.errno == errno.ENOENT: raise WebDriverException( "'{}' executable needs to be in PATH. {}".format( - os.path.basename(self.path), self.start_error_message) + os.path.basename(self.path), self.start_error_message + ) ) elif err.errno == errno.EACCES: raise WebDriverException( "'{}' executable may have wrong permissions. {}".format( - os.path.basename(self.path), self.start_error_message) + os.path.basename(self.path), self.start_error_message + ) ) else: raise except Exception as e: raise WebDriverException( - "The executable %s needs to be available in the path. %s\n%s" % - (os.path.basename(self.path), self.start_error_message, str(e))) + "The executable %s needs to be available in the path. %s\n%s" + % (os.path.basename(self.path), self.start_error_message, str(e)) + ) count = 0 while True: self.assert_process_still_running() @@ -106,10 +114,7 @@ def start(self): def assert_process_still_running(self): return_code = self.process.poll() if return_code: - raise WebDriverException( - 'Service %s unexpectedly exited. Status code was: %s' - % (self.path, return_code) - ) + raise WebDriverException(f"Service {self.path} unexpectedly exited. Status code was: {return_code}") def is_connectable(self): return utils.is_connectable(self.port) @@ -117,8 +122,9 @@ def is_connectable(self): def send_remote_shutdown_command(self): from urllib import request from urllib.error import URLError + try: - request.urlopen("%s/shutdown" % self.service_url) + request.urlopen(f"{self.service_url}/shutdown") except URLError: return @@ -127,15 +133,14 @@ def send_remote_shutdown_command(self): break sleep(1) - def stop(self): + def stop(self) -> None: """ Stops the service. """ if self.log_file != PIPE and not (self.log_file == DEVNULL and _HAS_NATIVE_DEVNULL): - try: + with contextlib.suppress(Exception): + # Todo: Be explicit in what we are catching here. self.log_file.close() - except Exception: - pass if not self.process: return @@ -147,9 +152,7 @@ def stop(self): try: if self.process: - for stream in [self.process.stdin, - self.process.stdout, - self.process.stderr]: + for stream in [self.process.stdin, self.process.stdout, self.process.stderr]: try: stream.close() except AttributeError: @@ -161,11 +164,9 @@ def stop(self): except OSError: pass - def __del__(self): + def __del__(self) -> None: # `subprocess.Popen` doesn't send signal on `__del__`; # so we attempt to close the launched process when `__del__` # is triggered. - try: + with contextlib.suppress(Exception): self.stop() - except Exception: - pass diff --git a/py/tox.ini b/py/tox.ini index 223c8d6f0de3c..00fbb4b16bcbe 100644 --- a/py/tox.ini +++ b/py/tox.ini @@ -57,5 +57,5 @@ deps = flake8-typing-imports==1.13.0 commands = isort selenium/ test/ - black test/ selenium/common/ selenium/webdriver/safari -l 120 + black test/ selenium/common/ selenium/webdriver/safari selenium/webdriver/common/service.py -l 120 flake8 selenium/ test/ --min-python-version=3.7