# Licensed to the Software Freedom Conservancy (SFC) under one # or more contributor license agreements. See the NOTICE file # distributed with this work for additional information # regarding copyright ownership. The SFC licenses this file # to you under the Apache License, Version 2.0 (the # "License"); you may not use this file except in compliance # with the License. You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, # software distributed under the License is distributed on an # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. import collections import os import re import shutil import socket import subprocess import time import urllib from selenium.webdriver.common.selenium_manager import SeleniumManager class Server: """Manage a Selenium Grid (Remote) Server in standalone mode. This class contains functionality for downloading the server and starting/stopping it. For more information on Selenium Grid, see: - https://www.selenium.dev/documentation/grid/getting_started/ Parameters: ----------- host : str Hostname or IP address to bind to (determined automatically if not specified) port : int or str Port to listen on (4444 if not specified) path : str Path/filename of existing server .jar file (Selenium Manager is used if not specified) version : str Version of server to download (latest version if not specified) log_level : str Logging level to control logging output ("INFO" if not specified) Available levels: "SEVERE", "WARNING", "INFO", "CONFIG", "FINE", "FINER", "FINEST" env: collections.abc.Mapping Mapping that defines the environment variables for the server process """ def __init__(self, host=None, port=4444, path=None, version=None, log_level="INFO", env=None): if path and version: raise TypeError("Not allowed to specify a version when using an existing server path") self.host = host self.port = self._validate_port(port) self.path = self._validate_path(path) self.version = self._validate_version(version) self.log_level = self._validate_log_level(log_level) self.env = self._validate_env(env) self.process = None self.status_url = self._get_status_url() def _get_status_url(self): host = self.host if self.host is not None else "localhost" return f"http://{host}:{self.port}/status" def _validate_path(self, path): if path and not os.path.exists(path): raise OSError(f"Can't find server .jar located at {path}") return path def _validate_port(self, port): try: port = int(port) except ValueError: raise TypeError(f"{__class__.__name__}.__init__() got an invalid port: '{port}'") if not (0 <= port <= 65535): raise ValueError("port must be 0-65535") return port def _validate_version(self, version): if version: if not re.match(r"^\d+\.\d+\.\d+$", str(version)): raise TypeError(f"{__class__.__name__}.__init__() got an invalid version: '{version}'") return version def _validate_log_level(self, log_level): levels = ("SEVERE", "WARNING", "INFO", "CONFIG", "FINE", "FINER", "FINEST") if log_level not in levels: raise TypeError(f"log_level must be one of: {', '.join(levels)}") return log_level def _validate_env(self, env): if env is not None and not isinstance(env, collections.abc.Mapping): raise TypeError("env must be a mapping of environment variables") return env def _wait_for_server(self, timeout=10): start = time.time() while time.time() - start < timeout: try: urllib.request.urlopen(self.status_url) return True except urllib.error.URLError: time.sleep(0.2) return False def download_if_needed(self, version=None): """Download the server if it doesn't already exist. Latest version is downloaded unless specified. """ args = ["--grid"] if version is not None: args.append(version) return SeleniumManager().binary_paths(args)["driver_path"] def start(self): """Start the server. Selenium Manager will detect the server location and download it if necessary, unless an existing server path was specified. """ path = self.download_if_needed(self.version) if self.path is None else self.path java_path = shutil.which("java") if java_path is None: raise OSError("Can't find java on system PATH. JRE is required to run the Selenium server") command = [ java_path, "-jar", path, "standalone", "--port", str(self.port), "--log-level", self.log_level, "--selenium-manager", "true", "--enable-managed-downloads", "true", ] if self.host is not None: command.extend(["--host", self.host]) host = self.host if self.host is not None else "localhost" try: with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: sock.connect((host, self.port)) raise ConnectionError(f"Selenium server is already running, or something else is using port {self.port}") except ConnectionRefusedError: print("Starting Selenium server...") self.process = subprocess.Popen(command, env=self.env) print(f"Selenium server running as process: {self.process.pid}") if not self._wait_for_server(): raise TimeoutError(f"Timed out waiting for Selenium server at {self.status_url}") print("Selenium server is ready") return self.process def stop(self): """Stop the server.""" if self.process is None: raise RuntimeError("Selenium server isn't running") else: if self.process.poll() is None: self.process.terminate() self.process.wait() self.process = None print("Selenium server has been terminated")