Handle dict list responses in SharedInformer initial list

kubernetes/informer/informer.py

@@ -39,6 +39,20 @@
 ERROR = "ERROR"
 
 
+def _list_items_and_resource_version(resp):
+    """Return the items and resourceVersion of a list response.
+
+    Supports both generated model objects and the plain dicts returned by
+    CustomObjectsApi list functions.
+    """
+    if isinstance(resp, dict):
+        meta = resp.get("metadata") or {}
+        return resp.get("items") or [], meta.get("resourceVersion")
+    items = getattr(resp, "items", []) or []
+    meta = getattr(resp, "metadata", None)
+    return items, getattr(meta, "resource_version", None)
+
+
 class SharedInformer:
     """Watch a Kubernetes resource and maintain a local cache.
 
@@ -201,7 +215,7 @@ def _initial_list(self):
         """
         kw = self._build_kwargs()
         resp = self._list_func(**kw)
-        items = getattr(resp, "items", []) or []
+        items, rv = _list_items_and_resource_version(resp)
 
         # Build key → item map for incoming items.
         new_items_map = {}
@@ -229,10 +243,6 @@ def _initial_list(self):
             else:
                 self._fire(ADDED, item)
 
-        rv = None
-        meta = getattr(resp, "metadata", None)
-        if meta is not None:
-            rv = getattr(meta, "resource_version", None)
         self._resource_version = rv or "0"
 
     def _run_loop(self):

kubernetes/test/test_informer.py

@@ -736,6 +736,48 @@ def fake_stream(func, **kw):
         self.assertIn(pod2, received)
         self.assertEqual(len(received), 2)
 
+    def test_initial_list_handles_dict_response(self):
+        """A dict list response (e.g. CustomObjectsApi) populates the cache and
+        the watch resumes from the list's metadata.resourceVersion."""
+        obj1 = _make_pod("default", "obj1")
+        obj2 = _make_pod("default", "obj2")
+        list_func = MagicMock(return_value={
+            "apiVersion": "example.com/v1",
+            "kind": "FooList",
+            "metadata": {"resourceVersion": "42"},
+            "items": [obj1, obj2],
+        })
+
+        informer = SharedInformer(list_func=list_func, namespace="default")
+        received = {ADDED: [], ERROR: []}
+        for event_type, events in received.items():
+            informer.add_event_handler(event_type, events.append)
+
+        with patch("kubernetes.informer.informer.Watch") as MockWatch:
+            mock_w = MagicMock()
+            mock_w.resource_version = None
+
+            def fake_stream(func, **kw):
+                informer._stop_event.set()
+                return iter([])
+
+            mock_w.stream.side_effect = fake_stream
+            MockWatch.return_value = mock_w
+
+            informer.start()
+            informer._thread.join(timeout=3)
+
+        self.assertEqual(
+            sorted(informer.cache.list_keys()),
+            ["default/obj1", "default/obj2"],
+        )
+        self.assertEqual(received[ADDED], [obj1, obj2])
+        self.assertEqual(received[ERROR], [])
+        self.assertEqual(list_func.call_count, 1)
+        mock_w.stream.assert_called_once()
+        _, stream_kw = mock_w.stream.call_args
+        self.assertEqual(stream_kw.get("resource_version"), "42")
+
     def test_relist_after_410_fires_delete_for_removed_items(self):
         """After a 410-triggered re-list, items absent from the new list fire DELETED."""
         from kubernetes.client.exceptions import ApiException