frigate/frigate/test/test_debug_replay.py

"""Tests for the simplified DebugReplayManager.

Startup orchestration lives in ``frigate.jobs.debug_replay`` (covered by
``test_debug_replay_job``). The manager owns only session presence and
cleanup.
"""

import unittest
import unittest.mock
from unittest.mock import MagicMock, patch


class TestDebugReplayManagerSession(unittest.TestCase):
    def test_inactive_by_default(self) -> None:
        from frigate.debug_replay import DebugReplayManager

        manager = DebugReplayManager()

        self.assertFalse(manager.active)
        self.assertIsNone(manager.replay_camera_name)
        self.assertIsNone(manager.source_camera)
        self.assertIsNone(manager.clip_path)
        self.assertIsNone(manager.start_ts)
        self.assertIsNone(manager.end_ts)

    def test_mark_starting_sets_session_pointers_and_active(self) -> None:
        from frigate.debug_replay import DebugReplayManager

        manager = DebugReplayManager()

        manager.mark_starting(
            source_camera="front",
            replay_camera_name="_replay_front",
            start_ts=100.0,
            end_ts=200.0,
        )

        self.assertTrue(manager.active)
        self.assertEqual(manager.replay_camera_name, "_replay_front")
        self.assertEqual(manager.source_camera, "front")
        self.assertEqual(manager.start_ts, 100.0)
        self.assertEqual(manager.end_ts, 200.0)
        self.assertIsNone(manager.clip_path)

    def test_mark_session_ready_sets_clip_path(self) -> None:
        from frigate.debug_replay import DebugReplayManager

        manager = DebugReplayManager()
        manager.mark_starting("front", "_replay_front", 100.0, 200.0)

        manager.mark_session_ready(clip_path="/tmp/replay/_replay_front.mp4")

        self.assertEqual(manager.clip_path, "/tmp/replay/_replay_front.mp4")
        self.assertTrue(manager.active)

    def test_clear_session_resets_all_pointers(self) -> None:
        from frigate.debug_replay import DebugReplayManager

        manager = DebugReplayManager()
        manager.mark_starting("front", "_replay_front", 100.0, 200.0)
        manager.mark_session_ready("/tmp/replay/clip.mp4")

        manager.clear_session()

        self.assertFalse(manager.active)
        self.assertIsNone(manager.replay_camera_name)
        self.assertIsNone(manager.source_camera)
        self.assertIsNone(manager.clip_path)
        self.assertIsNone(manager.start_ts)
        self.assertIsNone(manager.end_ts)


class TestDebugReplayManagerStop(unittest.TestCase):
    def setUp(self) -> None:
        # stop() publishes a terminal job_state via a real JobStatePublisher,
        # which opens a ZMQ REQ socket and blocks on REP. No dispatcher runs
        # in unit tests, so substitute a no-op publisher.
        patcher = patch("frigate.debug_replay.JobStatePublisher")
        patcher.start()
        self.addCleanup(patcher.stop)

    def test_stop_when_inactive_is_a_noop(self) -> None:
        from frigate.debug_replay import DebugReplayManager

        manager = DebugReplayManager()
        frigate_config = MagicMock()
        frigate_config.cameras = {}
        publisher = MagicMock()

        # Should not raise; should not publish any events.
        manager.stop(frigate_config=frigate_config, config_publisher=publisher)

        publisher.publish_update.assert_not_called()

    def test_stop_publishes_remove_when_camera_was_published(self) -> None:
        from frigate.config.camera.updater import CameraConfigUpdateEnum
        from frigate.debug_replay import DebugReplayManager

        manager = DebugReplayManager()
        manager.mark_starting("front", "_replay_front", 100.0, 200.0)
        manager.mark_session_ready("/tmp/replay/_replay_front.mp4")

        camera_config = MagicMock()
        frigate_config = MagicMock()
        frigate_config.cameras = {"_replay_front": camera_config}
        publisher = MagicMock()

        with (
            patch.object(manager, "_cleanup_db"),
            patch.object(manager, "_cleanup_files"),
            patch("frigate.debug_replay.cancel_debug_replay_job", return_value=False),
        ):
            manager.stop(frigate_config=frigate_config, config_publisher=publisher)

        # One publish_update call with a remove topic.
        self.assertEqual(publisher.publish_update.call_count, 1)
        topic_arg = publisher.publish_update.call_args.args[0]
        self.assertEqual(topic_arg.update_type, CameraConfigUpdateEnum.remove)
        self.assertFalse(manager.active)

    def test_stop_skips_remove_publish_when_camera_not_in_config(self) -> None:
        """Cancellation during preparing_clip: no camera was published yet."""
        from frigate.debug_replay import DebugReplayManager

        manager = DebugReplayManager()
        manager.mark_starting("front", "_replay_front", 100.0, 200.0)
        # clip_path stays None because we cancelled before camera publish.

        frigate_config = MagicMock()
        frigate_config.cameras = {}  # _replay_front not present
        publisher = MagicMock()

        with (
            patch.object(manager, "_cleanup_db"),
            patch.object(manager, "_cleanup_files"),
            patch("frigate.debug_replay.cancel_debug_replay_job", return_value=True),
        ):
            manager.stop(frigate_config=frigate_config, config_publisher=publisher)

        publisher.publish_update.assert_not_called()
        self.assertFalse(manager.active)

    def test_stop_calls_cancel_debug_replay_job(self) -> None:
        from frigate.debug_replay import DebugReplayManager

        manager = DebugReplayManager()
        manager.mark_starting("front", "_replay_front", 100.0, 200.0)

        frigate_config = MagicMock()
        frigate_config.cameras = {}
        publisher = MagicMock()

        with (
            patch.object(manager, "_cleanup_db"),
            patch.object(manager, "_cleanup_files"),
            patch(
                "frigate.debug_replay.cancel_debug_replay_job",
                return_value=True,
            ) as mock_cancel,
        ):
            manager.stop(frigate_config=frigate_config, config_publisher=publisher)

        mock_cancel.assert_called_once()


