Fix proxy URL schemes, the failed proxy lookup and the Selenium 4 fallback

scholarly/_navigator.py

@@ -177,7 +177,7 @@ def _get_page(self, pagerequest: str, premium: bool = False) -> str:
 
             tries += 1
             try:
-                session, timeout = pm.get_next_proxy(num_tries = tries, old_timeout = timeout, old_proxy=pm._proxies.get('http', None))
+                session, timeout = pm.get_next_proxy(num_tries = tries, old_timeout = timeout, old_proxy=pm._proxies.get('http://', None))
             except Exception:
                 self.logger.info("No other secondary connections possible. "
                                  "Using the primary proxy for all requests.")

scholarly/_proxy_generator.py

@@ -1,6 +1,7 @@
 from typing import Callable
 from fp.fp import FreeProxy
 import random
+import shutil
 import logging
 import time
 import requests
@@ -13,6 +14,9 @@
 from selenium.webdriver.common.by import By
 from selenium.common.exceptions import WebDriverException, UnexpectedAlertPresentException
 from selenium.webdriver.firefox.options import Options as FirefoxOptions
+from selenium.webdriver.chrome.service import Service as ChromeService
+from selenium.webdriver.firefox.service import Service as FirefoxService
+from selenium.webdriver.common.proxy import Proxy, ProxyType
 from urllib.parse import urlparse
 from contextlib import contextmanager
 from deprecated import deprecated
