Address 6 code review comments: _fire docstring, resync fix, test assertions, e2e data checks

Co-authored-by: brendandburns <5751682+brendandburns@users.noreply.github.com>

Author: Copilot

kubernetes/e2e_test/test_informer.py

@@ -83,7 +83,11 @@ def test_cache_populated_after_start(self):
         self.addCleanup(inf.stop)
 
         self._wait_in_cache(inf, "default/" + name)
-        self.assertEqual(_name_of(inf.cache.get_by_key("default/" + name)), name)
+        cached = inf.cache.get_by_key("default/" + name)
+        self.assertEqual(_name_of(cached), name)
+        # Verify the cached object actually contains the expected data payload.
+        data = cached.data if hasattr(cached, "data") else (cached.get("data") or {})
+        self.assertEqual(data.get("k"), "v")
 
     def test_added_event_and_cache_entry(self):
         """Creating a ConfigMap fires ADDED and the object appears in the cache."""
@@ -127,7 +131,11 @@ def test_modified_event_and_cache_refresh(self):
             name=name, namespace="default", body={"data": {"k": "updated"}}
         )
         self._expect(seen, "MODIFIED/" + name)
-        self.assertIsNotNone(inf.cache.get_by_key("default/" + name))
+        # Verify that the cache now holds the updated data.
+        cached = inf.cache.get_by_key("default/" + name)
+        self.assertIsNotNone(cached)
+        data = cached.data if hasattr(cached, "data") else (cached.get("data") or {})
+        self.assertEqual(data.get("k"), "updated")
 
     def test_deleted_event_removes_from_cache(self):
         """Deleting a ConfigMap fires DELETED and removes it from the cache."""

kubernetes/informer/informer.py

@@ -173,6 +173,12 @@ def _build_kwargs(self):
         return kw
 
     def _fire(self, event_type, obj):
+        """Execute all registered callbacks for *event_type*, passing *obj*.
+
+        Callbacks are invoked sequentially on the informer's background thread.
+        Any exception raised by an individual handler is logged and swallowed so
+        that remaining handlers still run.
+        """
         with self._handler_lock:
             handlers = list(self._handlers.get(event_type, []))
         for fn in handlers:
@@ -254,6 +260,13 @@ def _run_loop(self):
             self._watch = Watch()
             kw = self._build_kwargs()
             kw["resource_version"] = self._resource_version
+            # When a resync period is configured, set a matching server-side
+            # watch timeout so that the stream exits after resync_period seconds
+            # even if no events arrive.  Without this, a quiet period longer
+            # than resync_period would never trigger a resync because the check
+            # below only runs when the generator yields an event.
+            if self._resync_period > 0:
+                kw["timeout_seconds"] = max(1, int(self._resync_period))
             try:
                 for event in self._watch.stream(self._list_func, **kw):
                     if self._stop_event.is_set():
@@ -282,19 +295,6 @@ def _run_loop(self):
                         self._fire(BOOKMARK, event.get("raw_object", obj))
                     elif evt_type == ERROR:
                         self._fire(ERROR, obj)
-                    # Periodic resync: full re-list from the API server, firing
-                    # ADDED/MODIFIED/DELETED for any changes since the last list.
-                    if (
-                        self._resync_period > 0
-                        and (time.monotonic() - last_resync) >= self._resync_period
-                    ):
-                        logger.debug("Informer resync triggered")
-                        try:
-                            self._initial_list()
-                        except Exception as exc:
-                            logger.exception("Error during resync list; continuing")
-                            self._fire(ERROR, exc)
-                        last_resync = time.monotonic()
             except ApiException as exc:
                 if exc.status == 410:
                     # The stored resource version is too old; force a full re-list.
@@ -323,3 +323,20 @@ def _run_loop(self):
                 ):
                     self._resource_version = self._watch.resource_version
                 self._watch = None
+
+            # Periodic resync: after the watch stream exits (whether due to the
+            # server-side timeout_seconds, a stop request, or an error) check if
+            # a resync is due.  This path is what actually fires the resync when
+            # the cluster is quiet and no events arrive for resync_period seconds.
+            if (
+                not self._stop_event.is_set()
+                and self._resource_version is not None  # 410 already schedules a re-list
+                and self._resync_period > 0
+                and (time.monotonic() - last_resync) >= self._resync_period
+            ):
+                logger.debug("Informer resync triggered")
+                try:
+                    self._initial_list()
+                except Exception as exc:
+                    logger.exception("Error during resync list; continuing")
+                    self._fire(ERROR, exc)

