frigate/frigate/test/test_camera_lifecycle.py

"""Regression tests for runtime camera add and delete handling."""

import asyncio
import threading
import unittest
from types import SimpleNamespace
from unittest.mock import MagicMock

# LicensePlatePostProcessor is imported via the maintainer rather than from
# data_processing.post.license_plate, which circularly imports back through
# frigate.embeddings before that package finishes initializing
from frigate.embeddings.maintainer import (
    EmbeddingMaintainer,
    LicensePlatePostProcessor,
)
from frigate.ptz.autotrack import PtzAutoTracker
from frigate.review.maintainer import ReviewSegmentMaintainer
from frigate.track.object_processing import TrackedObjectProcessor


def _make_processor() -> TrackedObjectProcessor:
    """Build a processor with no cameras, bypassing __init__."""
    processor = TrackedObjectProcessor.__new__(TrackedObjectProcessor)
    processor.camera_states = {}
    processor.camera_states_lock = threading.Lock()
    processor.config = SimpleNamespace(cameras={})
    processor.event_sender = MagicMock()
    processor.detection_publisher = MagicMock()
    processor.ongoing_manual_events = {}
    return processor


class TestObjectProcessorUnknownCamera(unittest.TestCase):
    def test_save_lpr_snapshot_ignores_unknown_camera(self):
        processor = _make_processor()

        # 1x1 png, base64; decoding must not be what fails
        payload = (
            "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8z8BQDwAEhQGAhKmMIQAAAABJRU5ErkJggg==",
            "1234.5-abcdef",
            "deleted_cam",
        )

        processor.save_lpr_snapshot(payload)

        processor.event_sender.publish.assert_not_called()

    def test_create_manual_event_ignores_unknown_camera(self):
        processor = _make_processor()

        payload = (
            1234.5,
            "deleted_cam",
            "person",
            "1234.5-abcdef",
            True,
            0.9,
            None,
            None,
            "api",
            False,
            None,
        )

        processor.create_manual_event(payload)

        processor.event_sender.publish.assert_not_called()
        self.assertEqual(processor.ongoing_manual_events, {})

    def test_create_lpr_event_ignores_unknown_camera(self):
        processor = _make_processor()

        payload = (
            1234.5,
            "deleted_cam",
            "license_plate",
            "1234.5-abcdef",
            True,
            0.9,
            None,
            "ABC123",
        )

        processor.create_lpr_event(payload)

        processor.event_sender.publish.assert_not_called()
        self.assertEqual(processor.ongoing_manual_events, {})

    def test_create_manual_event_ignores_camera_added_but_not_yet_drained(self):
        """The add window: present in config.cameras, absent from camera_states.

        debug_replay writes the camera into the shared config before publishing
        add, so a guard on config.cameras passes here and falls through to
        camera_states. This test fails against such a guard.
        """
        processor = _make_processor()
        processor.config = SimpleNamespace(
            cameras={
                "new_cam": SimpleNamespace(
                    record=SimpleNamespace(event_pre_capture=5, enabled=True)
                )
            }
        )

        payload = (
            1234.5,
            "new_cam",
            "person",
            "1234.5-abcdef",
            True,
            0.9,
            None,
            None,
            "api",
            False,
            None,
        )

        processor.create_manual_event(payload)

        processor.event_sender.publish.assert_not_called()


