Windows compatibility fixes - additional changes

Author: PraveenMudalgeri

.github/workflows/test.yaml

@@ -48,3 +48,27 @@ jobs:
       with:
         fail_ci_if_error: false
         verbose: true
+
+  build_windows:
+    runs-on: windows-latest
+    strategy:
+      matrix:
+        python-version: ["3.8", "3.9", "3.10", "3.11", "3.12"]
+
+    steps:
+    - uses: actions/checkout@v4
+      with:
+        submodules: true
+    - name: Set up Python ${{ matrix.python-version }}
+      uses: actions/setup-python@v5
+      with:
+        python-version: ${{ matrix.python-version }}
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        pip install -r requirements.txt
+    - name: Install Tox and any other packages
+      run: pip install tox
+
+    - name: Test
+      run: tox -e py  # Run tox using the version of Python in `PATH`

kubernetes/__init__.py

@@ -22,4 +22,7 @@
 from . import watch
 from . import stream
 from . import utils
-from . import leaderelection
+import sys
+
+if sys.platform != 'win32':
+    from . import leaderelection

kubernetes/base/config/kube_config.py

@@ -423,21 +423,21 @@ def _refresh_oidc(self, provider):
         config = Configuration()
 
         if 'idp-certificate-authority-data' in provider['config']:
-            ca_cert = tempfile.NamedTemporaryFile(delete=True)
-
-            if PY3:
+            ca_cert = tempfile.NamedTemporaryFile(delete=False, mode='w', encoding='utf-8')
+            try:
                 cert = base64.b64decode(
                     provider['config']['idp-certificate-authority-data']
                 ).decode('utf-8')
-            else:
-                cert = base64.b64decode(
-                    provider['config']['idp-certificate-authority-data'] + "=="
-                )
-
-            with open(ca_cert.name, 'w') as fh:
-                fh.write(cert)
-
-            config.ssl_ca_cert = ca_cert.name
+                ca_cert.write(cert)
+                ca_cert.close()
+                config.ssl_ca_cert = ca_cert.name
+                if len(_temp_files) == 0:
+                    atexit.register(_cleanup_temp_files)
+                _temp_files[ca_cert.name] = ca_cert.name
+            except Exception:
+                ca_cert.close()
+                os.unlink(ca_cert.name)
+                raise
 
         elif 'idp-certificate-authority' in provider['config']:
             config.ssl_ca_cert = provider['config']['idp-certificate-authority']

kubernetes/base/watch/watch_test.py

@@ -20,7 +20,7 @@
 
 import json
 
-from unittest.mock import Mock, call, patch, MagicMock
+from unittest.mock import Mock, call
 
 from kubernetes import client,config
 
@@ -131,8 +131,8 @@ def test_watch_with_multibyte_utf8(self):
 '
                 # same copyright character with bytes split across two stream chunks
                 b'{"type":"MODIFIED","object":{"data":{"utf-8":"\xC2',
-                b'\xA9 3"},"metadata":{"n',
-                # more chunks of the same event, sent as a mix of bytes and strings
+                b'\xA9 3"},"metadata":{
+',                # more chunks of the same event, sent as a mix of bytes and strings
                 'ame":"test3","resourceVersion":"3"',
                 '}}}
 ',r'                b'\n'
@@ -180,8 +180,8 @@ def test_watch_with_invalid_utf8(self):
                 b'',
                 b'3","invalid":"\xE0\x80',
                 b'\xAF ',
-                '3"},"metadata":{"n',
-                'ame":"test3"',
+                '3"},"metadata":{
+',                'ame":"test3"',
                 '}}}
 ',r'                b'\n'
             ])
@@ -592,60 +592,47 @@ def test_pod_log_empty_lines(self):
 
     def test_watch_with_deserialize_param(self):
         """test watch.stream() deserialize param"""
