blakeblackshear.frigate/frigate/test/test_profiles.py

"""Tests for the profiles system."""

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)


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)


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()