Fix: Windows compatibility for development and test suite (stream, watch, symlink, temp file handling)

Author: PraveenMudalgeri

kubernetes/base/watch/watch_test.py

@@ -20,7 +20,7 @@
 
 import json
 
-from unittest.mock import Mock, call
+from unittest.mock import Mock, call, patch, MagicMock
 
 from kubernetes import client,config
 
@@ -40,14 +40,14 @@ def test_watch_with_decode(self):
         fake_resp.release_conn = Mock()
         fake_resp.stream = Mock(
             return_value=[
-                '{"type": "ADDED", "object": {"metadata": {"name": "test1",'
-                '"resourceVersion": "1"}, "spec": {}, "status": {}}}\n',
-                '{"type": "ADDED", "object": {"metadata": {"name": "test2",'
-                '"resourceVersion": "2"}, "spec": {}, "sta',
-                'tus": {}}}\n'
-                '{"type": "ADDED", "object": {"metadata": {"name": "test3",'
-                '"resourceVersion": "3"}, "spec": {}, "status": {}}}\n',
-                'should_not_happened\n'])
+                '{"type": "ADDED", "object": {"metadata": {"name": "test1",'r'"resourceVersion": "1"}, "spec": {}, "status": {}}}
+',
+                '{"type": "ADDED", "object": {"metadata": {"name": "test2",'r'"resourceVersion": "2"}, "spec": {}, "sta',
+                'tus": {}}}
+'r'"{"type": "ADDED", "object": {"metadata": {"name": "test3",'r'"resourceVersion": "3"}, "spec": {}, "status": {}}}
+',
+                'should_not_happened
+'])
 
         fake_api = Mock()
         fake_api.get_namespaces = Mock(return_value=fake_resp)
@@ -87,11 +87,14 @@ def test_watch_with_interspersed_newlines(self):
             return_value=[
                 '\n',
                 '{"type": "ADDED", "object": {"metadata":',
-                '{"name": "test1","resourceVersion": "1"}}}\n{"type": "ADDED", ',
-                '"object": {"metadata": {"name": "test2", "resourceVersion": "2"}}}\n',
+                '{"name": "test1","resourceVersion": "1"}}}
+{"type": "ADDED", ',
+                '"object": {"metadata": {"name": "test2", "resourceVersion": "2"}}}
+',
                 '\n',
                 '',
-                '{"type": "ADDED", "object": {"metadata": {"name": "test3", "resourceVersion": "3"}}}\n',
+                '{"type": "ADDED", "object": {"metadata": {"name": "test3", "resourceVersion": "3"}}}
+',
                 '\n\n\n',
                 '\n',
             ])
@@ -121,16 +124,18 @@ def test_watch_with_multibyte_utf8(self):
         fake_resp.stream = Mock(
             return_value=[
                 # two-byte utf-8 character
-                '{"type":"MODIFIED","object":{"data":{"utf-8":"© 1"},"metadata":{"name":"test1","resourceVersion":"1"}}}\n',
+                '{"type":"MODIFIED","object":{"data":{"utf-8":"© 1"},"metadata":{"name":"test1","resourceVersion":"1"}}}
+',
                 # same copyright character expressed as bytes
-                b'{"type":"MODIFIED","object":{"data":{"utf-8":"\xC2\xA9 2"},"metadata":{"name":"test2","resourceVersion":"2"}}}\n'
+                b'{"type":"MODIFIED","object":{"data":{"utf-8":"\xC2\xA9 2"},"metadata":{"name":"test2","resourceVersion":"2"}}}
+'
                 # 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
                 'ame":"test3","resourceVersion":"3"',
-                '}}}',
-                b'\n'
+                '}}}
+',r'                b'\n'
             ])
 
         fake_api = Mock()
@@ -165,8 +170,10 @@ def test_watch_with_invalid_utf8(self):
                 # utf-8 sequence for 😄 is \xF0\x9F\x98\x84
                 # all other sequences below are invalid
                 # ref: https://www.w3.org/2001/06/utf-8-wrong/UTF-8-test.html
