Your advanced code editor is loading.

Please wait a moment.

#editor-loading { display: none; }

Licensed under the Apache License, Version 2.0 (the “License”);

you may not use this file except in compliance with the License.

You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software

distributed under the License is distributed on an “AS IS” BASIS,

WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

See the License for the specific language governing permissions and

limitations under the License.

import unittest

from .ws_client import get_websocket_url from .ws_client import websocket_proxycare from kubernetes.client.configuration import Configuration

try: import urllib3 urllib3.disable_warnings() except ImportError: pass

def dictval(dict, key, default=None): try: val = dict[key] except KeyError: val = default return val

class WSClientTest(unittest.TestCase):

def test_websocket_client(self):
          for url, ws_url in [
          ('http://localhost/api', 'ws://localhost/api'),
          ('https://localhost/api', 'wss://localhost/api'),
          ('https://domain.com/api', 'wss://domain.com/api'),
          ('https://api.domain.com/api', 'wss://api.domain.com/api'),
          ('http://api.domain.com', 'ws://api.domain.com'),
          ('https://api.domain.com', 'wss://api.domain.com'),
          ('http://api.domain.com/', 'ws://api.domain.com/'),
          ('https://api.domain.com/', 'wss://api.domain.com/'),
          ]:
          self.assertEqual(get_websocket_url(url), ws_url)
          
          def test_websocket_proxycare(self):
          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()

Explanation

This code defines a test suite for a websocket client implementation within the kubernetes-client python library.

test_websocket_client

This tests the get_websocket_url function which converts a given HTTP URL to its websocket equivalent. It iterates over different URL formats (with and without protocol, with and without trailing slashes) and asserts that the function correctly produces the expected websocket URL.

test_websocket_proxycare

This test checks how the websocket_proxycare function handles proxy configurations. It iterates over various proxy settings, including different proxy URLs, authentication credentials, and no_proxy settings. For each scenario, it sets up a Configuration object with the relevant proxy parameters and calls websocket_proxycare to get the connection options. It then asserts that the returned options (connect_opt) contain the expected proxy host, port, authentication credentials, and no_proxy list.

Key points:

get_websocket_url simply replaces “http://” with “ws://” or “https://” with “wss://”, which is how websocket URLs are constructed.
websocket_proxycare uses the provided Configuration object to determine the proxy settings and then creates a dictionary (connect_opt) containing the appropriate connection parameters for the websocket client.
The dictval helper function safely retrieves a value from a dictionary, returning a default value if the key is not found.

The tests cover various aspects of the websocket client’s handling of URLs and proxy configurations, ensuring that it behaves correctly in different scenarios.

Your advanced code editor is loading.

Please wait a moment.

Explanation

Graph

Symbols

We couldn't identify any entrypoints. If you believe this to be incorrect then please contact support.