"""Tests for the profiles system.""" import copy import json import os import unittest from unittest.mock import MagicMock, patch from frigate.config import FrigateConfig from frigate.config.camera.profile import CameraProfileConfig from frigate.config.profile import ProfileDefinitionConfig from frigate.config.profile_manager import PERSISTENCE_FILE, ProfileManager from frigate.const import MODEL_CACHE_DIR class TestCameraProfileConfig(unittest.TestCase): """Test the CameraProfileConfig Pydantic model.""" def test_empty_profile(self): """All sections default to None.""" profile = CameraProfileConfig() assert profile.detect is None assert profile.motion is None assert profile.objects is None assert profile.review is None assert profile.notifications is None def test_partial_detect(self): """Profile with only detect.enabled set.""" profile = CameraProfileConfig(detect={"enabled": False}) assert profile.detect is not None assert profile.detect.enabled is False dumped = profile.detect.model_dump(exclude_unset=True) assert dumped == {"enabled": False} def test_partial_notifications(self): """Profile with only notifications.enabled set.""" profile = CameraProfileConfig(notifications={"enabled": True}) assert profile.notifications is not None assert profile.notifications.enabled is True dumped = profile.notifications.model_dump(exclude_unset=True) assert dumped == {"enabled": True} def test_partial_objects(self): """Profile with objects.track set.""" profile = CameraProfileConfig(objects={"track": ["car", "package"]}) assert profile.objects is not None assert profile.objects.track == ["car", "package"] def test_partial_review(self): """Profile with nested review.alerts.labels.""" profile = CameraProfileConfig(review={"alerts": {"labels": ["person", "car"]}}) assert profile.review is not None assert profile.review.alerts.labels == ["person", "car"] def test_enabled_field(self): """Profile with enabled set to False.""" profile = CameraProfileConfig(enabled=False) assert profile.enabled is False dumped = profile.model_dump(exclude_unset=True) assert dumped == {"enabled": False} def test_enabled_field_true(self): """Profile with enabled set to True.""" profile = CameraProfileConfig(enabled=True) assert profile.enabled is True def test_enabled_default_none(self): """Enabled defaults to None when not set.""" profile = CameraProfileConfig() assert profile.enabled is None def test_zones_field(self): """Profile with zones override.""" profile = CameraProfileConfig( zones={ "driveway": { "coordinates": "0.1,0.1,0.9,0.1,0.9,0.9,0.1,0.9", "objects": ["car"], } } ) assert profile.zones is not None assert "driveway" in profile.zones def test_zones_default_none(self): """Zones defaults to None when not set.""" profile = CameraProfileConfig() assert profile.zones is None def test_none_sections_not_in_dump(self): """Sections left as None should not appear in exclude_unset dump.""" profile = CameraProfileConfig(detect={"enabled": False}) dumped = profile.model_dump(exclude_unset=True) assert "detect" in dumped assert "motion" not in dumped assert "objects" not in dumped def test_invalid_field_value_rejected(self): """Invalid field values are caught by Pydantic.""" from pydantic import ValidationError with self.assertRaises(ValidationError): CameraProfileConfig(detect={"fps": "not_a_number"}) def test_invalid_section_key_rejected(self): """Unknown section keys are rejected (extra=forbid from FrigateBaseModel).""" from pydantic import ValidationError with self.assertRaises(ValidationError): CameraProfileConfig(ffmpeg={"inputs": []}) def test_invalid_nested_field_rejected(self): """Invalid nested field values are caught.""" from pydantic import ValidationError with self.assertRaises(ValidationError): CameraProfileConfig(review={"alerts": {"labels": "not_a_list"}}) def test_invalid_profile_in_camera_config(self): """Invalid profile section in full config is caught at parse time.""" from pydantic import ValidationError config_data = { "mqtt": {"host": "mqtt"}, "profiles": { "armed": {"friendly_name": "Armed"}, }, "cameras": { "front": { "ffmpeg": { "inputs": [ { "path": "rtsp://10.0.0.1:554/video", "roles": ["detect"], } ] }, "detect": {"height": 1080, "width": 1920, "fps": 5}, "profiles": { "armed": { "detect": {"fps": "invalid"}, }, }, }, }, } with self.assertRaises(ValidationError): FrigateConfig(**config_data) def test_undefined_profile_reference_rejected(self): """Camera referencing a profile not defined in top-level profiles is rejected.""" from pydantic import ValidationError config_data = { "mqtt": {"host": "mqtt"}, "profiles": { "armed": {"friendly_name": "Armed"}, }, "cameras": { "front": { "ffmpeg": { "inputs": [ { "path": "rtsp://10.0.0.1:554/video", "roles": ["detect"], } ] }, "detect": {"height": 1080, "width": 1920, "fps": 5}, "profiles": { "nonexistent": { "detect": {"enabled": False}, }, }, }, }, } with self.assertRaises(ValidationError): FrigateConfig(**config_data) def test_profile_zone_without_base_rejected(self): """Profile defining a zone not present on the base camera is rejected.""" from pydantic import ValidationError config_data = { "mqtt": {"host": "mqtt"}, "profiles": { "armed": {"friendly_name": "Armed"}, }, "cameras": { "front": { "ffmpeg": { "inputs": [ { "path": "rtsp://10.0.0.1:554/video", "roles": ["detect"], } ] }, "detect": {"height": 1080, "width": 1920, "fps": 5}, "zones": { "front_yard": {"coordinates": "0,0,100,0,100,100,0,100"}, }, "profiles": { "armed": { "zones": { "phantom": { "coordinates": "0,0,50,0,50,50,0,50", }, }, }, }, }, }, } with self.assertRaises(ValidationError) as ctx: FrigateConfig(**config_data) self.assertIn("phantom", str(ctx.exception)) def test_profile_motion_mask_without_base_rejected(self): """Profile defining a motion mask not present on the base camera is rejected.""" from pydantic import ValidationError config_data = { "mqtt": {"host": "mqtt"}, "profiles": { "armed": {"friendly_name": "Armed"}, }, "cameras": { "front": { "ffmpeg": { "inputs": [ { "path": "rtsp://10.0.0.1:554/video", "roles": ["detect"], } ] }, "detect": {"height": 1080, "width": 1920, "fps": 5}, "motion": { "mask": { "base_mask": { "coordinates": "0,0,100,0,100,100,0,100", }, }, }, "profiles": { "armed": { "motion": { "mask": { "phantom_mask": { "coordinates": "0,0,50,0,50,50,0,50", }, }, }, }, }, }, }, } with self.assertRaises(ValidationError) as ctx: FrigateConfig(**config_data) self.assertIn("phantom_mask", str(ctx.exception)) def test_profile_overrides_matching_base_accepted(self): """Profile overrides that reference existing base zones/masks parse cleanly.""" config_data = { "mqtt": {"host": "mqtt"}, "profiles": { "armed": {"friendly_name": "Armed"}, }, "cameras": { "front": { "ffmpeg": { "inputs": [ { "path": "rtsp://10.0.0.1:554/video", "roles": ["detect"], } ] }, "detect": {"height": 1080, "width": 1920, "fps": 5}, "zones": { "front_yard": {"coordinates": "0,0,100,0,100,100,0,100"}, }, "motion": { "mask": { "tree": { "coordinates": "0,0,100,0,100,100,0,100", }, }, }, "profiles": { "armed": { "zones": { "front_yard": { "coordinates": "0,0,50,0,50,50,0,50", "inertia": 5, }, }, "motion": { "mask": { "tree": { "coordinates": "0,0,75,0,75,75,0,75", }, }, }, }, }, }, }, } config = FrigateConfig(**config_data) assert "armed" in config.cameras["front"].profiles class TestProfileInConfig(unittest.TestCase): """Test that profiles parse correctly in FrigateConfig.""" def setUp(self): self.base_config = { "mqtt": {"host": "mqtt"}, "profiles": { "armed": {"friendly_name": "Armed"}, "disarmed": {"friendly_name": "Disarmed"}, }, "cameras": { "front": { "ffmpeg": { "inputs": [ { "path": "rtsp://10.0.0.1:554/video", "roles": ["detect"], } ] }, "detect": {"height": 1080, "width": 1920, "fps": 5}, "profiles": { "armed": { "notifications": {"enabled": True}, "objects": {"track": ["person", "car", "package"]}, }, "disarmed": { "notifications": {"enabled": False}, "objects": {"track": ["package"]}, }, }, }, "back": { "ffmpeg": { "inputs": [ { "path": "rtsp://10.0.0.2:554/video", "roles": ["detect"], } ] }, "detect": {"height": 1080, "width": 1920, "fps": 5}, "profiles": { "armed": { "detect": {"enabled": True}, }, }, }, }, } if not os.path.exists(MODEL_CACHE_DIR) and not os.path.islink(MODEL_CACHE_DIR): os.makedirs(MODEL_CACHE_DIR) def test_profiles_parse(self): """Profiles are parsed into Dict[str, CameraProfileConfig].""" config = FrigateConfig(**self.base_config) front = config.cameras["front"] assert "armed" in front.profiles assert "disarmed" in front.profiles assert isinstance(front.profiles["armed"], CameraProfileConfig) def test_profile_sections_parsed(self): """Profile sections are properly typed.""" config = FrigateConfig(**self.base_config) armed = config.cameras["front"].profiles["armed"] assert armed.notifications is not None assert armed.notifications.enabled is True assert armed.objects is not None assert armed.objects.track == ["person", "car", "package"] assert armed.detect is None # not set in this profile def test_camera_without_profiles(self): """Camera with no profiles has empty dict.""" config_data = { "mqtt": {"host": "mqtt"}, "cameras": { "front": { "ffmpeg": { "inputs": [ { "path": "rtsp://10.0.0.1:554/video", "roles": ["detect"], } ] }, "detect": {"height": 1080, "width": 1920, "fps": 5}, }, }, } config = FrigateConfig(**config_data) assert config.cameras["front"].profiles == {} class TestProfileManager(unittest.TestCase): """Test ProfileManager activation, deactivation, and switching.""" def setUp(self): self.config_data = { "mqtt": {"host": "mqtt"}, "profiles": { "armed": {"friendly_name": "Armed"}, "disarmed": {"friendly_name": "Disarmed"}, }, "cameras": { "front": { "ffmpeg": { "inputs": [ { "path": "rtsp://10.0.0.1:554/video", "roles": ["detect"], } ] }, "detect": {"height": 1080, "width": 1920, "fps": 5}, "notifications": {"enabled": False}, "objects": {"track": ["person"]}, "profiles": { "armed": { "notifications": {"enabled": True}, "objects": {"track": ["person", "car", "package"]}, }, "disarmed": { "notifications": {"enabled": False}, "objects": {"track": ["package"]}, }, }, }, "back": { "ffmpeg": { "inputs": [ { "path": "rtsp://10.0.0.2:554/video", "roles": ["detect"], } ] }, "detect": {"height": 1080, "width": 1920, "fps": 5}, "profiles": { "armed": { "notifications": {"enabled": True}, }, }, }, }, } if not os.path.exists(MODEL_CACHE_DIR) and not os.path.islink(MODEL_CACHE_DIR): os.makedirs(MODEL_CACHE_DIR) self.config = FrigateConfig(**self.config_data) self.mock_updater = MagicMock() self.manager = ProfileManager(self.config, self.mock_updater) def test_get_available_profiles(self): """Available profiles come from top-level profile definitions.""" profiles = self.manager.get_available_profiles() assert len(profiles) == 2 names = [p["name"] for p in profiles] assert "armed" in names assert "disarmed" in names # Verify friendly_name is included armed = next(p for p in profiles if p["name"] == "armed") assert armed["friendly_name"] == "Armed" def test_activate_invalid_profile(self): """Activating non-existent profile returns error.""" err = self.manager.activate_profile("nonexistent") assert err is not None assert "not defined" in err @patch.object(ProfileManager, "_persist_active_profile") def test_activate_profile(self, mock_persist): """Activating a profile applies overrides.""" err = self.manager.activate_profile("armed") assert err is None assert self.config.active_profile == "armed" # Front camera should have armed overrides front = self.config.cameras["front"] assert front.notifications.enabled is True assert front.objects.track == ["person", "car", "package"] # Back camera should have armed overrides back = self.config.cameras["back"] assert back.notifications.enabled is True @patch.object(ProfileManager, "_persist_active_profile") def test_deactivate_profile(self, mock_persist): """Deactivating a profile restores base config.""" # Activate first self.manager.activate_profile("armed") assert self.config.cameras["front"].notifications.enabled is True # Deactivate err = self.manager.activate_profile(None) assert err is None assert self.config.active_profile is None # Should be back to base front = self.config.cameras["front"] assert front.notifications.enabled is False assert front.objects.track == ["person"] @patch.object(ProfileManager, "_persist_active_profile") def test_switch_profiles(self, mock_persist): """Switching from one profile to another works.""" self.manager.activate_profile("armed") assert self.config.cameras["front"].objects.track == [ "person", "car", "package", ] self.manager.activate_profile("disarmed") assert self.config.active_profile == "disarmed" assert self.config.cameras["front"].objects.track == ["package"] assert self.config.cameras["front"].notifications.enabled is False @patch.object(ProfileManager, "_persist_active_profile") def test_unaffected_camera(self, mock_persist): """Camera without the activated profile is unaffected.""" back_base_notifications = self.config.cameras["back"].notifications.enabled self.manager.activate_profile("disarmed") # Back camera has no "disarmed" profile, should be unchanged assert ( self.config.cameras["back"].notifications.enabled == back_base_notifications ) @patch.object(ProfileManager, "_persist_active_profile") def test_activate_profile_disables_camera(self, mock_persist): """Profile with enabled=false disables the camera.""" self.config.profiles["away"] = ProfileDefinitionConfig(friendly_name="Away") self.config.cameras["front"].profiles["away"] = CameraProfileConfig( enabled=False ) self.manager = ProfileManager(self.config, self.mock_updater) assert self.config.cameras["front"].enabled is True err = self.manager.activate_profile("away") assert err is None assert self.config.cameras["front"].enabled is False @patch.object(ProfileManager, "_persist_active_profile") def test_deactivate_restores_enabled(self, mock_persist): """Deactivating a profile restores the camera's base enabled state.""" self.config.profiles["away"] = ProfileDefinitionConfig(friendly_name="Away") self.config.cameras["front"].profiles["away"] = CameraProfileConfig( enabled=False ) self.manager = ProfileManager(self.config, self.mock_updater) self.manager.activate_profile("away") assert self.config.cameras["front"].enabled is False self.manager.activate_profile(None) assert self.config.cameras["front"].enabled is True @patch.object(ProfileManager, "_persist_active_profile") def test_activate_profile_adds_zone(self, mock_persist): """Profile with zones adds/overrides zones on camera.""" from frigate.config.camera.zone import ZoneConfig self.config.profiles["away"] = ProfileDefinitionConfig(friendly_name="Away") self.config.cameras["front"].profiles["away"] = CameraProfileConfig( zones={ "driveway": ZoneConfig( coordinates="0.1,0.1,0.9,0.1,0.9,0.9,0.1,0.9", objects=["car"], ) } ) self.manager = ProfileManager(self.config, self.mock_updater) assert "driveway" not in self.config.cameras["front"].zones err = self.manager.activate_profile("away") assert err is None assert "driveway" in self.config.cameras["front"].zones @patch.object(ProfileManager, "_persist_active_profile") def test_deactivate_restores_zones(self, mock_persist): """Deactivating a profile restores base zones.""" from frigate.config.camera.zone import ZoneConfig self.config.profiles["away"] = ProfileDefinitionConfig(friendly_name="Away") self.config.cameras["front"].profiles["away"] = CameraProfileConfig( zones={ "driveway": ZoneConfig( coordinates="0.1,0.1,0.9,0.1,0.9,0.9,0.1,0.9", objects=["car"], ) } ) self.manager = ProfileManager(self.config, self.mock_updater) self.manager.activate_profile("away") assert "driveway" in self.config.cameras["front"].zones self.manager.activate_profile(None) assert "driveway" not in self.config.cameras["front"].zones @patch.object(ProfileManager, "_persist_active_profile") def test_zones_zmq_published(self, mock_persist): """ZMQ update is published for zones change.""" from frigate.config.camera.updater import ( CameraConfigUpdateEnum, CameraConfigUpdateTopic, ) from frigate.config.camera.zone import ZoneConfig self.config.profiles["away"] = ProfileDefinitionConfig(friendly_name="Away") self.config.cameras["front"].profiles["away"] = CameraProfileConfig( zones={ "driveway": ZoneConfig( coordinates="0.1,0.1,0.9,0.1,0.9,0.9,0.1,0.9", objects=["car"], ) } ) self.manager = ProfileManager(self.config, self.mock_updater) self.mock_updater.reset_mock() self.manager.activate_profile("away") zones_calls = [ call for call in self.mock_updater.publish_update.call_args_list if call[0][0] == CameraConfigUpdateTopic(CameraConfigUpdateEnum.zones, "front") ] assert len(zones_calls) == 1 @patch.object(ProfileManager, "_persist_active_profile") def test_enabled_zmq_published(self, mock_persist): """ZMQ update is published for enabled state change.""" from frigate.config.camera.updater import ( CameraConfigUpdateEnum, CameraConfigUpdateTopic, ) self.config.profiles["away"] = ProfileDefinitionConfig(friendly_name="Away") self.config.cameras["front"].profiles["away"] = CameraProfileConfig( enabled=False ) self.manager = ProfileManager(self.config, self.mock_updater) self.mock_updater.reset_mock() self.manager.activate_profile("away") # Find the enabled update call enabled_calls = [ call for call in self.mock_updater.publish_update.call_args_list if call[0][0] == CameraConfigUpdateTopic(CameraConfigUpdateEnum.enabled, "front") ] assert len(enabled_calls) == 1 assert enabled_calls[0][0][1] is False @patch.object(ProfileManager, "_persist_active_profile") def test_zmq_updates_published(self, mock_persist): """ZMQ updates are published when a profile is activated.""" self.manager.activate_profile("armed") assert self.mock_updater.publish_update.called def test_get_profile_info(self): """Profile info returns correct structure with friendly names.""" with patch.object( ProfileManager, "_load_persisted_data", return_value={"active": None, "last_activated": {}}, ): info = self.manager.get_profile_info() assert "profiles" in info assert "active_profile" in info assert "last_activated" in info assert info["active_profile"] is None assert info["last_activated"] == {} names = [p["name"] for p in info["profiles"]] assert "armed" in names assert "disarmed" in names @patch.object(ProfileManager, "_persist_active_profile") def test_base_configs_for_api_unchanged_after_activation(self, mock_persist): """API base configs reflect pre-profile values after activation.""" base_track = self.config.cameras["front"].objects.track[:] assert base_track == ["person"] self.manager.activate_profile("armed") # In-memory config has the profile-merged values assert self.config.cameras["front"].objects.track == [ "person", "car", "package", ] # But the API base configs still return the original base values api_base = self.manager.get_base_configs_for_api("front") assert "objects" in api_base assert api_base["objects"]["track"] == ["person"] def test_base_configs_for_api_are_json_serializable(self): """API base configs are JSON-serializable (mode='json').""" import json api_base = self.manager.get_base_configs_for_api("front") # 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_profile_change_republishes_switch_states(self, mock_persist): """Profile changes republish MQTT switch states so HA stays in sync. Regression: activating/deactivating a profile updated the in-memory config (and Frigate's behavior) but left the retained MQTT state topics stale, so external integrations like Home Assistant kept showing the pre-profile toggle position. """ config_data = copy.deepcopy(self.config_data) config_data["cameras"]["front"]["profiles"]["disarmed"]["review"] = { "alerts": {"enabled": False}, } config = FrigateConfig(**config_data) dispatcher = MagicMock() manager = ProfileManager(config, self.mock_updater, dispatcher) # Activating disarmed turns alerts off -> MQTT state must follow manager.activate_profile("disarmed") dispatcher.publish.assert_any_call( "front/review_alerts/state", "OFF", retain=True ) # Deactivating restores the base (alerts on) -> MQTT state must follow dispatcher.publish.reset_mock() manager.activate_profile(None) dispatcher.publish.assert_any_call( "front/review_alerts/state", "ON", retain=True ) @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.""" def test_persist_and_load(self): """Active profile name can be persisted and loaded via JSON.""" data = {"active": "armed", "last_activated": {"armed": 1700000000.0}} with patch.object( ProfileManager, "_load_persisted_data", return_value=data, ): result = ProfileManager.load_persisted_profile() assert result == "armed" def test_load_empty_file(self): """Empty persistence file returns None.""" with patch.object(type(PERSISTENCE_FILE), "exists", return_value=True): with patch.object(type(PERSISTENCE_FILE), "read_text", return_value=""): result = ProfileManager.load_persisted_profile() assert result is None def test_load_missing_file(self): """Missing persistence file returns None.""" with patch.object(type(PERSISTENCE_FILE), "exists", return_value=False): result = ProfileManager.load_persisted_profile() assert result is None def test_load_persisted_data_valid_json(self): """Valid JSON file is loaded correctly.""" data = {"active": "home", "last_activated": {"home": 1700000000.0}} with patch.object(type(PERSISTENCE_FILE), "exists", return_value=True): with patch.object( type(PERSISTENCE_FILE), "read_text", return_value=json.dumps(data), ): result = ProfileManager._load_persisted_data() assert result == data def test_load_persisted_data_invalid_json(self): """Invalid JSON returns default structure.""" with patch.object(type(PERSISTENCE_FILE), "exists", return_value=True): with patch.object( type(PERSISTENCE_FILE), "read_text", return_value="not json" ): result = ProfileManager._load_persisted_data() assert result == {"active": None, "last_activated": {}} def test_load_persisted_data_missing_file(self): """Missing file returns default structure.""" with patch.object(type(PERSISTENCE_FILE), "exists", return_value=False): result = ProfileManager._load_persisted_data() assert result == {"active": None, "last_activated": {}} def test_persist_records_timestamp(self): """Persisting a profile records the activation timestamp.""" config_data = { "mqtt": {"host": "mqtt"}, "profiles": {"armed": {"friendly_name": "Armed"}}, "cameras": { "front": { "ffmpeg": { "inputs": [ { "path": "rtsp://10.0.0.1:554/video", "roles": ["detect"], } ] }, "detect": {"height": 1080, "width": 1920, "fps": 5}, "profiles": {"armed": {"detect": {"enabled": True}}}, }, }, } if not os.path.exists(MODEL_CACHE_DIR) and not os.path.islink(MODEL_CACHE_DIR): os.makedirs(MODEL_CACHE_DIR) config = FrigateConfig(**config_data) manager = ProfileManager(config, MagicMock()) written_data = {} def mock_write(_self, content): written_data.update(json.loads(content)) with patch.object( ProfileManager, "_load_persisted_data", return_value={"active": None, "last_activated": {}}, ): with patch.object(type(PERSISTENCE_FILE), "write_text", mock_write): manager._persist_active_profile("armed") assert written_data["active"] == "armed" assert "armed" in written_data["last_activated"] assert isinstance(written_data["last_activated"]["armed"], float) def test_persist_deactivate_keeps_timestamps(self): """Deactivating sets active to None but preserves last_activated.""" existing = { "active": "armed", "last_activated": {"armed": 1700000000.0}, } written_data = {} def mock_write(_self, content): written_data.update(json.loads(content)) config_data = { "mqtt": {"host": "mqtt"}, "profiles": {"armed": {"friendly_name": "Armed"}}, "cameras": { "front": { "ffmpeg": { "inputs": [ { "path": "rtsp://10.0.0.1:554/video", "roles": ["detect"], } ] }, "detect": {"height": 1080, "width": 1920, "fps": 5}, "profiles": {"armed": {"detect": {"enabled": True}}}, }, }, } if not os.path.exists(MODEL_CACHE_DIR) and not os.path.islink(MODEL_CACHE_DIR): os.makedirs(MODEL_CACHE_DIR) config = FrigateConfig(**config_data) manager = ProfileManager(config, MagicMock()) with patch.object( ProfileManager, "_load_persisted_data", return_value=existing ): with patch.object(type(PERSISTENCE_FILE), "write_text", mock_write): manager._persist_active_profile(None) assert written_data["active"] is None assert written_data["last_activated"]["armed"] == 1700000000.0 if __name__ == "__main__": unittest.main()