From bfbed918d8e7cad3799a65983f71748de3e5da93 Mon Sep 17 00:00:00 2001 From: Titus Fortner Date: Mon, 15 Apr 2024 04:23:51 -0500 Subject: [PATCH] [py] update SOC for driver finder and selenium manager classes (#13387) * [py] update SOC for driver finder and selenium manager classes * [py] change error and log handling and add tests * [py] adjustments * [py] change names and validations and make tweaks * [py] make the driver finder class non-static * [py] missed renaming a bunch of things * [py] memoize and fix tests * [py] bad logic * [py] tidy things from merges --------- Co-authored-by: Diego Molina --- Rakefile | 17 ++ py/selenium/webdriver/chromium/webdriver.py | 7 +- py/selenium/webdriver/common/driver_finder.py | 69 ++++++- .../webdriver/common/selenium_manager.py | 134 ++++++------- py/selenium/webdriver/firefox/webdriver.py | 7 +- py/selenium/webdriver/ie/webdriver.py | 2 +- py/selenium/webdriver/safari/webdriver.py | 2 +- py/selenium/webdriver/webkitgtk/webdriver.py | 2 +- py/selenium/webdriver/wpewebkit/webdriver.py | 2 +- .../webdriver/common/driver_finder_tests.py | 65 +++++++ .../common/selenium_manager_tests.py | 177 ++++++++++-------- 11 files changed, 312 insertions(+), 172 deletions(-) create mode 100644 py/test/selenium/webdriver/common/driver_finder_tests.py diff --git a/Rakefile b/Rakefile index 66e7bcde966d1..0cb931a841d7a 100644 --- a/Rakefile +++ b/Rakefile @@ -693,6 +693,23 @@ namespace :py do Bazel.execute('test', [],"//py:test-remote") end end + + namespace :test do + desc 'Python unit tests' + task :unit do + Rake::Task['py:clean'].invoke + Bazel.execute('test', ['--test_size_filters=small'], '//py/...') + end + + %i[chrome edge firefox safari].each do |browser| + desc "Python #{browser} tests" + task browser do + Rake::Task['py:clean'].invoke + Bazel.execute('test', %w[--test_output all],"//py:common-#{browser}") + Bazel.execute('test', %w[--test_output all],"//py:test-#{browser}") + end + end + end end def ruby_version diff --git a/py/selenium/webdriver/chromium/webdriver.py b/py/selenium/webdriver/chromium/webdriver.py index 3d9b94a1f0eee..6fc3adeb4086d 100644 --- a/py/selenium/webdriver/chromium/webdriver.py +++ b/py/selenium/webdriver/chromium/webdriver.py @@ -46,7 +46,12 @@ def __init__( """ self.service = service - self.service.path = DriverFinder.get_path(self.service, options) + finder = DriverFinder(self.service, options) + if finder.get_browser_path(): + options.binary_location = finder.get_browser_path() + options.browser_version = None + + self.service.path = finder.get_driver_path() self.service.start() executor = ChromiumRemoteConnection( diff --git a/py/selenium/webdriver/common/driver_finder.py b/py/selenium/webdriver/common/driver_finder.py index 6aa5f96eef01d..5a5cb01c5db7f 100644 --- a/py/selenium/webdriver/common/driver_finder.py +++ b/py/selenium/webdriver/common/driver_finder.py @@ -26,21 +26,74 @@ class DriverFinder: + """A Driver finding class responsible for obtaining the correct driver and + associated browser. + + :param service: instance of the driver service class. + :param options: instance of the browser options class. + """ + + def __init__(self, service: Service, options: BaseOptions) -> None: + self._service = service + self._options = options + self._paths = {"driver_path": "", "browser_path": ""} + """Utility to find if a given file is present and executable. This implementation is still in beta, and may change. """ - @staticmethod - def get_path(service: Service, options: BaseOptions) -> str: - path = service.path + def get_browser_path(self) -> str: + return self._binary_paths()["browser_path"] + + def get_driver_path(self) -> str: + return self._binary_paths()["driver_path"] + + def _binary_paths(self) -> dict: + if self._paths["driver_path"]: + return self._paths + + browser = self._options.capabilities["browserName"] try: - path = SeleniumManager().driver_location(options) if path is None else path + path = self._service.path + if path: + logger.debug( + "Skipping Selenium Manager; path to %s driver specified in Service class: %s", browser, path + ) + if not Path(path).is_file(): + raise ValueError(f"The path is not a valid file: {path}") + self._paths["driver_path"] = path + else: + output = SeleniumManager().binary_paths(self._to_args()) + if Path(output["driver_path"]).is_file(): + self._paths["driver_path"] = output["driver_path"] + else: + raise ValueError(f'The driver path is not a valid file: {output["driver_path"]}') + if Path(output["browser_path"]).is_file(): + self._paths["browser_path"] = output["browser_path"] + else: + raise ValueError(f'The browser path is not a valid file: {output["browser_path"]}') except Exception as err: - msg = f"Unable to obtain driver for {options.capabilities['browserName']} using Selenium Manager." + msg = f"Unable to obtain driver for {browser}" raise NoSuchDriverException(msg) from err + return self._paths + + def _to_args(self) -> list: + args = ["--browser", self._options.capabilities["browserName"]] + + if self._options.browser_version: + args.append("--browser-version") + args.append(str(self._options.browser_version)) + + binary_location = getattr(self._options, "binary_location", None) + if binary_location: + args.append("--browser-path") + args.append(str(binary_location)) - if path is None or not Path(path).is_file(): - raise NoSuchDriverException(f"Unable to locate or obtain driver for {options.capabilities['browserName']}") + proxy = self._options.proxy + if proxy and (proxy.http_proxy or proxy.ssl_proxy): + args.append("--proxy") + value = proxy.ssl_proxy if proxy.ssl_proxy else proxy.http_proxy + args.append(value) - return path + return args diff --git a/py/selenium/webdriver/common/selenium_manager.py b/py/selenium/webdriver/common/selenium_manager.py index 93dd590839f64..46e23f52b606e 100644 --- a/py/selenium/webdriver/common/selenium_manager.py +++ b/py/selenium/webdriver/common/selenium_manager.py @@ -24,7 +24,6 @@ from typing import List from selenium.common import WebDriverException -from selenium.webdriver.common.options import BaseOptions logger = logging.getLogger(__name__) @@ -35,8 +34,26 @@ class SeleniumManager: This implementation is still in beta, and may change. """ + def binary_paths(self, args: List) -> dict: + """Determines the locations of the requested assets. + + :Args: + - args: the commands to send to the selenium manager binary. + :Returns: dictionary of assets and their path + """ + + args = [str(self._get_binary())] + args + if logger.getEffectiveLevel() == logging.DEBUG: + args.append("--debug") + args.append("--language-binding") + args.append("python") + args.append("--output") + args.append("json") + + return self._run(args) + @staticmethod - def get_binary() -> Path: + def _get_binary() -> Path: """Determines the path of the correct Selenium Manager binary. :Returns: The Selenium Manager executable location @@ -45,29 +62,27 @@ def get_binary() -> Path: """ if (path := os.getenv("SE_MANAGER_PATH")) is not None: - return Path(path) - - dirs = { - ("darwin", "any"): "macos", - ("win32", "any"): "windows", - ("cygwin", "any"): "windows", - ("linux", "x86_64"): "linux", - ("freebsd", "x86_64"): "linux", - ("openbsd", "x86_64"): "linux", - } - - arch = platform.machine() if sys.platform in ("linux", "freebsd", "openbsd") else "any" - - directory = dirs.get((sys.platform, arch)) - if directory is None: - raise WebDriverException(f"Unsupported platform/architecture combination: {sys.platform}/{arch}") - - if sys.platform in ["freebsd", "openbsd"]: - logger.warning("Selenium Manager binary may not be compatible with %s; verify settings", sys.platform) - - file = "selenium-manager.exe" if directory == "windows" else "selenium-manager" - - path = Path(__file__).parent.joinpath(directory, file) + logger.debug("Selenium Manager set by env SE_MANAGER_PATH to: %s", path) + path = Path(path) + else: + allowed = { + ("darwin", "any"): "macos/selenium-manager", + ("win32", "any"): "windows/selenium-manager.exe", + ("cygwin", "any"): "windows/selenium-manager.exe", + ("linux", "x86_64"): "linux/selenium-manager", + ("freebsd", "x86_64"): "linux/selenium-manager", + ("openbsd", "x86_64"): "linux/selenium-manager", + } + + arch = platform.machine() if sys.platform in ("linux", "freebsd", "openbsd") else "any" + if sys.platform in ["freebsd", "openbsd"]: + logger.warning("Selenium Manager binary may not be compatible with %s; verify settings", sys.platform) + + location = allowed.get((sys.platform, arch)) + if location is None: + raise WebDriverException(f"Unsupported platform/architecture combination: {sys.platform}/{arch}") + + path = Path(__file__).parent.joinpath(location) if not path.is_file(): raise WebDriverException(f"Unable to obtain working Selenium Manager binary; {path}") @@ -76,60 +91,14 @@ def get_binary() -> Path: return path - def driver_location(self, options: BaseOptions) -> str: - """Determines the path of the correct driver. - - :Args: - - browser: which browser to get the driver path for. - :Returns: The driver path to use - """ - - browser = options.capabilities["browserName"] - - args = [str(self.get_binary()), "--browser", browser] - - if options.browser_version: - args.append("--browser-version") - args.append(str(options.browser_version)) - - binary_location = getattr(options, "binary_location", None) - if binary_location: - args.append("--browser-path") - args.append(str(binary_location)) - - proxy = options.proxy - if proxy and (proxy.http_proxy or proxy.ssl_proxy): - args.append("--proxy") - value = proxy.ssl_proxy if proxy.ssl_proxy else proxy.http_proxy - args.append(value) - - output = self.run(args) - - browser_path = output["browser_path"] - driver_path = output["driver_path"] - logger.debug("Using driver at: %s", driver_path) - - if hasattr(options.__class__, "binary_location") and browser_path: - options.binary_location = browser_path - options.browser_version = None # if we have the binary location we no longer need the version - - return driver_path - @staticmethod - def run(args: List[str]) -> dict: + def _run(args: List[str]) -> dict: """Executes the Selenium Manager Binary. :Args: - args: the components of the command being executed. :Returns: The log string containing the driver location. """ - if logger.getEffectiveLevel() == logging.DEBUG: - args.append("--debug") - args.append("--language-binding") - args.append("python") - args.append("--output") - args.append("json") - command = " ".join(args) logger.debug("Executing process: %s", command) try: @@ -139,17 +108,22 @@ def run(args: List[str]) -> dict: completed_proc = subprocess.run(args, capture_output=True) stdout = completed_proc.stdout.decode("utf-8").rstrip("\n") stderr = completed_proc.stderr.decode("utf-8").rstrip("\n") - output = json.loads(stdout) - result = output["result"] + output = json.loads(stdout) if stdout != "" else {"logs": [], "result": {}} except Exception as err: raise WebDriverException(f"Unsuccessful command executed: {command}") from err - for item in output["logs"]: + SeleniumManager._process_logs(output["logs"]) + result = output["result"] + if completed_proc.returncode: + raise WebDriverException( + f"Unsuccessful command executed: {command}; code: {completed_proc.returncode}\n{result}\n{stderr}" + ) + return result + + @staticmethod + def _process_logs(log_items: List[dict]): + for item in log_items: if item["level"] == "WARN": logger.warning(item["message"]) - if item["level"] == "DEBUG" or item["level"] == "INFO": + elif item["level"] in ["DEBUG", "INFO"]: logger.debug(item["message"]) - - if completed_proc.returncode: - raise WebDriverException(f"Unsuccessful command executed: {command}.\n{result}{stderr}") - return result diff --git a/py/selenium/webdriver/firefox/webdriver.py b/py/selenium/webdriver/firefox/webdriver.py index 75b387c865fc8..d048b01ea5c1d 100644 --- a/py/selenium/webdriver/firefox/webdriver.py +++ b/py/selenium/webdriver/firefox/webdriver.py @@ -56,7 +56,12 @@ def __init__( self.service = service if service else Service() options = options if options else Options() - self.service.path = DriverFinder.get_path(self.service, options) + finder = DriverFinder(self.service, options) + if finder.get_browser_path(): + options.binary_location = finder.get_browser_path() + options.browser_version = None + + self.service.path = finder.get_driver_path() self.service.start() executor = FirefoxRemoteConnection( diff --git a/py/selenium/webdriver/ie/webdriver.py b/py/selenium/webdriver/ie/webdriver.py index 3c9872bc7ab08..64bf79fe250dc 100644 --- a/py/selenium/webdriver/ie/webdriver.py +++ b/py/selenium/webdriver/ie/webdriver.py @@ -46,7 +46,7 @@ def __init__( self.service = service if service else Service() options = options if options else Options() - self.service.path = DriverFinder.get_path(self.service, options) + self.service.path = DriverFinder(self.service, options).get_driver_path() self.service.start() executor = RemoteConnection( diff --git a/py/selenium/webdriver/safari/webdriver.py b/py/selenium/webdriver/safari/webdriver.py index 014b00aab0312..259b2d82047bf 100644 --- a/py/selenium/webdriver/safari/webdriver.py +++ b/py/selenium/webdriver/safari/webdriver.py @@ -45,7 +45,7 @@ def __init__( self.service = service if service else Service() options = options if options else Options() - self.service.path = DriverFinder.get_path(self.service, options) + self.service.path = DriverFinder(self.service, options).get_driver_path() if not self.service.reuse_service: self.service.start() diff --git a/py/selenium/webdriver/webkitgtk/webdriver.py b/py/selenium/webdriver/webkitgtk/webdriver.py index b364ca5f192be..4918d3b0d2904 100644 --- a/py/selenium/webdriver/webkitgtk/webdriver.py +++ b/py/selenium/webdriver/webkitgtk/webdriver.py @@ -60,7 +60,7 @@ def __init__( desired_capabilities = capabilities self.service = Service(executable_path, port=port, log_path=service_log_path) - self.service.path = DriverFinder.get_path(self.service, options) + self.service.path = DriverFinder(self.service, options).get_driver_path() self.service.start() super().__init__( diff --git a/py/selenium/webdriver/wpewebkit/webdriver.py b/py/selenium/webdriver/wpewebkit/webdriver.py index 0e69489fb5689..90af641ab48ca 100644 --- a/py/selenium/webdriver/wpewebkit/webdriver.py +++ b/py/selenium/webdriver/wpewebkit/webdriver.py @@ -44,7 +44,7 @@ def __init__( options = Options() self.service = service if service else Service() - self.service.path = DriverFinder.get_path(self.service, options) + self.service.path = DriverFinder(self.service, options).get_driver_path() self.service.start() super().__init__(command_executor=self.service.service_url, options=options) diff --git a/py/test/selenium/webdriver/common/driver_finder_tests.py b/py/test/selenium/webdriver/common/driver_finder_tests.py new file mode 100644 index 0000000000000..d42c82f2a8ab2 --- /dev/null +++ b/py/test/selenium/webdriver/common/driver_finder_tests.py @@ -0,0 +1,65 @@ +# 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. +from pathlib import Path +from unittest import mock + +import pytest + +from selenium import webdriver +from selenium.common.exceptions import NoSuchDriverException +from selenium.webdriver.common.driver_finder import DriverFinder + + +def test_get_results_with_valid_path(): + options = webdriver.ChromeOptions() + service = webdriver.ChromeService(executable_path="/valid/path/to/driver") + + with mock.patch.object(Path, "is_file", return_value=True): + result = DriverFinder(service, options).get_driver_path() + assert result == "/valid/path/to/driver" + + +def test_errors_with_invalid_path(): + options = webdriver.ChromeOptions() + service = webdriver.ChromeService(executable_path="/invalid/path/to/driver") + + with mock.patch.object(Path, "is_file", return_value=False): + with pytest.raises(NoSuchDriverException) as excinfo: + DriverFinder(service, options).get_driver_path() + assert "Unable to obtain driver for chrome; For documentation on this error" in str(excinfo.value) + + +def test_wraps_error_from_se_manager(): + options = webdriver.ChromeOptions() + service = webdriver.ChromeService(executable_path="/valid/path/to/driver") + + lib_path = "selenium.webdriver.common.selenium_manager.SeleniumManager" + with mock.patch(lib_path + ".binary_paths", side_effect=Exception("Error")): + with pytest.raises(NoSuchDriverException): + DriverFinder(service, options).get_driver_path() + + +def test_get_results_from_se_manager(monkeypatch): + executable_path = "/invalid/path/to/driver" + options = webdriver.ChromeOptions() + service = webdriver.ChromeService(executable_path=executable_path) + monkeypatch.setattr(Path, "is_file", lambda _: True) + + lib_path = "selenium.webdriver.common.selenium_manager.SeleniumManager" + with mock.patch(lib_path + ".binary_paths", return_value=executable_path): + path = DriverFinder(service, options).get_driver_path() + assert path == executable_path diff --git a/py/test/selenium/webdriver/common/selenium_manager_tests.py b/py/test/selenium/webdriver/common/selenium_manager_tests.py index c4a267c3d0f72..8692644cee0c4 100644 --- a/py/test/selenium/webdriver/common/selenium_manager_tests.py +++ b/py/test/selenium/webdriver/common/selenium_manager_tests.py @@ -14,88 +14,109 @@ # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. - -from unittest.mock import Mock +import json +import sys +from pathlib import Path +from unittest import mock import pytest +import selenium from selenium.common.exceptions import WebDriverException -from selenium.webdriver.chrome.options import Options -from selenium.webdriver.common.proxy import Proxy from selenium.webdriver.common.selenium_manager import SeleniumManager -def test_browser_version_is_used_for_sm(mocker): - import subprocess - - mock_run = mocker.patch("subprocess.run") - mocked_result = Mock() - mocked_result.configure_mock( - **{ - "stdout.decode.return_value": '{"result": {"driver_path": "driver", "browser_path": "browser"}, "logs": []}', - "returncode": 0, - } - ) - mock_run.return_value = mocked_result - options = Options() - options.capabilities["browserName"] = "chrome" - options.browser_version = "110" - - _ = SeleniumManager().driver_location(options) - args, kwargs = subprocess.run.call_args - assert "--browser-version" in args[0] - assert "110" in args[0] - - -def test_browser_path_is_used_for_sm(mocker): - import subprocess - - mock_run = mocker.patch("subprocess.run") - mocked_result = Mock() - mocked_result.configure_mock( - **{ - "stdout.decode.return_value": '{"result": {"driver_path": "driver", "browser_path": "browser"}, "logs": []}', - "returncode": 0, - } - ) - mock_run.return_value = mocked_result - options = Options() - options.capabilities["browserName"] = "chrome" - options.binary_location = "/opt/bin/browser-bin" - - _ = SeleniumManager().driver_location(options) - args, kwargs = subprocess.run.call_args - assert "--browser-path" in args[0] - assert "/opt/bin/browser-bin" in args[0] - - -def test_proxy_is_used_for_sm(mocker): - import subprocess - - mock_run = mocker.patch("subprocess.run") - mocked_result = Mock() - mocked_result.configure_mock( - **{ - "stdout.decode.return_value": '{"result": {"driver_path": "driver", "browser_path": "browser"}, "logs": []}', - "returncode": 0, - } - ) - mock_run.return_value = mocked_result - options = Options() - options.capabilities["browserName"] = "chrome" - proxy = Proxy() - proxy.http_proxy = "http-proxy" - options.proxy = proxy - - _ = SeleniumManager().driver_location(options) - args, kwargs = subprocess.run.call_args - assert "--proxy" in args[0] - assert "http-proxy" in args[0] - - -def test_stderr_is_propagated_to_exception_messages(): - msg = r"Unsuccessful command executed:.*\n.* 'Invalid browser name: foo'.*" - with pytest.raises(WebDriverException, match=msg): - manager = SeleniumManager() - binary = manager.get_binary() - _ = manager.run([str(binary), "--browser", "foo"]) +def test_gets_results(monkeypatch): + expected_output = {"driver_path": "/path/to/driver"} + lib_path = "selenium.webdriver.common.selenium_manager.SeleniumManager" + + with mock.patch(lib_path + "._get_binary", return_value="/path/to/sm") as mock_get_binary, mock.patch( + lib_path + "._run", return_value=expected_output + ) as mock_run: + SeleniumManager().binary_paths([]) + + mock_get_binary.assert_called_once() + expected_run_args = ["/path/to/sm", "--language-binding", "python", "--output", "json"] + mock_run.assert_called_once_with(expected_run_args) + + +def test_uses_environment_variable(monkeypatch): + monkeypatch.setenv("SE_MANAGER_PATH", "/path/to/manager") + monkeypatch.setattr(Path, "is_file", lambda _: True) + + binary = SeleniumManager()._get_binary() + + assert str(binary) == "/path/to/manager" + + +def test_uses_windows(monkeypatch): + monkeypatch.setattr(sys, "platform", "win32") + binary = SeleniumManager()._get_binary() + + project_root = Path(selenium.__file__).parent.parent + assert binary == project_root.joinpath("selenium/webdriver/common/windows/selenium-manager.exe") + + +def test_uses_linux(monkeypatch): + monkeypatch.setattr(sys, "platform", "linux") + binary = SeleniumManager()._get_binary() + + project_root = Path(selenium.__file__).parent.parent + assert binary == project_root.joinpath("selenium/webdriver/common/linux/selenium-manager") + + +def test_uses_mac(monkeypatch): + monkeypatch.setattr(sys, "platform", "darwin") + binary = SeleniumManager()._get_binary() + + project_root = Path(selenium.__file__).parent.parent + assert binary == project_root.joinpath("selenium/webdriver/common/macos/selenium-manager") + + +def test_errors_if_not_file(monkeypatch): + monkeypatch.setattr(Path, "is_file", lambda _: False) + + with pytest.raises(WebDriverException) as excinfo: + SeleniumManager()._get_binary() + assert "Unable to obtain working Selenium Manager binary" in str(excinfo.value) + + +def test_errors_if_invalid_os(monkeypatch): + monkeypatch.setattr(sys, "platform", "linux") + monkeypatch.setattr("platform.machine", lambda: "invalid") + + with pytest.raises(WebDriverException) as excinfo: + SeleniumManager()._get_binary() + assert "Unsupported platform/architecture combination" in str(excinfo.value) + + +def test_error_if_invalid_env_path(monkeypatch): + monkeypatch.setenv("SE_MANAGER_PATH", "/path/to/manager") + + with pytest.raises(WebDriverException) as excinfo: + SeleniumManager()._get_binary() + assert "Unable to obtain working Selenium Manager binary; /path/to/manager" in str(excinfo.value) + + +def test_run_successful(): + expected_result = {"driver_path": "/path/to/driver", "browser_path": "/path/to/browser"} + run_output = {"result": expected_result, "logs": []} + with mock.patch("subprocess.run") as mock_run, mock.patch("json.loads", return_value=run_output): + mock_run.return_value = mock.Mock(stdout=json.dumps(run_output).encode("utf-8"), stderr=b"", returncode=0) + result = SeleniumManager._run(["arg1", "arg2"]) + assert result == expected_result + + +def test_run_exception(): + with mock.patch("subprocess.run", side_effect=Exception("Test Error")): + with pytest.raises(WebDriverException) as excinfo: + SeleniumManager._run(["/path/to/sm", "arg1", "arg2"]) + assert "Unsuccessful command executed: /path/to/sm arg1 arg2" in str(excinfo.value) + + +def test_run_non_zero_exit_code(): + with mock.patch("subprocess.run") as mock_run, mock.patch("json.loads", return_value={"result": "", "logs": []}): + mock_run.return_value = mock.Mock(stdout=b"{}", stderr=b"Error Message", returncode=1) + with pytest.raises(WebDriverException) as excinfo: + SeleniumManager._run(["/path/to/sm", "arg1"]) + assert "Unsuccessful command executed: /path/to/sm arg1; code: 1\n\nError Message" in str(excinfo.value)