feat: add switch control when watching events whether deserialization…

kubernetes/base/watch/watch.py

@@ -172,14 +172,19 @@ def stream(self, func, *args, **kwargs):
         # We want to ensure we are returning within that timeout.
         disable_retries = ('timeout_seconds' in kwargs)
         retry_after_410 = False
+        deserialize = kwargs.pop('deserialize', False)
         while True:
             resp = func(*args, **kwargs)
             try:
                 for line in iter_resp_lines(resp):
                     # unmarshal when we are receiving events from watch,
                     # return raw string when we are streaming log
                     if watch_arg == "watch":
-                        event = self.unmarshal_event(line, return_type)
+                        if deserialize:
+                            event = self.unmarshal_event(line, return_type)
+                        else:
+                            # Only do basic JSON parsing, no deserialize
+                            event = json.loads(line)
                         if isinstance(event, dict) \
                                 and event['type'] == 'ERROR':
                             obj = event['raw_object']

kubernetes/base/watch/watch_test.py

@@ -489,6 +489,44 @@ def test_watch_with_error_event_and_timeout_param(self):
         fake_resp.close.assert_called_once()
         fake_resp.release_conn.assert_called_once()
 
+    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)
+        ])
 
 if __name__ == '__main__':
     unittest.main()