kubernetes-client/python
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #260 from itaru2622/no_proxy

Browse Source 
add no_proxy support to websocket client
This commit is contained in:
Kubernetes Prow Robot 2021-10-18 15:21:21 -07:00 committed by GitHub
commit 09dbbe521e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 21 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
stream/ws_client.py
View File

@ -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})

18
stream/ws_client_test.py
View File

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