Add options for downloading drivers via proxy as needed
This commit is contained in:
parent
d70426b9f3
commit
0881de9786
|
@ -518,6 +518,7 @@ pytest my_first_test.py --pdb
|
|||
--proxy-bypass-list=STRING # (";"-separated hosts to bypass, Eg "*.foo.com")
|
||||
--proxy-pac-url=URL # (Connect to a proxy server using a PAC_URL.pac file.)
|
||||
--proxy-pac-url=USERNAME:PASSWORD@URL # (Authenticated proxy with PAC URL.)
|
||||
--proxy-driver # (If a driver download is needed, will use: --proxy=PROXY.)
|
||||
--multi-proxy # (Allow multiple authenticated proxies when multi-threaded.)
|
||||
--agent=STRING # (Modify the web browser's User-Agent string.)
|
||||
--mobile # (Use the mobile device emulator while running tests.)
|
||||
|
|
|
@ -124,6 +124,7 @@ pytest my_first_test.py --settings-file=custom_settings.py
|
|||
--proxy-bypass-list=STRING # (";"-separated hosts to bypass, Eg "*.foo.com")
|
||||
--proxy-pac-url=URL # (Connect to a proxy server using a PAC_URL.pac file.)
|
||||
--proxy-pac-url=USERNAME:PASSWORD@URL # (Authenticated proxy with PAC URL.)
|
||||
--proxy-driver # (If a driver download is needed, will use: --proxy=PROXY.)
|
||||
--multi-proxy # (Allow multiple authenticated proxies when multi-threaded.)
|
||||
--agent=STRING # (Modify the web browser's User-Agent string.)
|
||||
--mobile # (Use the mobile device emulator while running tests.)
|
||||
|
|
|
@ -40,6 +40,7 @@ import colorama
|
|||
import sys
|
||||
import time
|
||||
from seleniumbase.fixtures import constants
|
||||
from seleniumbase.fixtures import shared_utils
|
||||
|
||||
colorama.init(autoreset=True)
|
||||
|
||||
|
@ -968,6 +969,14 @@ def main():
|
|||
need_another_retry = False
|
||||
retry_msg_1 = "* Unable to download driver! Retrying in 3s..."
|
||||
retry_msg_2 = "** Unable to download driver! Retrying in 5s..."
|
||||
if " --proxy=" in " ".join(sys.argv):
|
||||
for arg in sys.argv:
|
||||
if arg.startswith("--proxy="):
|
||||
proxy_string = arg.split("--proxy=")[1]
|
||||
if "@" in proxy_string:
|
||||
proxy_string = proxy_string.split("@")[1]
|
||||
shared_utils.validate_proxy_string(proxy_string)
|
||||
break
|
||||
try:
|
||||
sb_install.main()
|
||||
except Exception as e:
|
||||
|
|
|
@ -36,6 +36,8 @@ import tarfile
|
|||
import urllib3
|
||||
import zipfile
|
||||
from seleniumbase.fixtures import constants
|
||||
from seleniumbase.fixtures import shared_utils
|
||||
from seleniumbase import config as sb_config
|
||||
from seleniumbase import drivers # webdriver storage folder for SeleniumBase
|
||||
|
||||
urllib3.disable_warnings()
|
||||
|
@ -118,19 +120,66 @@ def requests_get(url):
|
|||
|
||||
|
||||
def requests_get_with_retry(url):
|
||||
use_proxy = None
|
||||
protocol = "http"
|
||||
user_and_pass = None
|
||||
if " --proxy=" in " ".join(sys.argv):
|
||||
for arg in sys.argv:
|
||||
if arg.startswith("--proxy="):
|
||||
proxy_string = arg.split("--proxy=")[1]
|
||||
if "@" in proxy_string:
|
||||
# Format => username:password@hostname:port
|
||||
try:
|
||||
user_and_pass = proxy_string.split("@")[0]
|
||||
proxy_string = proxy_string.split("@")[1]
|
||||
except Exception:
|
||||
raise Exception(
|
||||
"The format for using a proxy server with auth "
|
||||
'is: "username:password@hostname:port". If not '
|
||||
'using auth, the format is: "hostname:port".'
|
||||
)
|
||||
if proxy_string.endswith(":443"):
|
||||
protocol = "https"
|
||||
elif "socks4" in proxy_string:
|
||||
protocol = "socks4"
|
||||
elif "socks5" in proxy_string:
|
||||
protocol = "socks5"
|
||||
proxy_string = shared_utils.validate_proxy_string(proxy_string)
|
||||
if user_and_pass:
|
||||
proxy_string = "%s@%s" % (user_and_pass, proxy_string)
|
||||
use_proxy = True
|
||||
break
|
||||
response = None
|
||||
try:
|
||||
response = requests.get(url)
|
||||
except Exception:
|
||||
import time
|
||||
if use_proxy:
|
||||
proxies = {protocol: proxy_string}
|
||||
try:
|
||||
response = requests.get(url, proxies=proxies)
|
||||
except Exception:
|
||||
import time
|
||||
|
||||
time.sleep(0.75)
|
||||
response = requests.get(url)
|
||||
return response
|
||||
time.sleep(0.75)
|
||||
response = requests.get(url, proxies=proxies)
|
||||
return response
|
||||
else:
|
||||
try:
|
||||
response = requests.get(url)
|
||||
except Exception:
|
||||
import time
|
||||
|
||||
time.sleep(0.75)
|
||||
response = requests.get(url)
|
||||
return response
|
||||
|
||||
|
||||
def main(override=None, intel_for_uc=None):
|
||||
if override:
|
||||
found_proxy = None
|
||||
if hasattr(sb_config, "proxy_driver") and sb_config.proxy_driver:
|
||||
if " --proxy=" in " ".join(sys.argv):
|
||||
for arg in sys.argv:
|
||||
if arg.startswith("--proxy="):
|
||||
found_proxy = arg
|
||||
break
|
||||
if override == "chromedriver":
|
||||
sys.argv = ["seleniumbase", "get", "chromedriver"]
|
||||
elif override.startswith("chromedriver "):
|
||||
|
@ -151,6 +200,8 @@ def main(override=None, intel_for_uc=None):
|
|||
elif override.startswith("iedriver "):
|
||||
extra = override.split("iedriver ")[1]
|
||||
sys.argv = ["seleniumbase", "get", "iedriver", extra]
|
||||
if found_proxy:
|
||||
sys.argv.append(found_proxy)
|
||||
|
||||
num_args = len(sys.argv)
|
||||
if (
|
||||
|
|
|
@ -972,69 +972,6 @@ def _set_firefox_options(
|
|||
return options
|
||||
|
||||
|
||||
def display_proxy_warning(proxy_string):
|
||||
message = (
|
||||
'\n\nWARNING: Proxy String ["%s"] is NOT in the expected '
|
||||
'"ip_address:port" or "server:port" format, '
|
||||
"(OR the key does not exist in "
|
||||
"seleniumbase.config.proxy_list.PROXY_LIST)." % proxy_string
|
||||
)
|
||||
if settings.RAISE_INVALID_PROXY_STRING_EXCEPTION:
|
||||
raise Exception(message)
|
||||
else:
|
||||
message += " *** DEFAULTING to NOT USING a Proxy Server! ***"
|
||||
warnings.simplefilter("always", Warning) # See Warnings
|
||||
warnings.warn(message, category=Warning, stacklevel=2)
|
||||
warnings.simplefilter("default", Warning) # Set Default
|
||||
|
||||
|
||||
def validate_proxy_string(proxy_string):
|
||||
from seleniumbase.config import proxy_list
|
||||
from seleniumbase.fixtures import page_utils
|
||||
|
||||
if proxy_string in proxy_list.PROXY_LIST.keys():
|
||||
proxy_string = proxy_list.PROXY_LIST[proxy_string]
|
||||
if not proxy_string:
|
||||
return None
|
||||
valid = False
|
||||
val_ip = re.match(
|
||||
r"^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}:\d+$", proxy_string
|
||||
)
|
||||
if not val_ip:
|
||||
if proxy_string.startswith("http://"):
|
||||
proxy_string = proxy_string.split("http://")[1]
|
||||
elif proxy_string.startswith("https://"):
|
||||
proxy_string = proxy_string.split("https://")[1]
|
||||
elif "://" in proxy_string:
|
||||
if not proxy_string.startswith("socks4://") and not (
|
||||
proxy_string.startswith("socks5://")
|
||||
):
|
||||
proxy_string = proxy_string.split("://")[1]
|
||||
chunks = proxy_string.split(":")
|
||||
if len(chunks) == 2:
|
||||
if re.match(r"^\d+$", chunks[1]):
|
||||
if page_utils.is_valid_url("http://" + proxy_string):
|
||||
valid = True
|
||||
elif len(chunks) == 3:
|
||||
if re.match(r"^\d+$", chunks[2]):
|
||||
if page_utils.is_valid_url("http:" + ":".join(chunks[1:])):
|
||||
if chunks[0] == "http":
|
||||
valid = True
|
||||
elif chunks[0] == "https":
|
||||
valid = True
|
||||
elif chunks[0] == "socks4":
|
||||
valid = True
|
||||
elif chunks[0] == "socks5":
|
||||
valid = True
|
||||
else:
|
||||
proxy_string = val_ip.group()
|
||||
valid = True
|
||||
if not valid:
|
||||
display_proxy_warning(proxy_string)
|
||||
proxy_string = None
|
||||
return proxy_string
|
||||
|
||||
|
||||
def get_driver(
|
||||
browser_name=None,
|
||||
headless=False,
|
||||
|
@ -1162,7 +1099,7 @@ def get_driver(
|
|||
"that has authentication! (If using a proxy server "
|
||||
"without auth, Chrome, Edge, or Firefox may be used.)"
|
||||
)
|
||||
proxy_string = validate_proxy_string(proxy_string)
|
||||
proxy_string = shared_utils.validate_proxy_string(proxy_string)
|
||||
if proxy_string and proxy_user and proxy_pass:
|
||||
proxy_auth = True
|
||||
elif proxy_pac_url:
|
||||
|
|
|
@ -1,7 +1,7 @@
|
|||
"""Shared utility methods."""
|
||||
|
||||
import subprocess
|
||||
import sys
|
||||
from seleniumbase.config import settings
|
||||
from seleniumbase.fixtures import constants
|
||||
from seleniumbase import config as sb_config
|
||||
|
||||
|
@ -48,6 +48,72 @@ def get_terminal_width():
|
|||
return width
|
||||
|
||||
|
||||
def display_proxy_warning(proxy_string):
|
||||
import warnings
|
||||
|
||||
message = (
|
||||
'\nWARNING: Proxy String ["%s"] is NOT in the expected '
|
||||
'"ip_address:port" or "server:port" format, '
|
||||
"(OR the key does not exist in "
|
||||
"seleniumbase.config.proxy_list.PROXY_LIST)." % proxy_string
|
||||
)
|
||||
if settings.RAISE_INVALID_PROXY_STRING_EXCEPTION:
|
||||
raise Exception(message)
|
||||
else:
|
||||
message += " *** DEFAULTING to NOT USING a Proxy Server! ***"
|
||||
warnings.simplefilter("always", Warning) # See Warnings
|
||||
warnings.warn(message, category=Warning, stacklevel=2)
|
||||
warnings.simplefilter("default", Warning) # Set Default
|
||||
|
||||
|
||||
def validate_proxy_string(proxy_string):
|
||||
import re
|
||||
from seleniumbase.config import proxy_list
|
||||
from seleniumbase.fixtures import page_utils
|
||||
|
||||
if proxy_string in proxy_list.PROXY_LIST.keys():
|
||||
proxy_string = proxy_list.PROXY_LIST[proxy_string]
|
||||
if not proxy_string:
|
||||
return None
|
||||
valid = False
|
||||
val_ip = re.match(
|
||||
r"^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}:\d+$", proxy_string
|
||||
)
|
||||
if not val_ip:
|
||||
if proxy_string.startswith("http://"):
|
||||
proxy_string = proxy_string.split("http://")[1]
|
||||
elif proxy_string.startswith("https://"):
|
||||
proxy_string = proxy_string.split("https://")[1]
|
||||
elif "://" in proxy_string:
|
||||
if not proxy_string.startswith("socks4://") and not (
|
||||
proxy_string.startswith("socks5://")
|
||||
):
|
||||
proxy_string = proxy_string.split("://")[1]
|
||||
chunks = proxy_string.split(":")
|
||||
if len(chunks) == 2:
|
||||
if re.match(r"^\d+$", chunks[1]):
|
||||
if page_utils.is_valid_url("http://" + proxy_string):
|
||||
valid = True
|
||||
elif len(chunks) == 3:
|
||||
if re.match(r"^\d+$", chunks[2]):
|
||||
if page_utils.is_valid_url("http:" + ":".join(chunks[1:])):
|
||||
if chunks[0] == "http":
|
||||
valid = True
|
||||
elif chunks[0] == "https":
|
||||
valid = True
|
||||
elif chunks[0] == "socks4":
|
||||
valid = True
|
||||
elif chunks[0] == "socks5":
|
||||
valid = True
|
||||
else:
|
||||
proxy_string = val_ip.group()
|
||||
valid = True
|
||||
if not valid:
|
||||
display_proxy_warning(proxy_string)
|
||||
proxy_string = None
|
||||
return proxy_string
|
||||
|
||||
|
||||
def format_exc(exception, message):
|
||||
"""Formats an exception message to make the output cleaner."""
|
||||
from selenium.common.exceptions import ElementNotVisibleException
|
||||
|
|
|
@ -45,6 +45,7 @@ def pytest_addoption(parser):
|
|||
--proxy-bypass-list=STRING (";"-separated hosts to bypass, Eg "*.foo.com")
|
||||
--proxy-pac-url=URL (Connect to a proxy server using a PAC_URL.pac file.)
|
||||
--proxy-pac-url=USERNAME:PASSWORD@URL (Authenticated proxy with PAC URL.)
|
||||
--proxy-driver (If a driver download is needed, will use: --proxy=PROXY.)
|
||||
--multi-proxy (Allow multiple authenticated proxies when multi-threaded.)
|
||||
--agent=STRING (Modify the web browser's User-Agent string.)
|
||||
--mobile (Use the mobile device emulator while running tests.)
|
||||
|
@ -489,6 +490,15 @@ def pytest_addoption(parser):
|
|||
A username:password@URL string
|
||||
Default: None.""",
|
||||
)
|
||||
parser.addoption(
|
||||
"--proxy-driver",
|
||||
"--proxy_driver",
|
||||
action="store_true",
|
||||
dest="proxy_driver",
|
||||
default=False,
|
||||
help="""If a driver download is needed for tests,
|
||||
uses proxy settings set via --proxy=PROXY.""",
|
||||
)
|
||||
parser.addoption(
|
||||
"--multi-proxy",
|
||||
"--multi_proxy",
|
||||
|
@ -1432,6 +1442,7 @@ def pytest_configure(config):
|
|||
sb_config.proxy_string = config.getoption("proxy_string")
|
||||
sb_config.proxy_bypass_list = config.getoption("proxy_bypass_list")
|
||||
sb_config.proxy_pac_url = config.getoption("proxy_pac_url")
|
||||
sb_config.proxy_driver = config.getoption("proxy_driver")
|
||||
sb_config.multi_proxy = config.getoption("multi_proxy")
|
||||
sb_config.cap_file = config.getoption("cap_file")
|
||||
sb_config.cap_string = config.getoption("cap_string")
|
||||
|
|
|
@ -359,6 +359,9 @@ def SB(
|
|||
if recorder_mode and headless:
|
||||
headless = False
|
||||
headless2 = True
|
||||
sb_config.proxy_driver = False
|
||||
if "--proxy-driver" in sys_argv or "--proxy_driver" in sys_argv:
|
||||
sb_config.proxy_driver = True
|
||||
if variables and type(variables) is str and len(variables) > 0:
|
||||
import ast
|
||||
|
||||
|
|
|
@ -29,6 +29,7 @@ class SeleniumBrowser(Plugin):
|
|||
--proxy-bypass-list=STRING (";"-separated hosts to bypass, Eg "*.foo.com")
|
||||
--proxy-pac-url=URL (Connect to a proxy server using a PAC_URL.pac file.)
|
||||
--proxy-pac-url=USERNAME:PASSWORD@URL (Authenticated proxy with PAC URL.)
|
||||
--proxy-driver (If a driver download is needed, will use: --proxy=PROXY.)
|
||||
--multi-proxy (Allow multiple authenticated proxies when multi-threaded.)
|
||||
--agent=STRING (Modify the web browser's User-Agent string.)
|
||||
--mobile (Use the mobile device emulator while running tests.)
|
||||
|
@ -256,6 +257,15 @@ class SeleniumBrowser(Plugin):
|
|||
A username:password@URL string
|
||||
Default: None.""",
|
||||
)
|
||||
parser.addoption(
|
||||
"--proxy-driver",
|
||||
"--proxy_driver",
|
||||
action="store_true",
|
||||
dest="proxy_driver",
|
||||
default=False,
|
||||
help="""If a driver download is needed for tests,
|
||||
uses proxy settings set via --proxy=PROXY.""",
|
||||
)
|
||||
parser.addoption(
|
||||
"--multi-proxy",
|
||||
"--multi_proxy",
|
||||
|
@ -1207,6 +1217,7 @@ class SeleniumBrowser(Plugin):
|
|||
sb_config._SMALL_TIMEOUT = settings.SMALL_TIMEOUT
|
||||
sb_config._LARGE_TIMEOUT = settings.LARGE_TIMEOUT
|
||||
sb_config._context_of_runner = False # Context Manager Compatibility
|
||||
sb_config.proxy_driver = self.options.proxy_driver
|
||||
sb_config.multi_proxy = self.options.multi_proxy
|
||||
# The driver will be received later
|
||||
self.driver = None
|
||||
|
|
Loading…
Reference in New Issue