add tests

frigate/test/test_dispatcher_runtime_state.py

@@ -0,0 +1,217 @@
+"""Tests for Dispatcher runtime state persistence wiring."""
+
+import os
+import tempfile
+import unittest
+from unittest.mock import MagicMock, patch
+
+from frigate.comms.dispatcher import Dispatcher
+from frigate.comms.runtime_state import RuntimeStatePersistence
+
+
+def _make_camera_mock(
+    *,
+    enabled: bool = True,
+    enabled_in_config: bool = True,
+    detect_enabled: bool = True,
+    record_enabled: bool = True,
+    record_enabled_in_config: bool = True,
+    snapshots_enabled: bool = True,
+    audio_enabled: bool = True,
+    audio_enabled_in_config: bool = True,
+) -> MagicMock:
+    """Build a camera config mock with the fields the in-scope handlers read."""
+    camera = MagicMock()
+    camera.enabled = enabled
+    camera.enabled_in_config = enabled_in_config
+    camera.detect.enabled = detect_enabled
+    camera.motion.enabled = True  # avoid the detect→motion side-effect path
+    camera.record.enabled = record_enabled
+    camera.record.enabled_in_config = record_enabled_in_config
+    camera.snapshots.enabled = snapshots_enabled
+    camera.audio.enabled = audio_enabled
+    camera.audio.enabled_in_config = audio_enabled_in_config
+    return camera
+
+
+def _build_dispatcher(cameras: dict[str, MagicMock]) -> Dispatcher:
+    """Construct a Dispatcher with the bare-minimum mocks the tests need."""
+    config = MagicMock()
+    config.cameras = cameras
+    config_updater = MagicMock()
+    onvif = MagicMock()
+    ptz_metrics: dict = {}
+    communicators: list = []
+
+    with (
+        patch("frigate.comms.dispatcher.CameraActivityManager"),
+        patch("frigate.comms.dispatcher.AudioActivityManager"),
+    ):
+        return Dispatcher(config, config_updater, onvif, ptz_metrics, communicators)
+
+
+class TestRestoreRuntimeState(unittest.TestCase):
+    """Verify replay routes through handlers and tolerates missing entries."""
+
+    def setUp(self) -> None:
+        self.dispatcher = _build_dispatcher(
+            {
+                "front_door": _make_camera_mock(),
+                "back_yard": _make_camera_mock(),
+            }
+        )
+        # Swap each in-scope handler for a MagicMock so we can assert calls
+        # without exercising the handler's own logic.
+        self.handler_mocks: dict[str, MagicMock] = {}
+        for topic in ("enabled", "detect", "snapshots", "recordings", "audio"):
+            mock = MagicMock()
+            self.dispatcher._camera_settings_handlers[topic] = mock
+            self.handler_mocks[topic] = mock
+
+    def test_replays_each_stored_entry_through_its_handler(self) -> None:
+        self.dispatcher._runtime_state = MagicMock(
+            spec=RuntimeStatePersistence,
+            load=MagicMock(
+                return_value={
+                    "front_door": {"detect": False, "recordings": False},
+                    "back_yard": {"audio": False},
+                }
+            ),
+        )
+        self.dispatcher.restore_runtime_state()
+
+        self.handler_mocks["detect"].assert_called_once_with("front_door", "OFF")
+        self.handler_mocks["recordings"].assert_called_once_with("front_door", "OFF")
+        self.handler_mocks["audio"].assert_called_once_with("back_yard", "OFF")
+        self.handler_mocks["enabled"].assert_not_called()
+        self.handler_mocks["snapshots"].assert_not_called()
+
+    def test_skips_unknown_cameras(self) -> None:
+        self.dispatcher._runtime_state = MagicMock(
+            spec=RuntimeStatePersistence,
+            load=MagicMock(return_value={"removed_cam": {"detect": False}}),
+        )
+        self.dispatcher.restore_runtime_state()
+        for mock in self.handler_mocks.values():
+            mock.assert_not_called()
+
+    def test_skips_unknown_topics(self) -> None:
+        self.dispatcher._runtime_state = MagicMock(
+            spec=RuntimeStatePersistence,
+            load=MagicMock(return_value={"front_door": {"some_old_topic": True}}),
+        )
+        self.dispatcher.restore_runtime_state()
+        for mock in self.handler_mocks.values():
+            mock.assert_not_called()
+
+    def test_continues_after_handler_exception(self) -> None:
+        self.handler_mocks["detect"].side_effect = RuntimeError("boom")
+        self.dispatcher._runtime_state = MagicMock(
+            spec=RuntimeStatePersistence,
+            load=MagicMock(
+                return_value={
+                    "front_door": {"detect": False, "recordings": False},
+                }
+            ),
+        )
+        # Must not raise; the recordings handler must still run.
+        self.dispatcher.restore_runtime_state()
+        self.handler_mocks["recordings"].assert_called_once_with("front_door", "OFF")
+
+    def test_true_value_routes_as_on_payload(self) -> None:
+        self.dispatcher._runtime_state = MagicMock(
+            spec=RuntimeStatePersistence,
+            load=MagicMock(return_value={"front_door": {"detect": True}}),
+        )
+        self.dispatcher.restore_runtime_state()
+        self.handler_mocks["detect"].assert_called_once_with("front_door", "ON")
+
+
+class TestHandlersPersistViaSet(unittest.TestCase):
+    """Verify each in-scope handler writes to the runtime state on success."""
+
+    def setUp(self) -> None:
+        self.tmp_dir = tempfile.mkdtemp()
+        self.config_path = os.path.join(self.tmp_dir, "config.yml")
+        with open(self.config_path, "w") as f:
+            f.write("")
+        self._patcher = patch(
+            "frigate.comms.runtime_state.find_config_file",
+            return_value=self.config_path,
+        )
+        self._patcher.start()
+
+        # Start with everything OFF so each ON payload triggers a real change
+        self.cameras = {
+            "front_door": _make_camera_mock(
+                enabled=False,
+                detect_enabled=False,
+                record_enabled=False,
+                snapshots_enabled=False,
+                audio_enabled=False,
+            )
+        }
+        self.dispatcher = _build_dispatcher(self.cameras)
+
+    def tearDown(self) -> None:
+        self._patcher.stop()
+        for name in os.listdir(self.tmp_dir):
+            os.remove(os.path.join(self.tmp_dir, name))
+        os.rmdir(self.tmp_dir)
+
+    def _stored_state(self) -> dict:
+        return RuntimeStatePersistence().load()
+
+    def test_enabled_handler_persists(self) -> None:
+        self.dispatcher._on_enabled_command("front_door", "ON")
+        self.assertEqual(self._stored_state(), {"front_door": {"enabled": True}})
+
+    def test_detect_handler_persists(self) -> None:
+        self.dispatcher._on_detect_command("front_door", "ON")
+        self.assertEqual(self._stored_state(), {"front_door": {"detect": True}})
+
+    def test_recordings_handler_persists(self) -> None:
+        self.dispatcher._on_recordings_command("front_door", "ON")
+        self.assertEqual(self._stored_state(), {"front_door": {"recordings": True}})
+
+    def test_snapshots_handler_persists(self) -> None:
+        self.dispatcher._on_snapshots_command("front_door", "ON")
+        self.assertEqual(self._stored_state(), {"front_door": {"snapshots": True}})
+
+    def test_audio_handler_persists(self) -> None:
+        self.dispatcher._on_audio_command("front_door", "ON")
+        self.assertEqual(self._stored_state(), {"front_door": {"audio": True}})
+
+    def test_enabled_in_config_gate_blocks_persistence(self) -> None:
+        """An ON payload rejected by the gate must not be persisted."""
+        cam = self.cameras["front_door"]
+        cam.enabled_in_config = False
+        cam.record.enabled_in_config = False
+        cam.audio.enabled_in_config = False
+
+        self.dispatcher._on_enabled_command("front_door", "ON")
+        self.dispatcher._on_recordings_command("front_door", "ON")
+        self.dispatcher._on_audio_command("front_door", "ON")
+
+        self.assertEqual(self._stored_state(), {})
+
+
+class TestClearPassthrough(unittest.TestCase):
+    """The dispatcher's public clear methods delegate to the store."""
+
+    def test_clear_runtime_state_for_yaml_keys_passthrough(self) -> None:
+        dispatcher = _build_dispatcher({})
+        dispatcher._runtime_state = MagicMock(spec=RuntimeStatePersistence)
+        keys = ["cameras.front_door.detect.enabled"]
+        dispatcher.clear_runtime_state_for_yaml_keys(keys)
+        dispatcher._runtime_state.clear_for_yaml_keys.assert_called_once_with(keys)
+
+    def test_clear_runtime_state_passthrough(self) -> None:
+        dispatcher = _build_dispatcher({})
+        dispatcher._runtime_state = MagicMock(spec=RuntimeStatePersistence)
+        dispatcher.clear_runtime_state()
+        dispatcher._runtime_state.clear_all.assert_called_once_with()
+
+
+if __name__ == "__main__":
+    unittest.main()

