frigate/frigate/test/http_api/test_http_media.py

"""Unit tests for recordings/media API endpoints."""

from dataclasses import dataclass
from datetime import UTC, datetime
from unittest.mock import patch

import pytz
from fastapi import Request

from frigate.api.auth import get_allowed_cameras_for_filter, get_current_user
from frigate.const import MAX_SEGMENT_DURATION
from frigate.models import Recordings
from frigate.test.http_api.base_http_test import AuthTestClient, BaseTestHttp


@dataclass(frozen=True)
class RangeCase:
    """Expected behavior for one segment relative to the requested range.

    Offsets are seconds from REQUEST_START; the request ends at +100 seconds.
    """

    name: str
    start_offset: float
    end_offset: float
    included_in_recordings: bool
    vod_clip_from_ms: int | None = None
    vod_duration_ms: int | None = None


REQUEST_START = 1000
REQUEST_END = 1100
RANGE_CASES = (
    RangeCase("before", -MAX_SEGMENT_DURATION + 1, -1, False),
    RangeCase("meets_start", -10, 0, True),
    RangeCase(
        "overlaps_start",
        -MAX_SEGMENT_DURATION + 0.5,
        0.25,
        True,
        vod_clip_from_ms=599500,
        vod_duration_ms=250,
    ),
    RangeCase("starts_at_start", 0, 10, True, vod_duration_ms=10000),
    RangeCase("inside", 20, 80, True, vod_duration_ms=60000),
    RangeCase("ends_at_end", 90, 100, True, vod_duration_ms=10000),
    RangeCase("matches_range", 0, 100, True, vod_duration_ms=100000),
    RangeCase("starts_with_range", 0, 110, True, vod_duration_ms=100000),
    RangeCase(
        "covers_range",
        -20,
        120,
        True,
        vod_clip_from_ms=20000,
        vod_duration_ms=100000,
    ),
    RangeCase(
        "ends_with_range",
        -10,
        100,
        True,
        vod_clip_from_ms=10000,
        vod_duration_ms=100000,
    ),
    RangeCase("overlaps_end", 95, 105, True, vod_duration_ms=5000),
    RangeCase("starts_at_end", 100, 110, True),
    RangeCase("after", 101, 110, False),
)


