[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 <diemol@users.noreply.github.com>

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

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(

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

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

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(

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(

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()

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__(

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)