frigate/test/test_profiles.py

@@ -727,6 +727,55 @@ class TestProfileManager(unittest.TestCase):
         # Should not raise
         json.dumps(api_base)
 
+    @patch.object(ProfileManager, "_persist_active_profile")
+    def test_activate_profile_clears_dispatcher_runtime_state(self, mock_persist):
+        """User-initiated activation drops runtime overrides (steady-state rule)."""
+        dispatcher = MagicMock()
+        manager = ProfileManager(self.config, self.mock_updater, dispatcher)
+        manager.activate_profile("armed")
+        dispatcher.clear_runtime_state.assert_called_once_with()
+
+    @patch.object(ProfileManager, "_persist_active_profile")
+    def test_deactivate_profile_clears_dispatcher_runtime_state(self, mock_persist):
+        """Deactivating a profile also drops runtime overrides."""
+        dispatcher = MagicMock()
+        manager = ProfileManager(self.config, self.mock_updater, dispatcher)
+        manager.activate_profile("armed")
+        dispatcher.clear_runtime_state.reset_mock()
+
+        manager.activate_profile(None)
+        dispatcher.clear_runtime_state.assert_called_once_with()
+
+    @patch.object(ProfileManager, "_persist_active_profile")
+    def test_startup_replay_does_not_clear_runtime_state(self, mock_persist):
+        """Startup callers pass clear_runtime_overrides=False to preserve state."""
+        dispatcher = MagicMock()
+        manager = ProfileManager(self.config, self.mock_updater, dispatcher)
+        manager.activate_profile("armed", clear_runtime_overrides=False)
+        dispatcher.clear_runtime_state.assert_not_called()
+
+    @patch.object(ProfileManager, "_persist_active_profile")
+    def test_update_config_clears_when_active_profile_reapplies(self, mock_persist):
+        """After /api/config/set, an active-profile re-application drops state."""
+        dispatcher = MagicMock()
+        manager = ProfileManager(self.config, self.mock_updater, dispatcher)
+        manager.activate_profile("armed")
+        dispatcher.clear_runtime_state.reset_mock()
+
+        new_config = FrigateConfig(**self.config_data)
+        manager.update_config(new_config)
+        dispatcher.clear_runtime_state.assert_called_once_with()
+
+    @patch.object(ProfileManager, "_persist_active_profile")
+    def test_update_config_does_not_clear_when_no_active_profile(self, mock_persist):
+        """Plain /api/config/set without a profile doesn't trigger the broad clear."""
+        dispatcher = MagicMock()
+        manager = ProfileManager(self.config, self.mock_updater, dispatcher)
+        # No activate_profile call — config.active_profile is None
+        new_config = FrigateConfig(**self.config_data)
+        manager.update_config(new_config)
+        dispatcher.clear_runtime_state.assert_not_called()
+
 
 class TestProfilePersistence(unittest.TestCase):
     """Test profile persistence to disk."""