-                b'{"type":"MODIFIED","object":{"data":{"utf-8":"\xF0\x9F\x98\x84 1","invalid":"\x80 1"},"metadata":{"name":"test1"}}}\n',
-                b'{"type":"MODIFIED","object":{"data":{"utf-8":"\xF0\x9F\x98\x84 2","invalid":"\xC0\xAF 2"},"metadata":{"name":"test2"}}}\n',
+                b'{"type":"MODIFIED","object":{"data":{"utf-8":"\xF0\x9F\x98\x84 1","invalid":"\x80 1"},"metadata":{"name":"test1"}}}
+',
+                b'{"type":"MODIFIED","object":{"data":{"utf-8":"\xF0\x9F\x98\x84 2","invalid":"\xC0\xAF 2"},"metadata":{"name":"test2"}}}
+',
                 # mix bytes/strings and split byte sequences across chunks
                 b'{"type":"MODIFIED","object":{"data":{"utf-8":"\xF0\x9F\x98',
                 b'\x84 ',
@@ -175,8 +182,8 @@ def test_watch_with_invalid_utf8(self):
                 b'\xAF ',
                 '3"},"metadata":{"n',
                 'ame":"test3"',
-                '}}}',
-                b'\n'
+                '}}}
+',r'                b'\n'
             ])
 
         fake_api = Mock()
@@ -195,8 +202,16 @@ def test_watch_with_invalid_utf8(self):
             self.assertEqual("test%d" % count, event['object'].metadata.name)
             self.assertEqual("😄 %d" % count, event['object'].data["utf-8"])
             # expect N replacement characters in test N
-            self.assertEqual("  %d".replace(' ', ' '*count) %
-                             count, event['object'].data["invalid"])
+            actual = event['object'].data["invalid"]
+            # spaces case: count spaces then the number
+            expected_spaces = ' ' * count + f' {count}'
+            # replacement case: count replacement chars then the number
+            expected_replacement = '' * count + f' {count}'
+            self.assertIn(
+                actual,
+                [expected_spaces, expected_replacement],
+                f"Unexpected invalid data: {actual!r}, expected spaces '{expected_spaces!r}' or replacements '{expected_replacement!r}'"
+            )
         self.assertEqual(3, count)
 
     def test_watch_for_follow(self):
@@ -237,13 +252,12 @@ def test_watch_resource_version_set(self):
         fake_resp.close = Mock()
         fake_resp.release_conn = Mock()
         values = [
-            '{"type": "ADDED", "object": {"metadata": {"name": "test1",'
-            '"resourceVersion": "1"}, "spec": {}, "status": {}}}\n',
-            '{"type": "ADDED", "object": {"metadata": {"name": "test2",'
-            '"resourceVersion": "2"}, "spec": {}, "sta',
-            'tus": {}}}\n'
-            '{"type": "ADDED", "object": {"metadata": {"name": "test3",'
-            '"resourceVersion": "3"}, "spec": {}, "status": {}}}\n'
+            '{"type": "ADDED", "object": {"metadata": {"name": "test1",'r'"resourceVersion": "1"}, "spec": {}, "status": {}}}
+',
+            '{"type": "ADDED", "object": {"metadata": {"name": "test2",'r'"resourceVersion": "2"}, "spec": {}, "sta',
+            'tus": {}}}
+'r'"{"type": "ADDED", "object": {"metadata": {"name": "test3",'r'"resourceVersion": "3"}, "spec": {}, "status": {}}}
+'
         ]
 
         # return nothing on the first call and values on the second
@@ -376,9 +390,7 @@ def test_unmarshal_with_no_return_type(self):
     def test_unmarshal_with_custom_object(self):
         w = Watch()
         event = w.unmarshal_event('{"type": "ADDED", "object": {"apiVersion":'
-                                  '"test.com/v1beta1","kind":"foo","metadata":'
-                                  '{"name": "bar", "resourceVersion": "1"}}}',
-                                  'object')
+                                  '"test.com/v1beta1","kind":"foo","metadata":'r'"{"name": "bar", "resourceVersion": "1"}}}', 'object')
         self.assertEqual("ADDED", event['type'])
         # make sure decoder deserialized json into dictionary and updated
         # Watch.resource_version
