mirror of
https://github.com/blakeblackshear/frigate.git
synced 2026-05-07 22:15:28 +03:00
814c497bef
Some checks are pending
CI / AMD64 Build (push) Waiting to run
CI / ARM Build (push) Waiting to run
CI / Jetson Jetpack 6 (push) Waiting to run
CI / AMD64 Extra Build (push) Blocked by required conditions
CI / ARM Extra Build (push) Blocked by required conditions
CI / Synaptics Build (push) Blocked by required conditions
CI / Assemble and push default build (push) Blocked by required conditions
* use ReplayState enum * extract shared ffmpeg progress helper * make start call non-blocking with worker thread * expose replay state on status endpoint and return 202 from start * cancel in-flight ffmpeg when stop is called during preparation * add replay i18n strings for preparing and error states * show status in replay UI * navigate immediately on 202 from debug replay menus and dialog * remove unused * simplify to use Job infrastructure * tests * cleanup and tweaks * fetch schema * update api spec * formatting * fix e2e test * mypy * clean up * formatting * fix * fix test * don't try to show camera image until status reports ready * simplify loading logic * fix race in latest_frame on debug replay shutdown * remove toast when successfully stopping it gets hidden almost immediately
124 lines
4.5 KiB
Python
124 lines
4.5 KiB
Python
"""Tests for /debug_replay API endpoints."""
|
|
|
|
from unittest.mock import patch
|
|
|
|
from frigate.models import Event, Recordings, ReviewSegment
|
|
from frigate.test.http_api.base_http_test import AuthTestClient, BaseTestHttp
|
|
|
|
|
|
class TestDebugReplayAPI(BaseTestHttp):
|
|
def setUp(self):
|
|
super().setUp([Event, Recordings, ReviewSegment])
|
|
self.app = self.create_app()
|
|
|
|
def test_start_returns_202_with_job_id(self):
|
|
# Stub the factory to skip validation/threading and just record the
|
|
# name on the manager the way the real factory's mark_starting would.
|
|
def fake_start(**kwargs):
|
|
kwargs["replay_manager"].mark_starting(
|
|
source_camera=kwargs["source_camera"],
|
|
replay_camera_name="_replay_front",
|
|
start_ts=kwargs["start_ts"],
|
|
end_ts=kwargs["end_ts"],
|
|
)
|
|
return "job-1234"
|
|
|
|
with patch(
|
|
"frigate.api.debug_replay.start_debug_replay_job",
|
|
side_effect=fake_start,
|
|
):
|
|
with AuthTestClient(self.app) as client:
|
|
resp = client.post(
|
|
"/debug_replay/start",
|
|
json={
|
|
"camera": "front",
|
|
"start_time": 100,
|
|
"end_time": 200,
|
|
},
|
|
)
|
|
|
|
self.assertEqual(resp.status_code, 202)
|
|
body = resp.json()
|
|
self.assertTrue(body["success"])
|
|
self.assertEqual(body["job_id"], "job-1234")
|
|
self.assertEqual(body["replay_camera"], "_replay_front")
|
|
|
|
def test_start_returns_400_on_validation_error(self):
|
|
with patch(
|
|
"frigate.api.debug_replay.start_debug_replay_job",
|
|
side_effect=ValueError("Camera 'missing' not found"),
|
|
):
|
|
with AuthTestClient(self.app) as client:
|
|
resp = client.post(
|
|
"/debug_replay/start",
|
|
json={
|
|
"camera": "missing",
|
|
"start_time": 100,
|
|
"end_time": 200,
|
|
},
|
|
)
|
|
|
|
self.assertEqual(resp.status_code, 400)
|
|
body = resp.json()
|
|
self.assertFalse(body["success"])
|
|
# Message is hard-coded so we don't echo exception text back to clients
|
|
# (CodeQL: information exposure through an exception).
|
|
self.assertEqual(body["message"], "Invalid debug replay parameters")
|
|
|
|
def test_start_returns_409_when_session_already_active(self):
|
|
with patch(
|
|
"frigate.api.debug_replay.start_debug_replay_job",
|
|
side_effect=RuntimeError("A replay session is already active"),
|
|
):
|
|
with AuthTestClient(self.app) as client:
|
|
resp = client.post(
|
|
"/debug_replay/start",
|
|
json={
|
|
"camera": "front",
|
|
"start_time": 100,
|
|
"end_time": 200,
|
|
},
|
|
)
|
|
|
|
self.assertEqual(resp.status_code, 409)
|
|
body = resp.json()
|
|
self.assertFalse(body["success"])
|
|
|
|
def test_status_inactive_when_no_session(self):
|
|
with AuthTestClient(self.app) as client:
|
|
resp = client.get("/debug_replay/status")
|
|
|
|
self.assertEqual(resp.status_code, 200)
|
|
body = resp.json()
|
|
self.assertFalse(body["active"])
|
|
self.assertIsNone(body["replay_camera"])
|
|
self.assertIsNone(body["source_camera"])
|
|
self.assertIsNone(body["start_time"])
|
|
self.assertIsNone(body["end_time"])
|
|
self.assertFalse(body["live_ready"])
|
|
# Make sure deprecated fields are gone
|
|
self.assertNotIn("state", body)
|
|
self.assertNotIn("progress_percent", body)
|
|
self.assertNotIn("error_message", body)
|
|
|
|
def test_status_active_after_mark_starting(self):
|
|
manager = self.app.replay_manager
|
|
manager.mark_starting(
|
|
source_camera="front",
|
|
replay_camera_name="_replay_front",
|
|
start_ts=100.0,
|
|
end_ts=200.0,
|
|
)
|
|
|
|
with AuthTestClient(self.app) as client:
|
|
resp = client.get("/debug_replay/status")
|
|
|
|
self.assertEqual(resp.status_code, 200)
|
|
body = resp.json()
|
|
self.assertTrue(body["active"])
|
|
self.assertEqual(body["replay_camera"], "_replay_front")
|
|
self.assertEqual(body["source_camera"], "front")
|
|
self.assertEqual(body["start_time"], 100.0)
|
|
self.assertEqual(body["end_time"], 200.0)
|
|
self.assertFalse(body["live_ready"])
|