frigate/test/test_runtime_state.py

@@ -0,0 +1,136 @@
+"""Tests for RuntimeStatePersistence."""
+
+import json
+import os
+import tempfile
+import unittest
+from unittest.mock import patch
+
+from frigate.comms.runtime_state import RuntimeStatePersistence
+
+
+class TestRuntimeStatePersistence(unittest.TestCase):
+    """Unit tests for the JSON-backed runtime state store."""
+
+    def setUp(self) -> None:
+        self.tmp_dir = tempfile.mkdtemp()
+        self.config_path = os.path.join(self.tmp_dir, "config.yml")
+        # Touch a placeholder config.yml so find_config_file returns a real path
+        with open(self.config_path, "w") as f:
+            f.write("")
+        self._patcher = patch(
+            "frigate.comms.runtime_state.find_config_file",
+            return_value=self.config_path,
+        )
+        self._patcher.start()
+        self.store = RuntimeStatePersistence()
+
+    def tearDown(self) -> None:
+        self._patcher.stop()
+        for name in os.listdir(self.tmp_dir):
+            os.remove(os.path.join(self.tmp_dir, name))
+        os.rmdir(self.tmp_dir)
+
+    def test_load_returns_empty_when_file_missing(self) -> None:
+        self.assertEqual(self.store.load(), {})
+
+    def test_set_then_load_round_trip(self) -> None:
+        self.store.set("front_door", "detect", False)
+        self.store.set("front_door", "recordings", True)
+        self.store.set("back_yard", "audio", False)
+
+        result = self.store.load()
+        self.assertEqual(
+            result,
+            {
+                "front_door": {"detect": False, "recordings": True},
+                "back_yard": {"audio": False},
+            },
+        )
+
+    def test_set_with_untracked_topic_is_noop(self) -> None:
+        self.store.set("front_door", "ptz_autotracker", True)
+        self.assertEqual(self.store.load(), {})
+        # File should not even be created if no tracked entries were written
+        runtime_path = os.path.join(self.tmp_dir, ".runtime_state.json")
+        self.assertFalse(os.path.exists(runtime_path))
+
+    def test_set_overwrites_previous_value(self) -> None:
+        self.store.set("front_door", "detect", True)
+        self.store.set("front_door", "detect", False)
+        self.assertEqual(self.store.load(), {"front_door": {"detect": False}})
+
+    def test_load_returns_empty_when_file_corrupt(self) -> None:
+        runtime_path = os.path.join(self.tmp_dir, ".runtime_state.json")
+        with open(runtime_path, "w") as f:
+            f.write("{not valid json")
+        self.assertEqual(self.store.load(), {})
+
+    def test_load_handles_unexpected_top_level_shape(self) -> None:
+        runtime_path = os.path.join(self.tmp_dir, ".runtime_state.json")
+        with open(runtime_path, "w") as f:
+            json.dump(["unexpected", "list"], f)
+        self.assertEqual(self.store.load(), {})
+
+    def test_clear_for_yaml_keys_removes_matching_entries(self) -> None:
+        self.store.set("front_door", "detect", False)
+        self.store.set("front_door", "recordings", False)
+        self.store.set("back_yard", "audio", False)
+
+        self.store.clear_for_yaml_keys(
+            [
+                "cameras.front_door.detect.enabled",
+                "cameras.back_yard.audio.enabled",
+            ]
+        )
+
+        self.assertEqual(
+            self.store.load(),
+            {"front_door": {"recordings": False}},
+        )
+
+    def test_clear_for_yaml_keys_collapses_empty_camera_dict(self) -> None:
+        self.store.set("front_door", "detect", False)
+        self.store.clear_for_yaml_keys(["cameras.front_door.detect.enabled"])
+        self.assertEqual(self.store.load(), {})
+
+    def test_clear_for_yaml_keys_ignores_unrelated_keys(self) -> None:
+        self.store.set("front_door", "detect", False)
+        self.store.clear_for_yaml_keys(
+            [
+                "ui.theme",
+                "go2rtc.streams.x",
+                "cameras.front_door.ffmpeg.inputs",
+                "not_cameras.front_door.detect.enabled",
+            ]
+        )
+        self.assertEqual(self.store.load(), {"front_door": {"detect": False}})
+
+    def test_clear_for_yaml_keys_handles_empty_iterable(self) -> None:
+        self.store.set("front_door", "detect", False)
+        self.store.clear_for_yaml_keys([])
+        self.assertEqual(self.store.load(), {"front_door": {"detect": False}})
+
+    def test_camera_level_enabled_uses_top_level_yaml_key(self) -> None:
+        """`enabled` topic maps to the camera-level `cameras.<cam>.enabled` key."""
+        self.store.set("front_door", "enabled", False)
+        self.store.clear_for_yaml_keys(["cameras.front_door.enabled"])
+        self.assertEqual(self.store.load(), {})
+
+    def test_clear_all_wipes_every_entry(self) -> None:
+        self.store.set("front_door", "detect", False)
+        self.store.set("front_door", "recordings", True)
+        self.store.set("back_yard", "audio", False)
+
+        self.store.clear_all()
+
+        self.assertEqual(self.store.load(), {})
+
+    def test_clear_all_is_safe_when_file_missing(self) -> None:
+        # No prior set() calls — file does not exist
+        self.store.clear_all()
+        self.assertEqual(self.store.load(), {})
+
+
+if __name__ == "__main__":
+    unittest.main()