class TestDebugReplayManagerPublishCamera(unittest.TestCase):
    def test_publish_camera_invokes_publisher_with_add_topic(self) -> None:
        from frigate.config.camera.updater import CameraConfigUpdateEnum
        from frigate.debug_replay import DebugReplayManager

        manager = DebugReplayManager()

        source_config = MagicMock()
        new_camera_config = MagicMock()
        frigate_config = MagicMock()
        frigate_config.cameras = {"front": source_config}
        publisher = MagicMock()

        with (
            patch.object(
                manager,
                "_build_camera_config_dict",
                return_value={"enabled": True},
            ),
            patch("frigate.debug_replay.find_config_file", return_value="/cfg.yml"),
            patch("frigate.debug_replay.YAML") as yaml_cls,
            patch("frigate.debug_replay.FrigateConfig.parse_object") as parse_object,
            patch("builtins.open", unittest.mock.mock_open(read_data="cameras:\n")),
        ):
            yaml_instance = yaml_cls.return_value
            yaml_instance.load.return_value = {"cameras": {}}
            parsed = MagicMock()
            parsed.cameras = {"_replay_front": new_camera_config}
            parse_object.return_value = parsed

            manager.publish_camera(
                source_camera="front",
                replay_name="_replay_front",
                clip_path="/tmp/clip.mp4",
                frigate_config=frigate_config,
                config_publisher=publisher,
            )

        # Camera registered into the live config dict
        self.assertIn("_replay_front", frigate_config.cameras)
        # Publisher invoked with an add topic
        self.assertEqual(publisher.publish_update.call_count, 1)
        topic_arg = publisher.publish_update.call_args.args[0]
        self.assertEqual(topic_arg.update_type, CameraConfigUpdateEnum.add)

    def test_publish_camera_wraps_parse_failure_in_runtime_error(self) -> None:
        from frigate.debug_replay import DebugReplayManager

        manager = DebugReplayManager()
        frigate_config = MagicMock()
        frigate_config.cameras = {"front": MagicMock()}
        publisher = MagicMock()

        with (
            patch.object(
                manager,
                "_build_camera_config_dict",
                return_value={"enabled": True},
            ),
            patch("frigate.debug_replay.find_config_file", return_value="/cfg.yml"),
            patch("frigate.debug_replay.YAML") as yaml_cls,
            patch(
                "frigate.debug_replay.FrigateConfig.parse_object",
                side_effect=ValueError("zone foo has invalid coordinates"),
            ),
            patch("builtins.open", unittest.mock.mock_open(read_data="cameras:\n")),
        ):
            yaml_cls.return_value.load.return_value = {"cameras": {}}

            with self.assertRaises(RuntimeError) as ctx:
                manager.publish_camera(
                    source_camera="front",
                    replay_name="_replay_front",
                    clip_path="/tmp/clip.mp4",
                    frigate_config=frigate_config,
                    config_publisher=publisher,
                )

        self.assertIn("replay camera config", str(ctx.exception))
        self.assertIn("invalid coordinates", str(ctx.exception))
        publisher.publish_update.assert_not_called()


if __name__ == "__main__":
    unittest.main()