kubernetes/test/test_informer.py

@@ -114,6 +114,14 @@ def worker(n):
         for t in threads:
             t.join()
         self.assertEqual(errors, [])
+        # Verify that the cache actually holds the objects that were put into it.
+        for n in range(5):
+            for i in range(50):
+                key = "default/pod-{}-{}".format(n, i)
+                self.assertIsNotNone(
+                    self.cache.get_by_key(key),
+                    "expected key {} in cache".format(key),
+                )
 
 
 class TestSharedInformerHandlers(unittest.TestCase):
@@ -360,7 +368,13 @@ def fake_stream(func, **kw):
         self.assertIs(cached[0], pod)
 
     def test_resync_period_triggers_full_list(self):
-        """A full List call must be made to the API server on every resync_period."""
+        """A full List call must be made to the API server on every resync_period.
+
+        With the new implementation the watch stream is given a server-side
+        timeout equal to resync_period (via timeout_seconds).  When the stream
+        exits, the elapsed-time check fires the resync even if no events
+        arrived – this is exactly the scenario this test exercises.
+        """
         pod = _make_pod("default", "resync-pod")
 
         list_func = MagicMock()
@@ -371,20 +385,31 @@ def test_resync_period_triggers_full_list(self):
 
         informer = SharedInformer(list_func=list_func, resync_period=60)
 
+        stream_calls = {"n": 0}
+
         with patch("kubernetes.informer.informer.Watch") as MockWatch, \
                 patch("kubernetes.informer.informer.time") as mock_time:
-            # Sequence of time.monotonic() calls:
-            #   1. last_resync = time.monotonic()          → 0.0
-            #   2. (time.monotonic() - last_resync) check  → 61.0  (triggers resync)
-            #   3. last_resync = time.monotonic()          → 61.0  (reset after resync)
+            # Sequence of time.monotonic() calls inside _run_loop:
+            #   1. last_resync = time.monotonic()            → 0.0  (watch-loop start)
+            #   2. post-stream: time.monotonic()             → 61.0 (≥60 → resync fires)
+            #   3. last_resync = time.monotonic()            → 61.0 (second watch-loop start)
+            # The stop_event is set during the second stream, so the
+            # post-stream check is short-circuited and no further calls occur.
             mock_time.monotonic.side_effect = [0.0, 61.0, 61.0]
+            mock_time.sleep = time.sleep  # keep real sleep/wait working
 
             mock_w = MagicMock()
             mock_w.resource_version = "5"
 
             def fake_stream(func, **kw):
-                yield {"type": "ADDED", "object": pod}
+                stream_calls["n"] += 1
+                if stream_calls["n"] == 1:
+                    # Simulate the stream timing out (timeout_seconds expired)
+                    # with no events – the resync should fire after this returns.
+                    return iter([])
+                # Second iteration: stop the informer.
                 informer._stop_event.set()
+                return iter([])
 
             mock_w.stream.side_effect = fake_stream
             MockWatch.return_value = mock_w
@@ -955,8 +980,12 @@ def list_func(**kw):
             return resp
 
         modified = []
+        added = []
+        deleted = []
         informer = SharedInformer(list_func=list_func)
         informer.add_event_handler(MODIFIED, modified.append)
+        informer.add_event_handler(ADDED, added.append)
+        informer.add_event_handler(DELETED, deleted.append)
 
         stream_calls = {"n": 0}
 
@@ -980,6 +1009,11 @@ def fake_stream(func, **kw):
         # pod was in both the initial list (call 1) and the re-list (call 2).
         # On the re-list it should fire MODIFIED (not ADDED again).
         self.assertIn(pod, modified)
+        # ADDED fires exactly once: for the initial list.  The re-list must
+        # NOT fire a second ADDED for an already-cached item.
+        self.assertEqual(len(added), 1, "ADDED should fire once (initial list) but not again on re-list")
+        # DELETED must not fire for an item present in both lists.
+        self.assertEqual(deleted, [], "DELETED should not fire for an item present in both lists")
         # Still in cache.
         self.assertIsNotNone(informer.cache.get_by_key("default/stable-pod"))