add test

frigate/test/test_shared_memory_frame_manager.py

@@ -23,28 +23,47 @@ def _fake_shm(size: int) -> SimpleNamespace:
 class TestSharedMemoryFrameManagerGet(unittest.TestCase):
     def test_get_reopens_when_cached_segment_is_smaller_than_shape(self) -> None:
         """A cached ref to an older smaller segment must be dropped and the
-        current (larger) segment reopened. Without this, np.ndarray would
-        raise "buffer is too small for requested array" when the in-memory
-        cache pointed at an old SHM after a same-name resize."""
+        current (correctly sized) segment reopened. Without this, np.ndarray
+        would raise "buffer is too small for requested array" when the
+        in-memory cache pointed at an old SHM after a same-name resize."""
         manager = SharedMemoryFrameManager()
 
         small = _fake_shm(size=100)
-        large = _fake_shm(size=10_000)
+        current = _fake_shm(size=2_500)
         manager.shm_store["cam_frame0"] = small
 
-        with patch("frigate.util.image.UntrackedSharedMemory", return_value=large):
+        with patch("frigate.util.image.UntrackedSharedMemory", return_value=current):
             arr = manager.get("cam_frame0", (50, 50))
 
         self.assertIsNotNone(arr)
-        # Numpy now reports against the large segment, not the small one.
         self.assertEqual(arr.shape, (50, 50))
-        self.assertIs(manager.shm_store["cam_frame0"], large)
+        self.assertIs(manager.shm_store["cam_frame0"], current)
 
-    def test_get_keeps_cached_segment_when_size_sufficient(self) -> None:
-        """Don't pay the reopen cost when the cached ref is fine."""
+    def test_get_reopens_when_cached_segment_is_larger_than_shape(self) -> None:
+        """Symmetric to the smaller-cache case: when detect resolution drops,
+        the SHM is unlinked and recreated at a smaller size. A cached ref to
+        the old, larger segment still satisfies any size check but points at
+        an orphaned inode whose stale bytes get reinterpreted at the new
+        shape — producing miscolored, distorted YUV frames downstream. Drop
+        the cache so we reopen by name and bind to the current segment."""
         manager = SharedMemoryFrameManager()
 
-        cached = _fake_shm(size=10_000)
+        old_large = _fake_shm(size=10_000)
+        current = _fake_shm(size=2_500)
+        manager.shm_store["cam_frame0"] = old_large
+
+        with patch("frigate.util.image.UntrackedSharedMemory", return_value=current):
+            arr = manager.get("cam_frame0", (50, 50))
+
+        self.assertIsNotNone(arr)
+        self.assertEqual(arr.shape, (50, 50))
+        self.assertIs(manager.shm_store["cam_frame0"], current)
+
+    def test_get_keeps_cached_segment_when_size_matches(self) -> None:
+        """Don't pay the reopen cost when the cached ref is the right size."""
+        manager = SharedMemoryFrameManager()
+
+        cached = _fake_shm(size=2_500)
         manager.shm_store["cam_frame0"] = cached
 
         with patch("frigate.util.image.UntrackedSharedMemory") as untracked_shm_cls:
@@ -56,7 +75,7 @@ class TestSharedMemoryFrameManagerGet(unittest.TestCase):
 
     def test_get_opens_fresh_when_no_cache_entry(self) -> None:
         manager = SharedMemoryFrameManager()
-        fresh = _fake_shm(size=10_000)
+        fresh = _fake_shm(size=2_500)
 
         with patch("frigate.util.image.UntrackedSharedMemory", return_value=fresh):
             arr = manager.get("cam_frame0", (50, 50))