diff --git a/frigate/test/http_api/test_http_media.py b/frigate/test/http_api/test_http_media.py index 6a05efd136..878b1765ef 100644 --- a/frigate/test/http_api/test_http_media.py +++ b/frigate/test/http_api/test_http_media.py @@ -1,5 +1,6 @@ """Unit tests for recordings/media API endpoints.""" +from dataclasses import dataclass from datetime import UTC, datetime from unittest.mock import patch @@ -12,6 +13,61 @@ 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.""" @@ -46,6 +102,26 @@ class TestHttpMedia(BaseTestHttp): 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). @@ -406,54 +482,101 @@ class TestHttpMedia(BaseTestHttp): assert "2024-03-10" in summary assert summary["2024-03-10"] is True - def test_recordings_returns_overlapping_segment(self): - """Recordings include a segment that overlaps the requested range.""" - after = 1000 - before = 1100 - + def test_recordings_handles_all_range_relations(self): + """Recordings return every interval relation that touches the range.""" with AuthTestClient(self.app) as client: - super().insert_mock_recording("too_old", after - 10, after - 1) - super().insert_mock_recording( - "overlap", - after - MAX_SEGMENT_DURATION + 1, - after, - ) - super().insert_mock_recording("too_new", before + 1, before + 10) + 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": after, "before": before}, - ) + response = client.get( + "/front_door/recordings", + params={"after": REQUEST_START, "before": REQUEST_END}, + ) - assert response.status_code == 200 - assert [recording["id"] for recording in response.json()] == ["overlap"] + 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_returns_overlapping_segment(self): - """VOD mapping includes a segment that overlaps the requested range.""" - start = 1000 - end = 1100 + 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", - return_value=None, + side_effect=keyframe_before, ), ): - super().insert_mock_recording("too_old", start - 10, start - 1) super().insert_mock_recording( - "overlap", - start - MAX_SEGMENT_DURATION + 2, - start + 1, + "previous_keyframe", + REQUEST_START - 10, + REQUEST_START, + ) + super().insert_mock_recording( + "missing_keyframe", + REQUEST_START - 5, + REQUEST_START, ) - super().insert_mock_recording("too_new", end + 1, end + 10) - response = client.get(f"/vod/front_door/start/{start}/end/{end}") + response = client.get( + f"/vod/front_door/start/{REQUEST_START}/end/{REQUEST_END}" + ) - assert response.status_code == 200 - assert [ - clip["path"] for clip in response.json()["sequences"][0]["clips"] - ] == ["overlap"] + 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."""