From 904b85db2d1a40f09e030908a4a489b558e1208c Mon Sep 17 00:00:00 2001 From: Josh Hawkins <32435876+hawkeye217@users.noreply.github.com> Date: Wed, 25 Feb 2026 10:42:21 -0600 Subject: [PATCH] add test --- frigate/test/http_api/test_http_config_set.py | 260 ++++++++++++++++++ 1 file changed, 260 insertions(+) create mode 100644 frigate/test/http_api/test_http_config_set.py diff --git a/frigate/test/http_api/test_http_config_set.py b/frigate/test/http_api/test_http_config_set.py new file mode 100644 index 000000000..e8e5f4553 --- /dev/null +++ b/frigate/test/http_api/test_http_config_set.py @@ -0,0 +1,260 @@ +"""Tests for the config_set endpoint's wildcard camera propagation.""" + +import os +import tempfile +import unittest +from unittest.mock import MagicMock, Mock, patch + +import ruamel.yaml + +from frigate.config import FrigateConfig +from frigate.config.camera.updater import ( + CameraConfigUpdateEnum, + CameraConfigUpdatePublisher, + CameraConfigUpdateTopic, +) +from frigate.models import Event, Recordings, ReviewSegment +from frigate.test.http_api.base_http_test import AuthTestClient, BaseTestHttp + + +class TestConfigSetWildcardPropagation(BaseTestHttp): + """Test that wildcard camera updates fan out to all cameras.""" + + def setUp(self): + super().setUp(models=[Event, Recordings, ReviewSegment]) + self.minimal_config = { + "mqtt": {"host": "mqtt"}, + "cameras": { + "front_door": { + "ffmpeg": { + "inputs": [ + {"path": "rtsp://10.0.0.1:554/video", "roles": ["detect"]} + ] + }, + "detect": { + "height": 1080, + "width": 1920, + "fps": 5, + }, + }, + "back_yard": { + "ffmpeg": { + "inputs": [ + {"path": "rtsp://10.0.0.2:554/video", "roles": ["detect"]} + ] + }, + "detect": { + "height": 720, + "width": 1280, + "fps": 10, + }, + }, + }, + } + + def _create_app_with_publisher(self): + """Create app with a mocked config publisher.""" + from fastapi import Request + + from frigate.api.auth import get_allowed_cameras_for_filter, get_current_user + from frigate.api.fastapi_app import create_fastapi_app + + mock_publisher = Mock(spec=CameraConfigUpdatePublisher) + mock_publisher.publisher = MagicMock() + + app = create_fastapi_app( + FrigateConfig(**self.minimal_config), + self.db, + None, + None, + None, + None, + None, + None, + mock_publisher, + enforce_default_admin=False, + ) + + async def mock_get_current_user(request: Request): + username = request.headers.get("remote-user") + role = request.headers.get("remote-role") + return {"username": username, "role": role} + + async def mock_get_allowed_cameras_for_filter(request: Request): + return list(self.minimal_config.get("cameras", {}).keys()) + + app.dependency_overrides[get_current_user] = mock_get_current_user + app.dependency_overrides[get_allowed_cameras_for_filter] = ( + mock_get_allowed_cameras_for_filter + ) + + return app, mock_publisher + + def _write_config_file(self): + """Write the minimal config to a temp YAML file and return the path.""" + yaml = ruamel.yaml.YAML() + f = tempfile.NamedTemporaryFile(mode="w", suffix=".yml", delete=False) + yaml.dump(self.minimal_config, f) + f.close() + return f.name + + @patch("frigate.api.app.find_config_file") + def test_wildcard_detect_update_fans_out_to_all_cameras(self, mock_find_config): + """config/cameras/*/detect fans out to all cameras.""" + config_path = self._write_config_file() + mock_find_config.return_value = config_path + + try: + app, mock_publisher = self._create_app_with_publisher() + with AuthTestClient(app) as client: + resp = client.put( + "/config/set", + json={ + "config_data": {"detect": {"fps": 15}}, + "update_topic": "config/cameras/*/detect", + "requires_restart": 0, + }, + ) + + self.assertEqual(resp.status_code, 200) + data = resp.json() + self.assertTrue(data["success"]) + + # Verify publish_update called for each camera + self.assertEqual(mock_publisher.publish_update.call_count, 2) + + published_cameras = set() + for c in mock_publisher.publish_update.call_args_list: + topic = c[0][0] + self.assertIsInstance(topic, CameraConfigUpdateTopic) + self.assertEqual(topic.update_type, CameraConfigUpdateEnum.detect) + published_cameras.add(topic.camera) + + self.assertEqual(published_cameras, {"front_door", "back_yard"}) + + # Global publisher should NOT be called for wildcard + mock_publisher.publisher.publish.assert_not_called() + finally: + os.unlink(config_path) + + @patch("frigate.api.app.find_config_file") + def test_wildcard_motion_update_fans_out(self, mock_find_config): + """config/cameras/*/motion fans out to all cameras.""" + config_path = self._write_config_file() + mock_find_config.return_value = config_path + + try: + app, mock_publisher = self._create_app_with_publisher() + with AuthTestClient(app) as client: + resp = client.put( + "/config/set", + json={ + "config_data": {"motion": {"threshold": 30}}, + "update_topic": "config/cameras/*/motion", + "requires_restart": 0, + }, + ) + + self.assertEqual(resp.status_code, 200) + + published_cameras = set() + for c in mock_publisher.publish_update.call_args_list: + topic = c[0][0] + self.assertEqual(topic.update_type, CameraConfigUpdateEnum.motion) + published_cameras.add(topic.camera) + + self.assertEqual(published_cameras, {"front_door", "back_yard"}) + finally: + os.unlink(config_path) + + @patch("frigate.api.app.find_config_file") + def test_camera_specific_topic_only_updates_one_camera(self, mock_find_config): + """config/cameras/front_door/detect only updates front_door.""" + config_path = self._write_config_file() + mock_find_config.return_value = config_path + + try: + app, mock_publisher = self._create_app_with_publisher() + with AuthTestClient(app) as client: + resp = client.put( + "/config/set", + json={ + "config_data": { + "cameras": {"front_door": {"detect": {"fps": 20}}} + }, + "update_topic": "config/cameras/front_door/detect", + "requires_restart": 0, + }, + ) + + self.assertEqual(resp.status_code, 200) + + # Only one camera updated + self.assertEqual(mock_publisher.publish_update.call_count, 1) + topic = mock_publisher.publish_update.call_args[0][0] + self.assertEqual(topic.camera, "front_door") + self.assertEqual(topic.update_type, CameraConfigUpdateEnum.detect) + + # Global publisher should NOT be called + mock_publisher.publisher.publish.assert_not_called() + finally: + os.unlink(config_path) + + @patch("frigate.api.app.find_config_file") + def test_wildcard_sends_merged_per_camera_config(self, mock_find_config): + """Wildcard fan-out sends each camera's own merged config.""" + config_path = self._write_config_file() + mock_find_config.return_value = config_path + + try: + app, mock_publisher = self._create_app_with_publisher() + with AuthTestClient(app) as client: + resp = client.put( + "/config/set", + json={ + "config_data": {"detect": {"fps": 15}}, + "update_topic": "config/cameras/*/detect", + "requires_restart": 0, + }, + ) + + self.assertEqual(resp.status_code, 200) + + for c in mock_publisher.publish_update.call_args_list: + camera_detect_config = c[0][1] + self.assertIsNotNone(camera_detect_config) + self.assertTrue(hasattr(camera_detect_config, "fps")) + finally: + os.unlink(config_path) + + @patch("frigate.api.app.find_config_file") + def test_non_camera_global_topic_uses_generic_publish(self, mock_find_config): + """Non-camera topics (e.g. config/live) use the generic publisher.""" + config_path = self._write_config_file() + mock_find_config.return_value = config_path + + try: + app, mock_publisher = self._create_app_with_publisher() + with AuthTestClient(app) as client: + resp = client.put( + "/config/set", + json={ + "config_data": {"live": {"height": 720}}, + "update_topic": "config/live", + "requires_restart": 0, + }, + ) + + self.assertEqual(resp.status_code, 200) + + # Global topic publisher called + mock_publisher.publisher.publish.assert_called_once() + + # Camera-level publish_update NOT called + mock_publisher.publish_update.assert_not_called() + finally: + os.unlink(config_path) + + +if __name__ == "__main__": + unittest.main()