US47036 Apply proxy settings

bzt/utils.py

@@ -52,6 +52,7 @@
 from subprocess import CalledProcessError, PIPE, check_output, STDOUT
 from urllib import parse
 from urllib.error import URLError
+from urllib.parse import urlparse
 from urllib.request import url2pathname
 from webbrowser import GenericBrowser
 
@@ -1157,7 +1158,21 @@ def __init__(self):
         self.proxy_settings = None
 
     def add_proxy_settings(self, proxy_settings):
+        if os.getenv("APPLY_PROXY_SETTINGS", "False").lower() in 'true':
+            self.log.info('Using proxy settings from environment variables')
+            self.proxy_settings = {}
+            http_proxy = os.getenv("HTTP_PROXY")
+            if http_proxy:
+                self.proxy_settings = self._get_proxy_settings_from_url(http_proxy)
+            https_proxy = os.getenv("HTTPS_PROXY")
+            if https_proxy:
+                self.proxy_settings = self._get_proxy_settings_from_url(https_proxy)
+            no_proxy = os.getenv("NO_PROXY")
+            if no_proxy:
+                self.proxy_settings['nonProxy'] = no_proxy
+
         if proxy_settings and proxy_settings.get("address"):
+            self.log.info('Overriding proxy settings using taurus configuration')
             self.proxy_settings = proxy_settings
             proxy_addr = proxy_settings.get("address")
             self.log.info("Using proxy %r", proxy_addr)
@@ -1172,6 +1187,9 @@ def add_proxy_settings(self, proxy_settings):
                 proxy_uri = "%s://%s" % (scheme, proxy_url.netloc)
             self.session.proxies = {"https": proxy_uri, "http": proxy_uri}
 
+        if not self.proxy_settings:
+            self.log.warning('Proxy settings not set')
+
         self.session.verify = proxy_settings.get('ssl-cert', True)
         self.session.cert = proxy_settings.get('ssl-client-cert', None)
 
@@ -1184,7 +1202,10 @@ def get_proxy_props(self):
         proxy_url = parse.urlsplit(self.proxy_settings.get("address"))
         username = self.proxy_settings.get("username")
         pwd = self.proxy_settings.get("password")
+        non_proxy = self.proxy_settings.get("nonProxy")
         for protocol in ["http", "https"]:
+            if non_proxy:
+                props[protocol + '.nonProxyHosts'] = non_proxy
             props[protocol + '.proxyHost'] = proxy_url.hostname
             props[protocol + '.proxyPort'] = proxy_url.port or 80
             if username and pwd:
@@ -1208,6 +1229,25 @@ def _save_file_from_connection(conn, filename, reporthook=None):
                     if reporthook:
                         reporthook(count, block_size, total)
 
+    @staticmethod
+    def _get_proxy_settings_from_url(url):
+        proxy_settings = {}
+        if not url:
+            return proxy_settings
+        proxy_url = urlparse(url)
+        if proxy_url.username:
+            proxy_settings['username'] = proxy_url.username
+        if proxy_url.password:
+            proxy_settings['password'] = proxy_url.password
+        port = ""
+        if proxy_url.port:
+            port = ":" + str(proxy_url.port)
+        if proxy_url.username or proxy_url.password:
+            proxy_url = proxy_url._replace(netloc=f"{proxy_url.hostname}{port}",
+                                           )
+        proxy_settings['address'] = proxy_url.geturl()
+        return proxy_settings
+
     def download_file(self, url, filename, reporthook=None, data=None, timeout=None):
         headers = None
         try:

tests/unit/test_utils.py

@@ -318,7 +318,8 @@ def test_proxy_setup(self):
         obj = HTTPClient()
         obj.add_proxy_settings({"address": "http://localhost:3128",
                                 "username": "me",
-                                "password": "too"})
+                                "password": "too",
+                                "nonProxy": "localhost"})
 
         self.assertIn('http', obj.session.proxies)
         self.assertIn('https', obj.session.proxies)
@@ -338,13 +339,51 @@ def test_jvm_args(self):
         obj = HTTPClient()
         obj.add_proxy_settings({"address": "http://localhost:3128",
                                 "username": "me",
-                                "password": "too"})
+                                "password": "too",
+                                "nonProxy": "localhost"})
         jvm_args = obj.get_proxy_props()
         for protocol in ['http', 'https']:
-            for key in ['proxyHost', 'proxyPort', 'proxyUser', 'proxyPass']:
+            for key in ['proxyHost', 'proxyPort', 'proxyUser', 'proxyPass', 'nonProxyHosts']:
                 combo_key = protocol + '.' + key
                 self.assertIn(combo_key, jvm_args)
 
+    def test_get_proxy_settings_from_url(self):
+        obj = HTTPClient()
+        settings = obj._get_proxy_settings_from_url("http://10.0.0.0:8080")
+        self.assertEqual('http://10.0.0.0:8080', settings['address'])
+        self.assertFalse('username' in settings)
+        self.assertFalse('password' in settings)
+        settings = obj._get_proxy_settings_from_url("http://10.0.0.0")
+        self.assertEqual('http://10.0.0.0', settings['address'])
+        self.assertFalse('username' in settings)
+        self.assertFalse('password' in settings)
+        settings = obj._get_proxy_settings_from_url("http://user:@10.0.0.0")
+        self.assertEqual('http://10.0.0.0', settings['address'])
+        self.assertEqual("user", settings['username'])
+        self.assertFalse('password' in settings)
+        settings = obj._get_proxy_settings_from_url("http://user:pass@10.0.0.0")
+        self.assertEqual('http://10.0.0.0', settings['address'])
+        self.assertEqual("user", settings['username'])
+        self.assertEqual('pass', settings['password'])
+
+        settings = obj._get_proxy_settings_from_url("https://10.0.0.0:8080")
+        self.assertEqual('https://10.0.0.0:8080', settings['address'])
+        self.assertFalse('username' in settings)
+        self.assertFalse('password' in settings)
+        settings = obj._get_proxy_settings_from_url("https://10.0.0.0")
+        self.assertEqual('https://10.0.0.0', settings['address'])
+        self.assertFalse('username' in settings)
+        self.assertFalse('password' in settings)
+        settings = obj._get_proxy_settings_from_url("https://user:@10.0.0.0")
+        self.assertEqual('https://10.0.0.0', settings['address'])
+        self.assertEqual("user", settings['username'])
+        self.assertFalse('password' in settings)
+        settings = obj._get_proxy_settings_from_url("https://user:pass@10.0.0.0")
+        self.assertEqual('https://10.0.0.0', settings['address'])
+        self.assertEqual("user", settings['username'])
+        self.assertEqual('pass', settings['password'])
+
+
     def test_download_file(self):
         obj = HTTPClient()
         tmpfile = temp_file()