add no_proxy support to stream/ws_client.py

stream/ws_client.py

@@ -30,6 +30,7 @@ from six import StringIO
 
 from websocket import WebSocket, ABNF, enableTrace
 from base64 import urlsafe_b64decode
+from requests.utils import should_bypass_proxies
 
 STDIN_CHANNEL = 0
 STDOUT_CHANNEL = 1
@@ -457,6 +458,12 @@ def create_websocket(configuration, url, headers=None):
     return websocket
 
 def websocket_proxycare(connect_opt, configuration, url, headers):
+    """ An internal function to be called in api-client when a websocket
+        create is requested.
+    """
+    if configuration.no_proxy:
+        connect_opt.update({ 'http_no_proxy': configuration.no_proxy.split(',') })
+
     if configuration.proxy:
         proxy_url = urlparse(configuration.proxy)
         connect_opt.update({'http_proxy_host': proxy_url.hostname, 'http_proxy_port': proxy_url.port})

stream/ws_client_test.py

@@ -47,20 +47,30 @@ class WSClientTest(unittest.TestCase):
             self.assertEqual(get_websocket_url(url), ws_url)
 
     def test_websocket_proxycare(self):
-        for proxy, idpass, expect_host, expect_port, expect_auth in [
-                ( None,                             None,        None,                None, None ),
-                ( 'http://proxy.example.com:8080/', None,        'proxy.example.com', 8080, None ),
-                ( 'http://proxy.example.com:8080/', 'user:pass', 'proxy.example.com', 8080, ('user','pass'))
+        for proxy, idpass, no_proxy, expect_host, expect_port, expect_auth, expect_noproxy in [
+                ( None,                             None,        None,                            None,                None, None, None ),
+                ( 'http://proxy.example.com:8080/', None,        None,                            'proxy.example.com', 8080, None, None ),
+                ( 'http://proxy.example.com:8080/', 'user:pass', None,                            'proxy.example.com', 8080, ('user','pass'), None),
+                ( 'http://proxy.example.com:8080/', 'user:pass', '',                              'proxy.example.com', 8080, ('user','pass'), None),
+                ( 'http://proxy.example.com:8080/', 'user:pass', '*',                             'proxy.example.com', 8080, ('user','pass'), ['*']),
+                ( 'http://proxy.example.com:8080/', 'user:pass', '.example.com',                  'proxy.example.com', 8080, ('user','pass'), ['.example.com']),
+                ( 'http://proxy.example.com:8080/', 'user:pass', 'localhost,.local,.example.com',  'proxy.example.com', 8080, ('user','pass'), ['localhost','.local','.example.com']),
                 ]:
+            # setup input
             config = Configuration()
             if proxy is not None:
                 setattr(config, 'proxy', proxy)
             if idpass is not None:
                 setattr(config, 'proxy_headers', urllib3.util.make_headers(proxy_basic_auth=idpass))
+            if no_proxy is not None:
+                setattr(config, 'no_proxy', no_proxy)
+            # setup done
+            # test starts
             connect_opt = websocket_proxycare( {}, config, None, None)
             self.assertEqual( dictval(connect_opt,'http_proxy_host'), expect_host)
             self.assertEqual( dictval(connect_opt,'http_proxy_port'), expect_port)
             self.assertEqual( dictval(connect_opt,'http_proxy_auth'), expect_auth)
+            self.assertEqual( dictval(connect_opt,'http_no_proxy'), expect_noproxy)
 
 if __name__ == '__main__':
     unittest.main()