class TestEmbeddingsUnknownCamera(unittest.TestCase):
    def _make_maintainer(self) -> EmbeddingMaintainer:
        maintainer = EmbeddingMaintainer.__new__(EmbeddingMaintainer)
        maintainer.config = SimpleNamespace(cameras={})
        maintainer.event_end_subscriber = MagicMock()
        maintainer.realtime_processors = [MagicMock()]
        # spec is required: the dispatch loop is a chain of isinstance checks,
        # and a bare MagicMock matches none of them, so the crashing branch
        # would never run and the test would pass against unfixed code
        maintainer.post_processors = [MagicMock(spec=LicensePlatePostProcessor)]
        maintainer.detected_license_plates = {"1234.5-abcdef": {"obj_data": {}}}
        maintainer.recordings_available_through = {"deleted_cam": 1234.5}
        maintainer.event_metadata_publisher = MagicMock()
        return maintainer

    def test_process_finalized_skips_unknown_camera(self):
        maintainer = self._make_maintainer()
        # updated_db=False bypasses the Event.get branch, which would hit the
        # database and mask the KeyError this test is about
        maintainer.event_end_subscriber.check_for_update.side_effect = [
            ("1234.5-abcdef", "deleted_cam", False),
            None,
        ]

        maintainer._process_finalized()

        maintainer.post_processors[0].process_data.assert_not_called()

    def test_process_finalized_still_expires_realtime_state(self):
        """The guard must not skip per-event cleanup, only post processing."""
        maintainer = self._make_maintainer()
        maintainer.event_end_subscriber.check_for_update.side_effect = [
            ("1234.5-abcdef", "deleted_cam", False),
            None,
        ]

        maintainer._process_finalized()

        maintainer.realtime_processors[0].expire_object.assert_called_once_with(
            "1234.5-abcdef", "deleted_cam"
        )

    def test_expire_dedicated_lpr_drops_entry_for_unknown_camera(self):
        maintainer = self._make_maintainer()
        maintainer.detected_license_plates = {
            "1234.5-abcdef": {"camera": "deleted_cam", "last_seen": 1.0}
        }

        maintainer._expire_dedicated_lpr()

        self.assertEqual(maintainer.detected_license_plates, {})


class TestReviewMaintainerRemoval(unittest.TestCase):
    def test_camera_removal_ends_segment_and_clears_state(self):
        maintainer = ReviewSegmentMaintainer.__new__(ReviewSegmentMaintainer)
        maintainer.active_review_segments = {"deleted_cam": MagicMock()}
        maintainer.indefinite_events = {"deleted_cam": {"1234.5-abcdef": 1.0}}
        maintainer.recent_classification_state_changes = {
            "deleted_cam": [{"model": "gate", "from": "a", "to": "b", "timestamp": 1.0}]
        }
        maintainer.forcibly_end_segment = MagicMock()
        maintainer.frame_manager = MagicMock()

        maintainer._handle_camera_removed("deleted_cam")

        maintainer.forcibly_end_segment.assert_called_once_with("deleted_cam")
        maintainer.frame_manager.close_camera.assert_called_once_with("deleted_cam")
        self.assertNotIn("deleted_cam", maintainer.indefinite_events)
        self.assertNotIn("deleted_cam", maintainer.recent_classification_state_changes)


class TestAutotrackerMoveQueue(unittest.TestCase):
    def test_move_queue_drops_move_for_removed_camera(self):
        tracker = PtzAutoTracker.__new__(PtzAutoTracker)
        tracker.stop_event = MagicMock()
        # one pass through the loop, then stop
        tracker.stop_event.is_set.side_effect = [False, True]
        tracker.ptz_metrics = {}
        tracker.move_queues = {"deleted_cam": asyncio.Queue()}
        tracker.move_queue_locks = {"deleted_cam": asyncio.Lock()}
        tracker.onvif = MagicMock()
        tracker.config = SimpleNamespace(cameras={})
        tracker.move_queues["deleted_cam"].put_nowait((1234.5, 0.1, 0.1, 0.0))

        asyncio.run(tracker._process_move_queue("deleted_cam"))

        tracker.onvif._move_relative.assert_not_called()


class TestCameraStateAccessors(unittest.TestCase):
    def test_get_camera_state_returns_none_for_unknown_camera(self):
        processor = _make_processor()

        self.assertIsNone(processor.get_camera_state("deleted_cam"))

    def test_get_camera_states_returns_a_snapshot_not_a_view(self):
        """A live values() view raises RuntimeError if the writer pops mid-iteration."""
        processor = _make_processor()
        processor.camera_states = {"one": MagicMock(), "two": MagicMock()}

        states = processor.get_camera_states()
        processor.camera_states.pop("one")

        self.assertEqual(len(states), 2)

    def test_get_current_frame_time_is_zero_for_unknown_camera(self):
        processor = _make_processor()

        self.assertEqual(processor.get_current_frame_time("deleted_cam"), 0.0)