mirror of
https://github.com/blakeblackshear/frigate.git
synced 2026-07-26 13:48:59 +03:00
43d97acd21
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
CI / AMD64 Build (push) Waiting to run
* start audio transcription post processor when enabled on any camera * Fetch embed key whenever an error occurs in case the llama server was restarted * mypy * add tooltips for colored dots in settings menu * add ability to reorder cameras from management pane * add ability to reorder birdseye * add reordering save text to camera management view * Include NPU in latency performance hint * Implement turbo for NPU on object detection * hide order fields * drop auto-derived field paths from camera value when unset globally * use correct field type for export hwaccel args * add debug replay to detail actions menu * clarify debug replay in docs * guard get_current_frame_time against missing camera state * Implement debug reply from export * Refactor debug replay to use sources for dynamic playback * Mypy * fix debug export replay source timestamp handling * skip replay cameras in stats immediately * broadcast debug replay state over ws and buffer pre-OPEN sends - push debug replay session state over the job_state ws topic so the status bar reacts instantly to start/stop without polling - fix child-effect-before-parent-effect race in WsProvider that silently dropped initial snapshot requests on cold load * fix debug replay test hang --------- Co-authored-by: Nicolas Mowen <nickmowen213@gmail.com>
125 lines
4.5 KiB
Python
125 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):
|
|
source = kwargs["source"]
|
|
kwargs["replay_manager"].mark_starting(
|
|
source_camera=source.source_camera,
|
|
replay_camera_name="_replay_front",
|
|
start_ts=source.start_ts,
|
|
end_ts=source.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"])
|