From 7c4e63af1f75340c2c634b05306861fdea2f28a8 Mon Sep 17 00:00:00 2001 From: Josh Hawkins <32435876+hawkeye217@users.noreply.github.com> Date: Wed, 27 May 2026 12:41:15 -0500 Subject: [PATCH] add tests --- frigate/test/test_dispatcher_runtime_state.py | 217 ++++++++++++++++++ frigate/test/test_profiles.py | 49 ++++ frigate/test/test_runtime_state.py | 136 +++++++++++ 3 files changed, 402 insertions(+) create mode 100644 frigate/test/test_dispatcher_runtime_state.py create mode 100644 frigate/test/test_runtime_state.py diff --git a/frigate/test/test_dispatcher_runtime_state.py b/frigate/test/test_dispatcher_runtime_state.py new file mode 100644 index 0000000000..dae0518d80 --- /dev/null +++ b/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() diff --git a/frigate/test/test_profiles.py b/frigate/test/test_profiles.py index 51c3d78292..6766b39163 100644 --- a/frigate/test/test_profiles.py +++ b/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.""" diff --git a/frigate/test/test_runtime_state.py b/frigate/test/test_runtime_state.py new file mode 100644 index 0000000000..6143184030 --- /dev/null +++ b/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..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()