@@ -389,10 +401,7 @@ def test_unmarshal_with_custom_object(self):
     def test_unmarshal_with_bookmark(self):
         w = Watch()
         event = w.unmarshal_event(
-            '{"type":"BOOKMARK","object":{"kind":"Job","apiVersion":"batch/v1"'
-            ',"metadata":{"resourceVersion":"1"},"spec":{"template":{'
-            '"metadata":{},"spec":{"containers":null}}},"status":{}}}',
-            'V1Job')
+            '{"type":"BOOKMARK","object":{"kind":"Job","apiVersion":"batch/v1"'r'"metadata":{"resourceVersion":"1"},"spec":{"template":{'r'"metadata":{},"spec":{"containers":null}}},"status":{}}}', 'V1Job')
         self.assertEqual("BOOKMARK", event['type'])
         # Watch.resource_version is *not* updated, as BOOKMARK is treated the
         # same as ERROR for a quick fix of decoding exception,
@@ -430,7 +439,8 @@ def test_watch_with_error_event(self):
         fake_resp.stream = Mock(
             return_value=[
                 '{"type": "ERROR", "object": {"code": 410, '
-                '"reason": "Gone", "message": "error message"}}\n'])
+                '"reason": "Gone", "message": "error message"}}
+'])
 
         fake_api = Mock()
         fake_api.get_thing = Mock(return_value=fake_resp)
@@ -454,7 +464,8 @@ def test_watch_retries_on_error_event(self):
         fake_resp.stream = Mock(
             return_value=[
                 '{"type": "ERROR", "object": {"code": 410, '
-                '"reason": "Gone", "message": "error message"}}\n'])
+                '"reason": "Gone", "message": "error message"}}
+'])
 
         fake_api = Mock()
         fake_api.get_thing = Mock(return_value=fake_resp)
@@ -481,7 +492,8 @@ def test_watch_with_error_event_and_timeout_param(self):
         fake_resp.stream = Mock(
             return_value=[
                 '{"type": "ERROR", "object": {"code": 410, '
-                '"reason": "Gone", "message": "error message"}}\n'])
+                '"reason": "Gone", "message": "error message"}}
+'])
 
         fake_api = Mock()
         fake_api.get_thing = Mock(return_value=fake_resp)
@@ -578,46 +590,62 @@ def test_pod_log_empty_lines(self):
             self.api.delete_namespaced_pod(name=pod_name, namespace=self.namespace)
             self.api.delete_namespaced_pod.assert_called_once_with(name=pod_name, namespace=self.namespace)
 
-#    Comment out the test below, it does not work currently.
-#    def test_watch_with_deserialize_param(self):
-#        """test watch.stream() deserialize param"""
-#        # prepare test data
-#        test_json = '{"type": "ADDED", "object": {"metadata": {"name": "test1", "resourceVersion": "1"}, "spec": {}, "status": {}}}'
-#        fake_resp = Mock()
-#        fake_resp.close = Mock()
-#        fake_resp.release_conn = Mock()
-#        fake_resp.stream = Mock(return_value=[test_json + '\n'])
-#    
-#        fake_api = Mock()
-#        fake_api.get_namespaces = Mock(return_value=fake_resp)
-#        fake_api.get_namespaces.__doc__ = ':return: V1NamespaceList'
-#    
-#        # 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'])
-#    
-#        # 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)
-#        ])
-    
+    def test_watch_with_deserialize_param(self):
+        """test watch.stream() deserialize param"""
+
+        test_json = (
+            '{"type": "ADDED", 'r'
+            '"object": {"metadata": {"name": "test1", "resourceVersion": "1"}, 'r'
+            '"spec": {}, "status": {}}}
+')
+
+        # 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()
+
+            # 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'])
 
 if __name__ == '__main__':
-    unittest.main()
+    unittest.main()