mirror of
https://github.com/blakeblackshear/frigate.git
synced 2026-07-28 14:49:01 +03:00
Josh HawkinsandGitHub
48aaafba3c
/api/config/set and camera deletion re-parse yaml into a fresh FrigateConfig and swap it in, then re-layered the persisted runtime toggle overrides so a camera the user turned off wouldn't come back on. That re-layer ran apply_runtime_state, which replays each override through the command handlers, so every save re-published a ZMQ config update, a retained MQTT state message, and a runtime-state disk write for every camera with a stored toggle. All of it was redundant: the worker processes were never swapped and still hold the live toggle values, so only the in-process config object the API and dispatcher read was out of date. The extra traffic churned the retained MQTT topics, amplified disk writes, and co-drained enabled updates with other topics on the config socket. Add Dispatcher.reapply_runtime_state_to_config, which corrects only the swapped-in config object, mirroring the field mutations and gates of the _on_*_command handlers with no ZMQ, MQTT, or disk writes. swap_runtime_config now calls it instead of apply_runtime_state; apply_runtime_state is unchanged and still used at startup, where the workers genuinely must be told.
368 lines
14 KiB
Python
368 lines
14 KiB
Python
"""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")
|
|
|
|
def test_apply_runtime_state_replays_through_handlers(self) -> None:
|
|
"""The extracted method replays every stored entry."""
|
|
with patch.object(
|
|
self.dispatcher._runtime_state,
|
|
"load",
|
|
return_value={"front_door": {"enabled": False, "detect": True}},
|
|
):
|
|
self.dispatcher.apply_runtime_state()
|
|
|
|
self.handler_mocks["enabled"].assert_called_once_with("front_door", "OFF")
|
|
self.handler_mocks["detect"].assert_called_once_with("front_door", "ON")
|
|
|
|
def test_apply_runtime_state_returns_applied_entries(self) -> None:
|
|
"""Callers get back what was replayed, for logging and assertions."""
|
|
with patch.object(
|
|
self.dispatcher._runtime_state,
|
|
"load",
|
|
return_value={"front_door": {"enabled": False}, "nope": {"enabled": True}},
|
|
):
|
|
applied = self.dispatcher.apply_runtime_state()
|
|
|
|
self.assertEqual(applied, {"front_door": {"enabled": False}})
|
|
|
|
def test_restore_runtime_state_still_replays(self) -> None:
|
|
"""The startup entry point keeps working after the extraction."""
|
|
with patch.object(
|
|
self.dispatcher._runtime_state,
|
|
"load",
|
|
return_value={"back_yard": {"snapshots": False}},
|
|
):
|
|
self.dispatcher.restore_runtime_state()
|
|
|
|
self.handler_mocks["snapshots"].assert_called_once_with("back_yard", "OFF")
|
|
|
|
|
|
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()
|
|
|
|
def test_clear_runtime_state_for_camera_passthrough(self) -> None:
|
|
dispatcher = _build_dispatcher({})
|
|
dispatcher._runtime_state = MagicMock(spec=RuntimeStatePersistence)
|
|
dispatcher.clear_runtime_state_for_camera("front_door")
|
|
dispatcher._runtime_state.clear_camera.assert_called_once_with("front_door")
|
|
|
|
|
|
class TestReapplyRuntimeStateToConfig(unittest.TestCase):
|
|
"""The silent re-apply corrects the config object with no side effects."""
|
|
|
|
def _dispatcher_with(
|
|
self, cameras: dict[str, MagicMock], state: dict
|
|
) -> Dispatcher:
|
|
dispatcher = _build_dispatcher(cameras)
|
|
dispatcher._runtime_state = MagicMock(spec=RuntimeStatePersistence)
|
|
dispatcher._runtime_state.load.return_value = state
|
|
dispatcher.publish = MagicMock()
|
|
return dispatcher
|
|
|
|
def test_mutates_every_tracked_field(self) -> None:
|
|
cameras = {"front_door": _make_camera_mock()}
|
|
dispatcher = self._dispatcher_with(
|
|
cameras,
|
|
{
|
|
"front_door": {
|
|
"enabled": False,
|
|
"detect": False,
|
|
"snapshots": False,
|
|
"recordings": False,
|
|
"audio": False,
|
|
}
|
|
},
|
|
)
|
|
|
|
dispatcher.reapply_runtime_state_to_config()
|
|
|
|
cam = cameras["front_door"]
|
|
self.assertFalse(cam.enabled)
|
|
self.assertFalse(cam.detect.enabled)
|
|
self.assertFalse(cam.snapshots.enabled)
|
|
self.assertFalse(cam.record.enabled)
|
|
self.assertFalse(cam.audio.enabled)
|
|
|
|
def test_makes_no_zmq_mqtt_or_disk_writes(self) -> None:
|
|
dispatcher = self._dispatcher_with(
|
|
{"front_door": _make_camera_mock()},
|
|
{"front_door": {"enabled": False}},
|
|
)
|
|
|
|
dispatcher.reapply_runtime_state_to_config()
|
|
|
|
dispatcher.config_updater.publish_update.assert_not_called()
|
|
dispatcher._runtime_state.set.assert_not_called()
|
|
dispatcher.publish.assert_not_called()
|
|
|
|
def test_respects_enabled_in_config_gate(self) -> None:
|
|
# an ON override for a camera disabled in yaml must not enable it
|
|
cameras = {
|
|
"front_door": _make_camera_mock(enabled=False, enabled_in_config=False)
|
|
}
|
|
dispatcher = self._dispatcher_with(cameras, {"front_door": {"enabled": True}})
|
|
|
|
dispatcher.reapply_runtime_state_to_config()
|
|
|
|
self.assertFalse(cameras["front_door"].enabled)
|
|
|
|
def test_respects_recordings_and_audio_gates(self) -> None:
|
|
# ON overrides for recordings/audio not enabled in yaml must be ignored
|
|
cameras = {
|
|
"front_door": _make_camera_mock(
|
|
record_enabled=False,
|
|
record_enabled_in_config=False,
|
|
audio_enabled=False,
|
|
audio_enabled_in_config=False,
|
|
)
|
|
}
|
|
dispatcher = self._dispatcher_with(
|
|
cameras, {"front_door": {"recordings": True, "audio": True}}
|
|
)
|
|
|
|
dispatcher.reapply_runtime_state_to_config()
|
|
|
|
self.assertFalse(cameras["front_door"].record.enabled)
|
|
self.assertFalse(cameras["front_door"].audio.enabled)
|
|
|
|
def test_applies_on_override_when_gate_passes(self) -> None:
|
|
# a camera off in yaml but enabled_in_config keeps its runtime-on state
|
|
cameras = {
|
|
"front_door": _make_camera_mock(enabled=False, enabled_in_config=True)
|
|
}
|
|
dispatcher = self._dispatcher_with(cameras, {"front_door": {"enabled": True}})
|
|
|
|
dispatcher.reapply_runtime_state_to_config()
|
|
|
|
self.assertTrue(cameras["front_door"].enabled)
|
|
|
|
def test_detect_on_couples_motion(self) -> None:
|
|
cam = _make_camera_mock(detect_enabled=False)
|
|
cam.motion.enabled = False
|
|
dispatcher = self._dispatcher_with(
|
|
{"front_door": cam}, {"front_door": {"detect": True}}
|
|
)
|
|
|
|
dispatcher.reapply_runtime_state_to_config()
|
|
|
|
self.assertTrue(cam.detect.enabled)
|
|
self.assertTrue(cam.motion.enabled)
|
|
|
|
def test_skips_camera_not_in_config(self) -> None:
|
|
dispatcher = self._dispatcher_with(
|
|
{"front_door": _make_camera_mock()}, {"ghost": {"enabled": False}}
|
|
)
|
|
|
|
# a stale entry for a deleted camera must be ignored, not raise
|
|
dispatcher.reapply_runtime_state_to_config()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|