@@ -190,13 +194,14 @@ def _use_proxy(self, http: str, https: str = None) -> bool:
         :returns: whether or not the proxy was set up successfully
         :rtype: {bool}
         """
-        # Reformat proxy for HTTPX
-        if http[:4] not in ("http", "sock"):
+        # Reformat proxy for HTTPX. Only add a scheme when none is given, so
+        # that an http:// proxy for https and socks5:// proxies are kept as is.
+        if "://" not in http:
             http = "http://" + http
         if https is None:
             https = http
-        elif https[:5] not in ("https", "socks"):
-            https = "https://" + https
+        elif "://" not in https:
+            https = "http://" + https
 
         proxies = {'http://': http, 'https://': https}
         if self.proxy_mode == ProxyMode.SCRAPERAPI:
@@ -365,32 +370,23 @@ def _get_webdriver(self):
                 self.logger.info("Neither Chrome nor Firefox/Geckodriver found in PATH")
 
     def _get_chrome_webdriver(self):
-        if self._proxy_works:
-            webdriver.DesiredCapabilities.CHROME['proxy'] = {
-                "httpProxy": self._proxies['http://'],
-                "sslProxy": self._proxies['https://'],
-                "proxyType": "MANUAL"
-            }
-
         options = webdriver.ChromeOptions()
         options.add_argument('--headless')
-        self._webdriver = webdriver.Chrome('chromedriver', options=options)
+        if self._proxy_works:
+            options.add_argument("--proxy-server=http={};https={}".format(
+                self._proxies['http://'], self._proxies['https://']))
+        self._webdriver = webdriver.Chrome(options=options, **self._driver_service('chromedriver', ChromeService))
         self._webdriver.get("https://scholar.google.com")  # Need to pre-load to set cookies later
 
         return self._webdriver
 
     def _get_firefox_webdriver(self):
-        if self._proxy_works:
-            # Redirect webdriver through proxy
-            webdriver.DesiredCapabilities.FIREFOX['proxy'] = {
-                "httpProxy": self._proxies['http://'],
-                "sslProxy": self._proxies['https://'],
-                "proxyType": "MANUAL",
-            }
-
         options = FirefoxOptions()
         options.add_argument('--headless')
-        self._webdriver = webdriver.Firefox(options=options)
+        if self._proxy_works:
+            # Redirect webdriver through proxy
+            options.proxy = self._selenium_proxy(self._proxies)
+        self._webdriver = webdriver.Firefox(options=options, **self._driver_service('geckodriver', FirefoxService))
         self._webdriver.get("https://scholar.google.com")  # Need to pre-load to set cookies later
 
         # It might make sense to (pre)set cookies as well, e.g., to set a GSP ID.
@@ -494,6 +490,30 @@ def _new_session(self, **kwargs):
 
         return self._session
 
+    @staticmethod
+    def _driver_service(name, service_class):
+        """Use the driver found on PATH, if any. Otherwise Selenium Manager
+        looks for one (or downloads it, where it supports the platform).
+        """
+        driver = shutil.which(name)
+        return {"service": service_class(executable_path=driver)} if driver else {}
+
+    @staticmethod
+    def _selenium_proxy(proxies):
+        """Translate a {'http://': url, 'https://': url} mapping into a
+        Selenium Proxy (host:port entries, socks URLs as socksProxy).
+        """
+        cfg = {"proxyType": ProxyType.MANUAL}
+        for key, url in (("httpProxy", proxies['http://']),
+                         ("sslProxy", proxies['https://'])):
+            u = urlparse(url)
+            if u.scheme.startswith("socks"):
+                cfg.update(socksProxy=f"{u.hostname}:{u.port}",
+                           socksVersion=int(u.scheme[5:6] or 5))
+            else:
+                cfg[key] = f"{u.hostname}:{u.port}"
+        return Proxy(cfg)
+
     def _close_session(self):
         if self._session:
             self._session.close()

test_module.py

@@ -11,6 +11,7 @@
 import requests
 from bs4 import BeautifulSoup
 from contextlib import contextmanager
+from unittest import mock
 try:
     import pandas as pd
 except ImportError:
@@ -892,5 +893,75 @@ def test_citedby(self):
         top10_citations = [citation for num, citation in enumerate(scholarly.citedby(pub)) if num<10]
         self.assertEqual(len(top10_citations), 10)
 
+
+class TestProxyGeneratorOffline(unittest.TestCase):
+    """Tests of the proxy setup that do not need network access."""
+
+    def setUp(self):
+        self.pg = ProxyGenerator()
+        self.pg._check_proxy = lambda proxies: True
+        self.pg._new_session = lambda **kwargs: None
+
+    def test_use_proxy_adds_missing_scheme(self):
+        self.assertTrue(self.pg._use_proxy("127.0.0.1:8080"))
+        self.assertEqual(self.pg._proxies, {'http://': "http://127.0.0.1:8080",
+                                            'https://': "http://127.0.0.1:8080"})
+
+        self.pg._use_proxy("127.0.0.1:8080", "127.0.0.1:8443")
+        self.assertEqual(self.pg._proxies, {'http://': "http://127.0.0.1:8080",
+                                            'https://': "http://127.0.0.1:8443"})
+
+    def test_use_proxy_keeps_given_scheme(self):
+        self.pg._use_proxy("http://127.0.0.1:8080", "http://127.0.0.1:8443")
+        self.assertEqual(self.pg._proxies, {'http://': "http://127.0.0.1:8080",
+                                            'https://': "http://127.0.0.1:8443"})
+
+        self.pg._use_proxy("socks5://127.0.0.1:9050")
+        self.assertEqual(self.pg._proxies, {'http://': "socks5://127.0.0.1:9050",
+                                            'https://': "socks5://127.0.0.1:9050"})
+
+
+    def _webdriver(self, browser, proxies=None, driver_path=None):
+        if proxies:
+            self.pg._proxy_works = True
+            self.pg._proxies = proxies
+        with mock.patch("selenium.webdriver." + browser) as driver_class,                 mock.patch("shutil.which", return_value=driver_path):
+            getattr(self.pg, "_get_%s_webdriver" % browser.lower())()
+        driver_class.return_value.get.assert_called_once_with("https://scholar.google.com")
+        args, kwargs = driver_class.call_args
+        self.assertEqual(args, ())
+        return kwargs
+
+    def test_chrome_webdriver(self):
+        kwargs = self._webdriver("Chrome")
+        self.assertNotIn("service", kwargs)
+        self.assertEqual(kwargs["options"].arguments, ["--headless"])
+
+    def test_chrome_webdriver_proxy(self):
+        kwargs = self._webdriver("Chrome", {'http://': "http://127.0.0.1:8080",
+                                            'https://': "http://127.0.0.1:8080"},
+                                 driver_path="/usr/local/bin/chromedriver")
+        self.assertEqual(kwargs["service"].path, "/usr/local/bin/chromedriver")
+        self.assertIn("--proxy-server=http=http://127.0.0.1:8080;https=http://127.0.0.1:8080",
+                      kwargs["options"].arguments)
+
+    def test_firefox_webdriver_proxy(self):
+        kwargs = self._webdriver("Firefox", {'http://': "http://127.0.0.1:8080",
+                                             'https://': "http://127.0.0.1:8443"},
+                                 driver_path="/usr/local/bin/geckodriver")
+        self.assertEqual(kwargs["service"].path, "/usr/local/bin/geckodriver")
+        proxy = kwargs["options"].proxy
+        self.assertEqual(proxy.http_proxy, "127.0.0.1:8080")
+        self.assertEqual(proxy.ssl_proxy, "127.0.0.1:8443")
+
+    def test_firefox_webdriver_socks_proxy(self):
+        kwargs = self._webdriver("Firefox", {'http://': "socks5://127.0.0.1:9050",
+                                             'https://': "socks5://127.0.0.1:9050"})
+        self.assertNotIn("service", kwargs)
+        proxy = kwargs["options"].proxy
+        self.assertEqual(proxy.socks_proxy, "127.0.0.1:9050")
+        self.assertEqual(proxy.socks_version, 5)
+
+
 if __name__ == '__main__':
     unittest.main()