Add tests for recordings retention and fix bug (#7183)

* Add tests for segment info

* Fix logic

frigate/record/maintainer.py

@@ -28,6 +28,25 @@
 logger = logging.getLogger(__name__)
 
 
+class SegmentInfo:
+    def __init__(
+        self, motion_box_count: int, active_object_count: int, average_dBFS: int
+    ) -> None:
+        self.motion_box_count = motion_box_count
+        self.active_object_count = active_object_count
+        self.average_dBFS = average_dBFS
+
+    def should_discard_segment(self, retain_mode: RetainModeEnum) -> bool:
+        return (
+            retain_mode == RetainModeEnum.motion
+            and self.motion_box_count == 0
+            and self.average_dBFS == 0
+        ) or (
+            retain_mode == RetainModeEnum.active_objects
+            and self.active_object_count == 0
+        )
+
+
 class RecordingMaintainer(threading.Thread):
     def __init__(
         self,
@@ -234,7 +253,7 @@ async def validate_and_move_segment(
 
     def segment_stats(
         self, camera: str, start_time: datetime.datetime, end_time: datetime.datetime
-    ) -> Tuple[int, int, int]:
+    ) -> SegmentInfo:
         active_count = 0
         motion_count = 0
         for frame in self.object_recordings_info[camera]:
@@ -269,7 +288,7 @@ def segment_stats(
 
         average_dBFS = 0 if not audio_values else np.average(audio_values)
 
-        return (motion_count, active_count, round(average_dBFS))
+        return SegmentInfo(motion_count, active_count, round(average_dBFS))
 
     def store_segment(
         self,
@@ -280,18 +299,10 @@ def store_segment(
         cache_path: str,
         store_mode: RetainModeEnum,
     ) -> None:
-        motion_count, active_count, averageDBFS = self.segment_stats(
-            camera, start_time, end_time
-        )
+        segment_info = self.segment_stats(camera, start_time, end_time)
 
         # check if the segment shouldn't be stored
-        if (
-            (store_mode == RetainModeEnum.motion and motion_count == 0)
-            or (
-                store_mode == RetainModeEnum.motion and averageDBFS == 0
-            )  # dBFS is stored in a negative scale
-            or (store_mode == RetainModeEnum.active_objects and active_count == 0)
-        ):
+        if segment_info.should_discard_segment(store_mode):
             Path(cache_path).unlink(missing_ok=True)
             self.end_time_cache.pop(cache_path, None)
             return
@@ -364,10 +375,10 @@ def store_segment(
                     start_time=start_time.timestamp(),
                     end_time=end_time.timestamp(),
                     duration=duration,
-                    motion=motion_count,
+                    motion=segment_info.motion_box_count,
                     # TODO: update this to store list of active objects at some point
-                    objects=active_count,
-                    dBFS=averageDBFS,
+                    objects=segment_info.active_object_count,
+                    dBFS=segment_info.average_dBFS,
                     segment_size=segment_size,
                 )
         except Exception as e:

frigate/test/test_record_retention.py

@@ -0,0 +1,33 @@
+import unittest
+
+from frigate.config import RetainModeEnum
+from frigate.record.maintainer import SegmentInfo
+
+
+class TestRecordRetention(unittest.TestCase):
+    def test_motion_should_keep_motion_not_object(self):
+        segment_info = SegmentInfo(
+            motion_box_count=1, active_object_count=0, average_dBFS=0
+        )
+        assert not segment_info.should_discard_segment(RetainModeEnum.motion)
+        assert segment_info.should_discard_segment(RetainModeEnum.active_objects)
+
+    def test_object_should_keep_object_not_motion(self):
+        segment_info = SegmentInfo(
+            motion_box_count=0, active_object_count=1, average_dBFS=0
+        )
+        assert segment_info.should_discard_segment(RetainModeEnum.motion)
+        assert not segment_info.should_discard_segment(RetainModeEnum.active_objects)
+
+    def test_all_should_keep_all(self):
+        segment_info = SegmentInfo(
+            motion_box_count=0, active_object_count=0, average_dBFS=0
+        )
+        assert not segment_info.should_discard_segment(RetainModeEnum.all)
+
+    def test_should_keep_audio_in_motion_mode(self):
+        segment_info = SegmentInfo(
+            motion_box_count=0, active_object_count=0, average_dBFS=1
+        )
+        assert not segment_info.should_discard_segment(RetainModeEnum.motion)
+        assert segment_info.should_discard_segment(RetainModeEnum.active_objects)