Update ws_client_test.py

Update configuration.py
What type of PR is this?
/kind bug

What this PR does / why we need it:
This PRs will read environment variables assigned for proxy and no_proxy.

Author: p172913

kubernetes/base/stream/ws_client_test.py

@@ -17,19 +17,70 @@
 from .ws_client import get_websocket_url
 from .ws_client import websocket_proxycare
 from kubernetes.client.configuration import Configuration
+import os
+import socket
+import threading
+import pytest
+from kubernetes import stream, client, config
 
 try:
     import urllib3
     urllib3.disable_warnings()
 except ImportError:
     pass
+@pytest.fixture(autouse=True)
+def dummy_kubeconfig(tmp_path, monkeypatch):
+    # Creating a kubeconfig
+    content = """
+        apiVersion: v1
+        kind: Config
+        clusters:
+        - name: default
+          cluster:
+            server: http://127.0.0.1:8888
+        contexts:
+        - name: default
+          context:
+            cluster: default
+            user: default
+        users:
+        - name: default
+          user: {}
+        current-context: default
+        """
+    cfg_file = tmp_path / "kubeconfig"
+    cfg_file.write_text(content)
+    monkeypatch.setenv("KUBECONFIG", str(cfg_file))
 
-def dictval(dict, key, default=None):
-    try:
-        val = dict[key]
-    except KeyError:
-        val = default
-    return val
+
+def dictval(dict_obj, key, default=None):
+		
+    return dict_obj.get(key, default)
+
+class DummyProxy(threading.Thread):
+    """
+    A minimal HTTP proxy that flags any CONNECT request and returns 200 OK.
+    Listens on 127.0.0.1:8888 by default.
+    """
+    def __init__(self, host='127.0.0.1', port=8888):
+        super().__init__(daemon=True)
+        self.host = host
+        self.port = port
+        self.received_connect = False
+        self._server_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        self._server_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        self._server_sock.bind((self.host, self.port))
+        self._server_sock.listen(1)
+
+    def run(self):
+        conn, _ = self._server_sock.accept()
+        try:
+            data = conn.recv(1024).decode('utf-8', errors='ignore')
+            if data.startswith('CONNECT '):
+                self.received_connect = True
+                conn.sendall(b"HTTP/1.1 200 Connection established\r\n\r\n")
+        finally:
+            conn.close()
 
 class WSClientTest(unittest.TestCase):
 
@@ -56,21 +107,68 @@ def test_websocket_proxycare(self):
                 ( '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))
+            #  input setup
+            cfg = Configuration()
+            if proxy:
+                cfg.proxy = proxy
+            if idpass:
+                cfg.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)
+                cfg.no_proxy = no_proxy
+						
+						 
+            connect_opts = websocket_proxycare({}, cfg, None, None)
+            assert dictval(connect_opts, 'http_proxy_host') == expect_host
+            assert dictval(connect_opts, 'http_proxy_port') == expect_port
+            assert dictval(connect_opts, 'http_proxy_auth') == expect_auth
+            assert dictval(connect_opts, 'http_no_proxy') == expect_noproxy
+
+@pytest.fixture(scope="module")
+def dummy_proxy():
+    #Dummy Proxy
+    proxy = DummyProxy(port=8888)
+    proxy.start()
+    yield proxy
+
+@pytest.fixture(autouse=True)
+def clear_proxy_env(monkeypatch):
+    for var in ("HTTP_PROXY", "http_proxy", "HTTPS_PROXY", "https_proxy", "NO_PROXY", "no_proxy"):
+        monkeypatch.delenv(var, raising=False)
+
+def apply_proxy_to_conf():
+    #apply HTTPS_PROXY env var and set it as global.   
+    cfg = client.Configuration.get_default_copy()
+    cfg.proxy = os.getenv("HTTPS_PROXY")
+    cfg.no_proxy = os.getenv("NO_PROXY", "")
+    client.Configuration.set_default(cfg)
+
+def test_rest_call_ignores_env(dummy_proxy, monkeypatch):
+    # HTTPS_PROXY to dummy proxy
+    monkeypatch.setenv("HTTPS_PROXY", "http://127.0.0.1:8888")
+    # Avoid real HTTP request
+    monkeypatch.setattr(client.CoreV1Api, "list_namespace", lambda self, *_args, **_kwargs: None)
+    # Load config using kubeconfig
+    config.load_kube_config(config_file=os.environ["KUBECONFIG"])
+    apply_proxy_to_conf()
+    # HTTPS_PROXY to dummy proxy
+    monkeypatch.setenv("HTTPS_PROXY", "http://127.0.0.1:8888")
+    config.load_kube_config(config_file=os.environ["KUBECONFIG"])
+    apply_proxy_to_conf()
+    v1 = client.CoreV1Api()
+    v1.list_namespace(_preload_content=False)
+    assert not dummy_proxy.received_connect, "REST path should ignore HTTPS_PROXY"
+
+def test_websocket_call_honors_env(dummy_proxy, monkeypatch):
+    # set HTTPS_PROXY again
+    monkeypatch.setenv("HTTPS_PROXY", "http://127.0.0.1:8888")
+    # Load kubeconfig
+    config.load_kube_config(config_file=os.environ["KUBECONFIG"])
+    apply_proxy_to_conf()
+    opts = websocket_proxycare({}, client.Configuration.get_default_copy(), None, None)
+    assert opts.get('http_proxy_host') == '127.0.0.1'
+    assert opts.get('http_proxy_port') == 8888
+    # Optionally verify no_proxy parsing
+    assert opts.get('http_no_proxy') is None
 
 if __name__ == '__main__':
     unittest.main()