From 55e2a5b9160781fc118d9699a40b65a514764a72 Mon Sep 17 00:00:00 2001 From: Josh Hawkins <32435876+hawkeye217@users.noreply.github.com> Date: Thu, 11 Jun 2026 08:10:07 -0500 Subject: [PATCH] tests --- .../http_api/test_http_keyframe_analysis.py | 58 +++++++++ frigate/test/test_keyframe_analysis.py | 111 ++++++++++++++++++ 2 files changed, 169 insertions(+) create mode 100644 frigate/test/http_api/test_http_keyframe_analysis.py create mode 100644 frigate/test/test_keyframe_analysis.py diff --git a/frigate/test/http_api/test_http_keyframe_analysis.py b/frigate/test/http_api/test_http_keyframe_analysis.py new file mode 100644 index 0000000000..6a49e105ec --- /dev/null +++ b/frigate/test/http_api/test_http_keyframe_analysis.py @@ -0,0 +1,58 @@ +from unittest.mock import AsyncMock, patch + +from frigate.models import Event, Recordings, ReviewSegment +from frigate.test.http_api.base_http_test import AuthTestClient, BaseTestHttp + + +class TestHttpKeyframeAnalysis(BaseTestHttp): + def setUp(self): + super().setUp([Event, Recordings, ReviewSegment]) + + def test_invalid_camera_returns_404(self): + app = super().create_app() + with AuthTestClient(app) as client: + response = client.get("/keyframe_analysis?camera=does_not_exist") + assert response.status_code == 404 + + def test_record_disabled_returns_neutral(self): + # default minimal_config has recording disabled + app = super().create_app() + with AuthTestClient(app) as client: + response = client.get("/keyframe_analysis?camera=front_door") + assert response.status_code == 200 + assert response.json()["severity"] == "record_disabled" + + def test_probes_record_input_and_returns_severity(self): + self.minimal_config["cameras"]["front_door"]["ffmpeg"]["inputs"] = [ + { + "path": "rtsp://10.0.0.1:554/record", + "roles": ["detect", "record"], + } + ] + self.minimal_config["cameras"]["front_door"]["record"] = {"enabled": True} + app = super().create_app() + + canned = { + "severity": "ok", + "keyframe_count": 5, + "max_gap": 1.0, + "mean_gap": 1.0, + "min_gap": 1.0, + "segment_time": 10, + "duration_observed": 4.0, + "thresholds": {"warning": 4.0, "error": 10}, + } + + with patch( + "frigate.api.camera.analyze_record_keyframes", + AsyncMock(return_value=canned), + ) as mock_probe: + with AuthTestClient(app) as client: + response = client.get("/keyframe_analysis?camera=front_door") + + assert response.status_code == 200 + assert response.json()["severity"] == "ok" + # index matches the input carrying the record role ("Stream 1") + assert response.json()["stream_index"] == 0 + # the record-role input path was probed + assert mock_probe.await_args.args[1] == "rtsp://10.0.0.1:554/record" diff --git a/frigate/test/test_keyframe_analysis.py b/frigate/test/test_keyframe_analysis.py new file mode 100644 index 0000000000..85302b5659 --- /dev/null +++ b/frigate/test/test_keyframe_analysis.py @@ -0,0 +1,111 @@ +"""Tests for keyframe-spacing analysis used to detect smart/+ codecs.""" + +import asyncio +import unittest +from unittest.mock import AsyncMock, MagicMock, patch + +from frigate.util.services import ( + analyze_record_keyframes, + classify_keyframe_gaps, + parse_keyframe_packets, +) + + +class TestClassifyKeyframeGaps(unittest.TestCase): + def test_ok_when_gaps_small(self): + # keyframes every ~1s + pts = [0.0, 1.0, 2.0, 3.0, 4.0] + result = classify_keyframe_gaps(pts, segment_time=10) + self.assertEqual(result["severity"], "ok") + self.assertEqual(result["max_gap"], 1.0) + self.assertEqual(result["keyframe_count"], 5) + self.assertEqual(result["thresholds"], {"warning": 4.0, "error": 10}) + + def test_warning_when_gap_exceeds_four_seconds(self): + pts = [0.0, 1.0, 6.5] # 5.5s gap + result = classify_keyframe_gaps(pts, segment_time=10) + self.assertEqual(result["severity"], "warning") + self.assertEqual(result["max_gap"], 5.5) + + def test_error_when_gap_exceeds_segment_time(self): + pts = [0.0, 12.0] # 12s gap > 10s segment + result = classify_keyframe_gaps(pts, segment_time=10) + self.assertEqual(result["severity"], "error") + + def test_error_threshold_tracks_segment_time(self): + pts = [0.0, 6.0] # 6s gap, segment_time=5 -> error + result = classify_keyframe_gaps(pts, segment_time=5) + self.assertEqual(result["severity"], "error") + + def test_unknown_with_single_keyframe(self): + result = classify_keyframe_gaps([1.0], segment_time=10) + self.assertEqual(result["severity"], "unknown") + self.assertIsNone(result["max_gap"]) + self.assertEqual(result["keyframe_count"], 1) + + def test_unknown_with_no_keyframes(self): + result = classify_keyframe_gaps([], segment_time=10) + self.assertEqual(result["severity"], "unknown") + self.assertEqual(result["keyframe_count"], 0) + + +class TestParseKeyframePackets(unittest.TestCase): + def test_extracts_keyframe_pts_and_max(self): + output = "0.000000,K__\n0.033333,___\n1.000000,K__\n1.500000,___\n" + keyframe_pts, max_pts = parse_keyframe_packets(output) + self.assertEqual(keyframe_pts, [0.0, 1.0]) + self.assertEqual(max_pts, 1.5) + + def test_skips_unparseable_and_empty_lines(self): + output = "N/A,K__\n\n2.0,K__\nbad line\n" + keyframe_pts, max_pts = parse_keyframe_packets(output) + self.assertEqual(keyframe_pts, [2.0]) + self.assertEqual(max_pts, 2.0) + + def test_empty_output(self): + keyframe_pts, max_pts = parse_keyframe_packets("") + self.assertEqual(keyframe_pts, []) + self.assertIsNone(max_pts) + + +class TestAnalyzeRecordKeyframes(unittest.IsolatedAsyncioTestCase): + async def test_merges_duration_and_classification(self): + csv = b"0.0,K__\n1.0,___\n6.0,K__\n7.0,___\n" + proc = MagicMock() + proc.communicate = AsyncMock(return_value=(csv, b"")) + ffmpeg = MagicMock() + ffmpeg.ffprobe_path = "/usr/bin/ffprobe" + + with patch( + "frigate.util.services.asyncio.create_subprocess_exec", + AsyncMock(return_value=proc), + ): + result = await analyze_record_keyframes( + ffmpeg, "rtsp://cam/stream", segment_time=10 + ) + + self.assertEqual(result["severity"], "warning") # 6s gap > 4s + self.assertEqual(result["max_gap"], 6.0) + self.assertEqual(result["duration_observed"], 7.0) + + async def test_timeout_returns_unknown(self): + proc = MagicMock() + proc.communicate = AsyncMock(side_effect=asyncio.TimeoutError()) + proc.kill = MagicMock() + ffmpeg = MagicMock() + ffmpeg.ffprobe_path = "/usr/bin/ffprobe" + + with patch( + "frigate.util.services.asyncio.create_subprocess_exec", + AsyncMock(return_value=proc), + ): + result = await analyze_record_keyframes( + ffmpeg, "rtsp://cam/stream", segment_time=10 + ) + + self.assertEqual(result["severity"], "unknown") + proc.kill.assert_called_once() + + +if __name__ == "__main__": + unittest.main()