administrator/frigate
1
0
Fork 0
mirror of https://github.com/blakeblackshear/frigate.git synced 2026-11-02 03:40:31 +03:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity
Files
8bfe78dbd9ea61e773e91dec8595eb1bd39e5dac
frigate/frigate/test/http_api/test_http_media.py
T
Josh HawkinsGitHubNicolas Mowen
 9b02c7318d
CI / Synaptics Build (push) Blocked by required conditions
Details
CI / Assemble and push default build (push) Blocked by required conditions
Details
CI / AMD64 Build (push) Waiting to run
Details
CI / ARM Build (push) Waiting to run
Details
CI / Jetson Jetpack 6 (push) Waiting to run
Details
CI / AMD64 Extra Build (push) Blocked by required conditions
Details
CI / ARM Extra Build (push) Blocked by required conditions
Details
Miscellaneous fixes (#23610) 
* Handle back seeking going to previous clip

* scope /recordings/unavailable query to the caller's allowed cameras

* listen for config updates in activity manager

* don't set search after awaited request

Intentionally do NOT setSearch() to mark the open event submitted. This runs after the awaited request, by which point the user may have closed the dialog; re-setting the parent's selected event would resurrect it and the force-open effect would reopen it (see #23599). The local "submitted" state covers the open card, and mutate() updates the events cache so the grid and any future open reflect the result.

* fix ruff

#23201 removed pathlib import but for some reason it's just now causing ruff to fail

---------

Co-authored-by: Nicolas Mowen <nickmowen213@gmail.com>
2026-07-01 15:03:34 -05:00

530 lines
19 KiB
Python
Raw Blame History

"""Unit tests for recordings/media API endpoints."""
from datetime import datetime, timezone
import pytz
from fastapi import Request
from frigate.api.auth import get_allowed_cameras_for_filter, get_current_user
from frigate.models import Recordings
from frigate.test.http_api.base_http_test import AuthTestClient, BaseTestHttp
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 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=timezone.utc).timestamp()
march_10_utc = datetime(2024, 3, 10, 17, 0, 0, tzinfo=timezone.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_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}]
Reference in New Issue View Git Blame Copy Permalink