codekingpro/portable-devtools

1# Copyright 2018 The Kubernetes Authors.2#3# Licensed under the Apache License, Version 2.0 (the "License");4# you may not use this file except in compliance with the License.5# You may obtain a copy of the License at6#7#     http://www.apache.org/licenses/LICENSE-2.08#9# Unless required by applicable law or agreed to in writing, software10# distributed under the License is distributed on an "AS IS" BASIS,11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12# See the License for the specific language governing permissions and13# limitations under the License.14 15import unittest16 17from .ws_client import get_websocket_url18from .ws_client import websocket_proxycare19from kubernetes.client.configuration import Configuration20import os21import socket22import threading23import pytest24from kubernetes import stream, client, config25 26try:27    import urllib328    urllib3.disable_warnings()29except ImportError:30    pass31@pytest.fixture(autouse=True)32def dummy_kubeconfig(tmp_path, monkeypatch):33    # Creating a kubeconfig34    content = """35        apiVersion: v136        kind: Config37        clusters:38        - name: default39          cluster:40            server: http://127.0.0.1:888841        contexts:42        - name: default43          context:44            cluster: default45            user: default46        users:47        - name: default48          user: {}49        current-context: default50        """51    cfg_file = tmp_path / "kubeconfig"52    cfg_file.write_text(content)53    monkeypatch.setenv("KUBECONFIG", str(cfg_file))54 55 56def dictval(dict_obj, key, default=None):57		58    return dict_obj.get(key, default)59 60class DummyProxy(threading.Thread):61    """62    A minimal HTTP proxy that flags any CONNECT request and returns 200 OK.63    Listens on 127.0.0.1:8888 by default.64    """65    def __init__(self, host='127.0.0.1', port=8888):66        super().__init__(daemon=True)67        self.host = host68        self.port = port69        self.received_connect = False70        self._server_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)71        self._server_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)72        self._server_sock.bind((self.host, 0))73        self._server_sock.listen(1)74 75    def run(self):76        conn, _ = self._server_sock.accept()77        try:78            data = conn.recv(1024).decode('utf-8', errors='ignore')79            if data.startswith('CONNECT '):80                self.received_connect = True81                conn.sendall(b"HTTP/1.1 200 Connection established\r\n\r\n")82        finally:83            conn.close()84 85class WSClientTest(unittest.TestCase):86 87    def test_websocket_client(self):88        for url, ws_url in [89                ('http://localhost/api', 'ws://localhost/api'),90                ('https://localhost/api', 'wss://localhost/api'),91                ('https://domain.com/api', 'wss://domain.com/api'),92                ('https://api.domain.com/api', 'wss://api.domain.com/api'),93                ('http://api.domain.com', 'ws://api.domain.com'),94                ('https://api.domain.com', 'wss://api.domain.com'),95                ('http://api.domain.com/', 'ws://api.domain.com/'),96                ('https://api.domain.com/', 'wss://api.domain.com/'),97                ]:98            self.assertEqual(get_websocket_url(url), ws_url)99 100    def test_websocket_proxycare(self):101        for proxy, idpass, no_proxy, expect_host, expect_port, expect_auth, expect_noproxy in [102                ( None,                             None,        None,                            None,                None, None, None ),103                ( 'http://proxy.example.com:8080/', None,        None,                            'proxy.example.com', 8080, None, None ),104                ( 'http://proxy.example.com:8080/', 'user:pass', None,                            'proxy.example.com', 8080, ('user','pass'), None),105                ( 'http://proxy.example.com:8080/', 'user:pass', '',                              'proxy.example.com', 8080, ('user','pass'), None),106                ( 'http://proxy.example.com:8080/', 'user:pass', '*',                             'proxy.example.com', 8080, ('user','pass'), ['*']),107                ( 'http://proxy.example.com:8080/', 'user:pass', '.example.com',                  'proxy.example.com', 8080, ('user','pass'), ['.example.com']),108                ( 'http://proxy.example.com:8080/', 'user:pass', 'localhost,.local,.example.com',  'proxy.example.com', 8080, ('user','pass'), ['localhost','.local','.example.com']),109                ]:110            #  input setup111            cfg = Configuration()112            if proxy:113                cfg.proxy = proxy114            if idpass:115                cfg.proxy_headers = urllib3.util.make_headers(proxy_basic_auth=idpass)116            if no_proxy is not None:117                cfg.no_proxy = no_proxy118						119						 120            connect_opts = websocket_proxycare({}, cfg, None, None)121            assert dictval(connect_opts, 'http_proxy_host') == expect_host122            assert dictval(connect_opts, 'http_proxy_port') == expect_port123            assert dictval(connect_opts, 'http_proxy_auth') == expect_auth124            assert dictval(connect_opts, 'http_no_proxy') == expect_noproxy125 126@pytest.fixture(scope="module")127def dummy_proxy():128    #Dummy Proxy129    proxy = DummyProxy(port=8888)130    proxy.start()131    yield proxy132 133@pytest.fixture(autouse=True)134def clear_proxy_env(monkeypatch):135    for var in ("HTTP_PROXY", "http_proxy", "HTTPS_PROXY", "https_proxy", "NO_PROXY", "no_proxy"):136        monkeypatch.delenv(var, raising=False)137 138def apply_proxy_to_conf():139    #apply HTTPS_PROXY env var and set it as global.   140    cfg = client.Configuration.get_default_copy()141    cfg.proxy = os.getenv("HTTPS_PROXY")142    cfg.no_proxy = os.getenv("NO_PROXY", "")143    client.Configuration.set_default(cfg)144 145def test_rest_call_ignores_env(dummy_proxy, monkeypatch):146    # HTTPS_PROXY to dummy proxy147    monkeypatch.setenv("HTTPS_PROXY", "http://127.0.0.1:8888")148    # Avoid real HTTP request149    monkeypatch.setattr(client.CoreV1Api, "list_namespace", lambda self, *_args, **_kwargs: None)150    # Load config using kubeconfig151    config.load_kube_config(config_file=os.environ["KUBECONFIG"])152    apply_proxy_to_conf()153    # HTTPS_PROXY to dummy proxy154    monkeypatch.setenv("HTTPS_PROXY", "http://127.0.0.1:8888")155    config.load_kube_config(config_file=os.environ["KUBECONFIG"])156    apply_proxy_to_conf()157    v1 = client.CoreV1Api()158    v1.list_namespace(_preload_content=False)159    assert not dummy_proxy.received_connect, "REST path should ignore HTTPS_PROXY"160 161def test_websocket_call_honors_env(dummy_proxy, monkeypatch):162    # set HTTPS_PROXY again163    monkeypatch.setenv("HTTPS_PROXY", "http://127.0.0.1:8888")164    # Load kubeconfig165    config.load_kube_config(config_file=os.environ["KUBECONFIG"])166    apply_proxy_to_conf()167    opts = websocket_proxycare({}, client.Configuration.get_default_copy(), None, None)168    assert opts.get('http_proxy_host') == '127.0.0.1'169    assert opts.get('http_proxy_port') == 8888170    # Optionally verify no_proxy parsing171    assert opts.get('http_no_proxy') is None172 173if __name__ == '__main__':174    unittest.main()175