Handle BOOKMARK events in watch.py

Verified with python -m unittest kubernetes/watch/watch_test.py.
Applied flake8 and isort on impacted lines.

Author: note35

kubernetes/base/watch/watch.py

@@ -114,11 +114,18 @@ def unmarshal_event(self, data, return_type):
         try:
             js = json.loads(data)
             js['raw_object'] = js['object']
-            # BOOKMARK event is treated the same as ERROR for a quick fix of
-            # decoding exception
-            # TODO: make use of the resource_version in BOOKMARK event for more
-            # efficient WATCH
-            if return_type and js['type'] != 'ERROR' and js['type'] != 'BOOKMARK':
+
+            if not return_type:
+                return js
+
+            if js['type'] == 'BOOKMARK':
+                # Extract and store resource_version from BOOKMARK event for
+                # efficiency. No deserialization as event can be incomplete.
+                if isinstance(js['object'], dict) and 'metadata' in js['object']:
+                    metadata = js['object']['metadata']
+                    if isinstance(metadata, dict) and 'resourceVersion' in metadata:
+                        self.resource_version = metadata['resourceVersion']
+            elif js['type'] != 'ERROR':
                 obj = SimpleNamespace(data=json.dumps(js['raw_object']))
                 js['object'] = self._api_client.deserialize(obj, return_type)
                 if hasattr(js['object'], 'metadata'):

kubernetes/base/watch/watch_test.py

@@ -12,20 +12,16 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import unittest
-
 import os
-
 import time
-
+import unittest
 from unittest.mock import Mock, call
 
-from kubernetes import client,config
+from kubernetes import client, config
+from kubernetes.client import ApiException
 
 from .watch import Watch
 
-from kubernetes.client import ApiException
-
 
 class WatchTests(unittest.TestCase):
     def setUp(self):
@@ -392,9 +388,22 @@ def test_unmarshal_with_bookmark(self):
             '"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,
-        # resource_version in BOOKMARK is *not* used at all.
+        self.assertEqual("1", w.resource_version)
+
+    def test_unmarshal_with_bookmark_metadata_not_in_dict(self):
+        w = Watch()
+        event = w.unmarshal_event(
+            '{"type":"BOOKMARK","object":{"metadata": "not-a-dict"}}',
+            'V1Job')
+        self.assertEqual("BOOKMARK", event['type'])
+        self.assertEqual(None, w.resource_version)
+
+    def test_unmarshal_with_bookmark_metadata_without_resource_version(self):
+        w = Watch()
+        event = w.unmarshal_event(
+            '{"type":"BOOKMARK","object":{"metadata": {"name": "foo"}}}',
+            'V1Job')
+        self.assertEqual("BOOKMARK", event['type'])
         self.assertEqual(None, w.resource_version)
 
     def test_watch_with_exception(self):