class TestHttpMedia(BaseTestHttp):
    """Test media API endpoints, particularly recordings with DST handling."""

    def setUp(self):
        """Set up test fixtures."""
        super().setUp([Recordings])
        self.app = super().create_app()

        # Mock get_current_user for all tests
        async def mock_get_current_user(request: Request):
            username = request.headers.get("remote-user")
            role = request.headers.get("remote-role")
            if not username or not role:
                from fastapi.responses import JSONResponse

                return JSONResponse(
                    content={"message": "No authorization headers."}, status_code=401
                )
            return {"username": username, "role": role}

        self.app.dependency_overrides[get_current_user] = mock_get_current_user

        async def mock_get_allowed_cameras_for_filter(request: Request):
            return ["front_door"]

        self.app.dependency_overrides[get_allowed_cameras_for_filter] = (
            mock_get_allowed_cameras_for_filter
        )

    def tearDown(self):
        """Clean up after tests."""
        self.app.dependency_overrides.clear()
        super().tearDown()

    def _assert_vod_response(
        self,
        response,
        expected_clips: list[tuple[str, int | None, int]],
    ) -> None:
        """Assert VOD clip metadata and its derived duration fields."""
        assert response.status_code == 200
        vod = response.json()
        assert [
            (
                clip["path"],
                clip.get("clipFrom"),
                clip["keyFrameDurations"][0],
            )
            for clip in vod["sequences"][0]["clips"]
        ] == expected_clips
        expected_durations = [clip[2] for clip in expected_clips]
        assert vod["durations"] == expected_durations
        assert vod["segment_duration"] == max(expected_durations)

    def test_recordings_summary_across_dst_spring_forward(self):
        """
        Test recordings summary across spring DST transition (spring forward).

        In 2024, DST in America/New_York transitions on March 10, 2024 at 2:00 AM
        Clocks spring forward from 2:00 AM to 3:00 AM (EST to EDT)
        """
        tz = pytz.timezone("America/New_York")

        # March 9, 2024 at 12:00 PM EST (before DST)
        march_9_noon = tz.localize(datetime(2024, 3, 9, 12, 0, 0)).timestamp()

        # March 10, 2024 at 12:00 PM EDT (after DST transition)
        march_10_noon = tz.localize(datetime(2024, 3, 10, 12, 0, 0)).timestamp()

        # March 11, 2024 at 12:00 PM EDT (after DST)
        march_11_noon = tz.localize(datetime(2024, 3, 11, 12, 0, 0)).timestamp()

        with AuthTestClient(self.app) as client:
            # Insert recordings for each day
            Recordings.insert(
                id="recording_march_9",
                path="/media/recordings/march_9.mp4",
                camera="front_door",
                start_time=march_9_noon,
                end_time=march_9_noon + 3600,  # 1 hour recording
                duration=3600,
                motion=100,
                objects=5,
            ).execute()

            Recordings.insert(
                id="recording_march_10",
                path="/media/recordings/march_10.mp4",
                camera="front_door",
                start_time=march_10_noon,
                end_time=march_10_noon + 3600,
                duration=3600,
                motion=150,
                objects=8,
            ).execute()

            Recordings.insert(
                id="recording_march_11",
                path="/media/recordings/march_11.mp4",
                camera="front_door",
                start_time=march_11_noon,
                end_time=march_11_noon + 3600,
                duration=3600,
                motion=200,
                objects=10,
            ).execute()

            # Test recordings summary with America/New_York timezone
            response = client.get(
                "/recordings/summary",
                params={"timezone": "America/New_York", "cameras": "all"},
            )

            assert response.status_code == 200
            summary = response.json()

            # Verify we get exactly 3 days
            assert len(summary) == 3, f"Expected 3 days, got {len(summary)}"

            # Verify the correct dates are returned (API returns dict with True values)
            assert "2024-03-09" in summary, f"Expected 2024-03-09 in {summary}"
            assert "2024-03-10" in summary, f"Expected 2024-03-10 in {summary}"
            assert "2024-03-11" in summary, f"Expected 2024-03-11 in {summary}"
            assert summary["2024-03-09"] is True
            assert summary["2024-03-10"] is True
            assert summary["2024-03-11"] is True

    def test_recordings_summary_across_dst_fall_back(self):
        """
        Test recordings summary across fall DST transition (fall back).

        In 2024, DST in America/New_York transitions on November 3, 2024 at 2:00 AM
        Clocks fall back from 2:00 AM to 1:00 AM (EDT to EST)
        """
        tz = pytz.timezone("America/New_York")

        # November 2, 2024 at 12:00 PM EDT (before DST transition)
        nov_2_noon = tz.localize(datetime(2024, 11, 2, 12, 0, 0)).timestamp()

        # November 3, 2024 at 12:00 PM EST (after DST transition)
        # Need to specify is_dst=False to get the time after fall back
        nov_3_noon = tz.localize(
            datetime(2024, 11, 3, 12, 0, 0), is_dst=False
        ).timestamp()

        # November 4, 2024 at 12:00 PM EST (after DST)
        nov_4_noon = tz.localize(datetime(2024, 11, 4, 12, 0, 0)).timestamp()

        with AuthTestClient(self.app) as client:
            # Insert recordings for each day
            Recordings.insert(
                id="recording_nov_2",
                path="/media/recordings/nov_2.mp4",
                camera="front_door",
                start_time=nov_2_noon,
                end_time=nov_2_noon + 3600,
                duration=3600,
                motion=100,
                objects=5,
            ).execute()

            Recordings.insert(
                id="recording_nov_3",
                path="/media/recordings/nov_3.mp4",
                camera="front_door",
                start_time=nov_3_noon,
                end_time=nov_3_noon + 3600,
                duration=3600,
                motion=150,
                objects=8,
            ).execute()

            Recordings.insert(
                id="recording_nov_4",
                path="/media/recordings/nov_4.mp4",
                camera="front_door",
                start_time=nov_4_noon,
                end_time=nov_4_noon + 3600,
                duration=3600,
                motion=200,
                objects=10,
            ).execute()

            # Test recordings summary with America/New_York timezone
            response = client.get(
                "/recordings/summary",
                params={"timezone": "America/New_York", "cameras": "all"},
            )

            assert response.status_code == 200
            summary = response.json()

            # Verify we get exactly 3 days
            assert len(summary) == 3, f"Expected 3 days, got {len(summary)}"

            # Verify the correct dates are returned (API returns dict with True values)
            assert "2024-11-02" in summary, f"Expected 2024-11-02 in {summary}"
            assert "2024-11-03" in summary, f"Expected 2024-11-03 in {summary}"
            assert "2024-11-04" in summary, f"Expected 2024-11-04 in {summary}"
            assert summary["2024-11-02"] is True
            assert summary["2024-11-03"] is True
            assert summary["2024-11-04"] is True

    def test_recordings_summary_multiple_cameras_across_dst(self):
        """
        Test recordings summary with multiple cameras across DST boundary.
        """
        tz = pytz.timezone("America/New_York")

        # March 9, 2024 at 10:00 AM EST (before DST)
        march_9_morning = tz.localize(datetime(2024, 3, 9, 10, 0, 0)).timestamp()

        # March 10, 2024 at 3:00 PM EDT (after DST transition)
        march_10_afternoon = tz.localize(datetime(2024, 3, 10, 15, 0, 0)).timestamp()

        with AuthTestClient(self.app) as client:
            # Override allowed cameras for this test to include both
            async def mock_get_allowed_cameras_for_filter(_request: Request):
                return ["front_door", "back_door"]

            self.app.dependency_overrides[get_allowed_cameras_for_filter] = (
                mock_get_allowed_cameras_for_filter
            )

            # Insert recordings for front_door on March 9
            Recordings.insert(
                id="front_march_9",
                path="/media/recordings/front_march_9.mp4",
                camera="front_door",
                start_time=march_9_morning,
                end_time=march_9_morning + 3600,
                duration=3600,
                motion=100,
                objects=5,
            ).execute()

            # Insert recordings for back_door on March 10
            Recordings.insert(
                id="back_march_10",
                path="/media/recordings/back_march_10.mp4",
                camera="back_door",
                start_time=march_10_afternoon,
                end_time=march_10_afternoon + 3600,
                duration=3600,
                motion=150,
                objects=8,
            ).execute()

            # Test with all cameras
            response = client.get(
                "/recordings/summary",
                params={"timezone": "America/New_York", "cameras": "all"},
            )

            assert response.status_code == 200
            summary = response.json()

            # Verify we get both days
            assert len(summary) == 2, f"Expected 2 days, got {len(summary)}"
            assert "2024-03-09" in summary
            assert "2024-03-10" in summary
            assert summary["2024-03-09"] is True
            assert summary["2024-03-10"] is True

            # Reset dependency override back to default single camera for other tests
            async def reset_allowed_cameras(_request: Request):
                return ["front_door"]

            self.app.dependency_overrides[get_allowed_cameras_for_filter] = (
                reset_allowed_cameras
            )

    def test_recordings_summary_at_dst_transition_time(self):
        """
        Test recordings that span the exact DST transition time.
        """
        tz = pytz.timezone("America/New_York")

        # March 10, 2024 at 1:00 AM EST (1 hour before DST transition)
        # At 2:00 AM, clocks jump to 3:00 AM
        before_transition = tz.localize(datetime(2024, 3, 10, 1, 0, 0)).timestamp()

        # Recording that spans the transition (1:00 AM to 3:30 AM EDT)
        # This is 1.5 hours of actual time but spans the "missing" hour
        after_transition = tz.localize(datetime(2024, 3, 10, 3, 30, 0)).timestamp()

        with AuthTestClient(self.app) as client:
            Recordings.insert(
                id="recording_during_transition",
                path="/media/recordings/transition.mp4",
                camera="front_door",
                start_time=before_transition,
                end_time=after_transition,
                duration=after_transition - before_transition,
                motion=100,
                objects=5,
            ).execute()

            response = client.get(
                "/recordings/summary",
                params={"timezone": "America/New_York", "cameras": "all"},
            )

            assert response.status_code == 200
            summary = response.json()

            # The recording should appear on March 10
            assert len(summary) == 1
            assert "2024-03-10" in summary
            assert summary["2024-03-10"] is True

    def test_recordings_summary_utc_timezone(self):
        """
        Test recordings summary with UTC timezone (no DST).
        """
        # Use UTC timestamps directly
        march_9_utc = datetime(2024, 3, 9, 17, 0, 0, tzinfo=UTC).timestamp()
        march_10_utc = datetime(2024, 3, 10, 17, 0, 0, tzinfo=UTC).timestamp()

        with AuthTestClient(self.app) as client:
            Recordings.insert(
                id="recording_march_9_utc",
                path="/media/recordings/march_9_utc.mp4",
                camera="front_door",
                start_time=march_9_utc,
                end_time=march_9_utc + 3600,
                duration=3600,
                motion=100,
                objects=5,
            ).execute()

            Recordings.insert(
                id="recording_march_10_utc",
                path="/media/recordings/march_10_utc.mp4",
                camera="front_door",
                start_time=march_10_utc,
                end_time=march_10_utc + 3600,
                duration=3600,
                motion=150,
                objects=8,
            ).execute()

            # Test with UTC timezone
            response = client.get(
                "/recordings/summary", params={"timezone": "utc", "cameras": "all"}
            )

            assert response.status_code == 200
            summary = response.json()

            # Verify we get both days
            assert len(summary) == 2
            assert "2024-03-09" in summary
            assert "2024-03-10" in summary
            assert summary["2024-03-09"] is True
            assert summary["2024-03-10"] is True

    def test_recordings_summary_no_recordings(self):
        """
        Test recordings summary when no recordings exist.
        """
        with AuthTestClient(self.app) as client:
            response = client.get(
                "/recordings/summary",
                params={"timezone": "America/New_York", "cameras": "all"},
            )

            assert response.status_code == 200
            summary = response.json()
            assert len(summary) == 0

    def test_recordings_summary_single_camera_filter(self):
        """
        Test recordings summary filtered to a single camera.
        """
        tz = pytz.timezone("America/New_York")
        march_10_noon = tz.localize(datetime(2024, 3, 10, 12, 0, 0)).timestamp()

        with AuthTestClient(self.app) as client:
            # Insert recordings for both cameras
            Recordings.insert(
                id="front_recording",
                path="/media/recordings/front.mp4",
                camera="front_door",
                start_time=march_10_noon,
                end_time=march_10_noon + 3600,
                duration=3600,
                motion=100,
                objects=5,
            ).execute()

            Recordings.insert(
                id="back_recording",
                path="/media/recordings/back.mp4",
                camera="back_door",
                start_time=march_10_noon,
                end_time=march_10_noon + 3600,
                duration=3600,
                motion=150,
                objects=8,
            ).execute()

            # Test with only front_door camera
            response = client.get(
                "/recordings/summary",
                params={"timezone": "America/New_York", "cameras": "front_door"},
            )

            assert response.status_code == 200
            summary = response.json()
            assert len(summary) == 1
            assert "2024-03-10" in summary
            assert summary["2024-03-10"] is True

    def test_recordings_handles_all_range_relations(self):
        """Recordings return every interval relation that touches the range."""
        with AuthTestClient(self.app) as client:
            for case in RANGE_CASES:
                with self.subTest(case=case.name):
                    Recordings.delete().execute()
                    super().insert_mock_recording(
                        case.name,
                        REQUEST_START + case.start_offset,
                        REQUEST_START + case.end_offset,
                    )

                    response = client.get(
                        "/front_door/recordings",
                        params={"after": REQUEST_START, "before": REQUEST_END},
                    )

                    assert response.status_code == 200
                    expected_ids = [case.name] if case.included_in_recordings else []
                    assert [
                        recording["id"] for recording in response.json()
                    ] == expected_ids

    def test_vod_handles_all_range_relations(self):
        """VOD clips every interval relation with positive playback duration."""
        with (
            AuthTestClient(self.app) as client,
            patch(
                "frigate.api.media.get_keyframe_before",
                side_effect=lambda _path, offset: offset,
            ),
        ):
            for case in RANGE_CASES:
                with self.subTest(case=case.name):
                    Recordings.delete().execute()
                    super().insert_mock_recording(
                        case.name,
                        REQUEST_START + case.start_offset,
                        REQUEST_START + case.end_offset,
                    )

                    response = client.get(
                        f"/vod/front_door/start/{REQUEST_START}/end/{REQUEST_END}"
                    )

                    if case.vod_duration_ms is None:
                        assert response.status_code == 404
                        continue

                    self._assert_vod_response(
                        response,
                        [
                            (
                                case.name,
                                case.vod_clip_from_ms,
                                case.vod_duration_ms,
                            )
                        ],
                    )

    def test_vod_handles_segment_ending_at_start_with_keyframe_fallbacks(self):
        """VOD keeps a boundary segment when keyframe lookup extends it."""

        def keyframe_before(path: str, offset: int) -> int | None:
            return offset - 1000 if path == "previous_keyframe" else None

        with (
            AuthTestClient(self.app) as client,
            patch(
                "frigate.api.media.get_keyframe_before",
                side_effect=keyframe_before,
            ),
        ):
            super().insert_mock_recording(
                "previous_keyframe",
                REQUEST_START - 10,
                REQUEST_START,
            )
            super().insert_mock_recording(
                "missing_keyframe",
                REQUEST_START - 5,
                REQUEST_START,
            )

            response = client.get(
                f"/vod/front_door/start/{REQUEST_START}/end/{REQUEST_END}"
            )

            self._assert_vod_response(
                response,
                [
                    ("previous_keyframe", 9000, 1000),
                    ("missing_keyframe", None, 5000),
                ],
            )

    def test_recordings_unavailable_reports_gap_between_recordings(self):
        """A gap between two recordings is reported as an unavailable segment."""
        with AuthTestClient(self.app) as client:
            # Two recordings with a 20s gap (1010-1030) between them.
            Recordings.insert(
                id="rec_a",
                path="/media/recordings/a.mp4",
                camera="front_door",
                start_time=1000,
                end_time=1010,
                duration=10,
                motion=0,
            ).execute()
            Recordings.insert(
                id="rec_b",
                path="/media/recordings/b.mp4",
                camera="front_door",
                start_time=1030,
                end_time=1040,
                duration=10,
                motion=0,
            ).execute()

            response = client.get(
                "/recordings/unavailable",
                params={
                    "after": 1000,
                    "before": 1040,
                    "scale": 5,
                    "cameras": "front_door",
                },
            )

            assert response.status_code == 200
            assert response.json() == [{"start_time": 1010, "end_time": 1030}]

    def test_recordings_unavailable_merges_overlapping_recordings(self):
        """Overlapping recordings are merged so no false gap is reported."""
        with AuthTestClient(self.app) as client:
            # Overlapping recordings spanning the whole requested range.
            Recordings.insert(
                id="rec_a",
                path="/media/recordings/a.mp4",
                camera="front_door",
                start_time=1000,
                end_time=1020,
                duration=20,
                motion=0,
            ).execute()
            Recordings.insert(
                id="rec_b",
                path="/media/recordings/b.mp4",
                camera="front_door",
                start_time=1010,
                end_time=1030,
                duration=20,
                motion=0,
            ).execute()

            response = client.get(
                "/recordings/unavailable",
                params={
                    "after": 1000,
                    "before": 1030,
                    "scale": 5,
                    "cameras": "front_door",
                },
            )

            assert response.status_code == 200
            assert response.json() == []

    def test_recordings_unavailable_cameras_all_scopes_to_allowed_cameras(self):
        """cameras=all must not error and must only consider allowed cameras.

        allowed_cameras is mocked to ["front_door"]. A back_door recording that
        would otherwise fill the gap must be ignored, and the request must not
        500 the way it did when cameras was reassigned to a list.
        """
        with AuthTestClient(self.app) as client:
            # front_door has a 20s gap (1010-1030).
            Recordings.insert(
                id="front_a",
                path="/media/recordings/front_a.mp4",
                camera="front_door",
                start_time=1000,
                end_time=1010,
                duration=10,
                motion=0,
            ).execute()
            Recordings.insert(
                id="front_b",
                path="/media/recordings/front_b.mp4",
                camera="front_door",
                start_time=1030,
                end_time=1040,
                duration=10,
                motion=0,
            ).execute()
            # back_door is not in allowed_cameras; its full-window coverage must
            # not mask the front_door gap.
            Recordings.insert(
                id="back_a",
                path="/media/recordings/back_a.mp4",
                camera="back_door",
                start_time=1000,
                end_time=1040,
                duration=40,
                motion=0,
            ).execute()

            response = client.get(
                "/recordings/unavailable",
                params={
                    "after": 1000,
                    "before": 1040,
                    "scale": 5,
                    "cameras": "all",
                },
            )

            assert response.status_code == 200
            assert response.json() == [{"start_time": 1010, "end_time": 1030}]