-
+        # prepare test data
         test_json = (
             '{"type": "ADDED", 'r'
             '"object": {"metadata": {"name": "test1", "resourceVersion": "1"}, 'r'
             '"spec": {}, "status": {}}}
 ')
+        fake_resp = Mock()
+        fake_resp.close = Mock()
+        fake_resp.release_conn = Mock()
+        fake_resp.stream = Mock(return_value=[test_json + '\n'])
 
-        # Mock object for deserialize=True case
-        metadata_mock = MagicMock()
-        metadata_mock.name = 'test1'
-        metadata_mock.resource_version = '1'
-
-        object_mock = MagicMock()
-        object_mock.metadata = metadata_mock
-
-        event_deserialized = {
-            'type': 'ADDED',
-            'object': object_mock,
-            'raw_object': json.loads(test_json)['object']
-        }
-
-        # Event for deserialize=False case - object is plain dict
-        event_raw = {
-            'type': 'ADDED',
-            'object': json.loads(test_json)['object'],
-            'raw_object': json.loads(test_json)['object']
-        }
-
-        # Patch Watch.stream to return event_deserialized for deserialize=True
-        # and event_raw for deserialize=False - handle both calls with side_effect
-        def stream_side_effect(func, deserialize):
-            if deserialize:
-                return [event_deserialized]
-            else:
-                return [event_raw]
-
-        with patch.object(Watch, 'stream', side_effect=stream_side_effect):
-
-            w = Watch()
+        fake_api = Mock()
+        fake_api.get_namespaces = Mock(return_value=fake_resp)
+        fake_api.get_namespaces.__doc__ = ':return: V1NamespaceList'
 
-            # test case with deserialize=True
-            for e in w.stream(lambda: None, deserialize=True):  # dummy API func
-                self.assertEqual("ADDED", e['type'])
-                self.assertTrue(hasattr(e['object'], 'metadata'))
-                self.assertEqual("test1", e['object'].metadata.name)
-                self.assertEqual("1", e['object'].metadata.resource_version)
-                self.assertEqual(event_deserialized['raw_object'], e['raw_object'])
-
-            # test case with deserialize=False
-            for e in w.stream(lambda: None, deserialize=False):
-                self.assertEqual("ADDED", e['type'])
-                self.assertIsInstance(e['object'], dict)
-                self.assertEqual("test1", e['object']['metadata']['name'])
-                self.assertEqual("1", e['object']['metadata']['resourceVersion'])
+        # test case with deserialize=True
+        w = Watch()
+        for e in w.stream(fake_api.get_namespaces, deserialize=True):
+            self.assertEqual("ADDED", e['type'])
+        # Verify that the object is deserialized correctly
+        self.assertTrue(hasattr(e['object'], 'metadata'))
+        self.assertEqual("test1", e['object'].metadata.name)
+        self.assertEqual("1", e['object'].metadata.resource_version)
+        # Verify that the original object is saved
+        self.assertEqual(json.loads(test_json)['object'], e['raw_object'])
+
+        # test case with deserialize=False
+        w = Watch()
+        for e in w.stream(fake_api.get_namespaces, deserialize=False):
+            self.assertEqual("ADDED", e['type'])
+            # The validation object remains in the original dictionary format
+            self.assertIsInstance(e['object'], dict)
+            self.assertEqual("test1", e['object']['metadata']['name'])
+            self.assertEqual("1", e['object']['metadata']['resourceVersion'])
+            w.stop()  # ensure the loop terminates
+
+        # verify the api is called twice
+        fake_api.get_namespaces.assert_has_calls([
+            call(_preload_content=False, watch=True),
+            call(_preload_content=False, watch=True)
+        ])
 
 if __name__ == '__main__':
-    unittest.main()
+    unittest.main()

kubernetes/config/kube_config.py

@@ -423,27 +423,21 @@ def _refresh_oidc(self, provider):
         config = Configuration()
 
         if 'idp-certificate-authority-data' in provider['config']:
-            ca_cert = tempfile.NamedTemporaryFile(delete=True)
-
-            if PY3:
+            ca_cert = tempfile.NamedTemporaryFile(delete=False, mode='w', encoding='utf-8')
+            try:
                 cert = base64.b64decode(
                     provider['config']['idp-certificate-authority-data']
                 ).decode('utf-8')
-            else:
-                cert = base64.b64decode(
-                    provider['config']['idp-certificate-authority-data'] + "=="
-                )
-
-            with open(ca_cert.name, 'w') as fh:
-                fh.write(cert)
-
-            config.ssl_ca_cert = ca_cert.name
-
-        elif 'idp-certificate-authority' in provider['config']:
-            config.ssl_ca_cert = provider['config']['idp-certificate-authority']
-
-        else:
-            config.verify_ssl = False
+                ca_cert.write(cert)
+                ca_cert.close()
+                config.ssl_ca_cert = ca_cert.name
+                if len(_temp_files) == 0:
+                    atexit.register(_cleanup_temp_files)
+                _temp_files[ca_cert.name] = ca_cert.name
+            except Exception:
+                ca_cert.close()
+                os.unlink(ca_cert.name)
+                raise
 
         client = ApiClient(configuration=config)