administrator/frigate
1
0
Fork 0
mirror of https://github.com/blakeblackshear/frigate.git synced 2025-12-06 05:24:11 +03:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

Enforce default admin role requirement for API endpoints (#21065)
Some checks are pending
CI / AMD64 Extra Build (push) Blocked by required conditions
Details
CI / ARM Extra 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 / Synaptics Build (push) Blocked by required conditions
Details
CI / Assemble and push default build (push) Blocked by required conditions
Details

Browse Source 
* require admin role by default

* update all endpoint access guards

* explicit paths and prefixes exception lists

* fix tests to use mock auth

* add helper and simplify auth conditions

* add missing exempt path

* fix test

* make metrics endpoint require auth
This commit is contained in:
Josh Hawkins 2025-11-26 15:07:28 -06:00 committed by GitHub
parent de2144f158
commit cd606ad240
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
13 changed files with 512 additions and 182 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

46
frigate/api/app.py
View File

@ -23,7 +23,7 @@ from markupsafe import escape
from peewee import SQL, fn, operator
from pydantic import ValidationError
from frigate.api.auth import require_role
from frigate.api.auth import allow_any_authenticated, allow_public, require_role
from frigate.api.defs.query.app_query_parameters import AppTimelineHourlyQueryParameters
from frigate.api.defs.request.app_body import AppConfigSetBody
from frigate.api.defs.tags import Tags
@ -56,29 +56,33 @@ logger = logging.getLogger(__name__)
router = APIRouter(tags=[Tags.app])
@router.get("/", response_class=PlainTextResponse)
@router.get(
"/", response_class=PlainTextResponse, dependencies=[Depends(allow_public())]
)
def is_healthy():
return "Frigate is running. Alive and healthy!"
@router.get("/config/schema.json")
@router.get("/config/schema.json", dependencies=[Depends(allow_public())])
def config_schema(request: Request):
return Response(
content=request.app.frigate_config.schema_json(), media_type="application/json"
)
@router.get("/version", response_class=PlainTextResponse)
@router.get(
"/version", response_class=PlainTextResponse, dependencies=[Depends(allow_public())]
)
def version():
return VERSION
@router.get("/stats")
@router.get("/stats", dependencies=[Depends(allow_any_authenticated())])
def stats(request: Request):
return JSONResponse(content=request.app.stats_emitter.get_latest_stats())
@router.get("/stats/history")
@router.get("/stats/history", dependencies=[Depends(allow_any_authenticated())])
def stats_history(request: Request, keys: str = None):
if keys:
keys = keys.split(",")
@ -86,7 +90,7 @@ def stats_history(request: Request, keys: str = None):
return JSONResponse(content=request.app.stats_emitter.get_stats_history(keys))
@router.get("/metrics")
@router.get("/metrics", dependencies=[Depends(allow_any_authenticated())])
def metrics(request: Request):
"""Expose Prometheus metrics endpoint and update metrics with latest stats"""
# Retrieve the latest statistics and update the Prometheus metrics
@ -103,7 +107,7 @@ def metrics(request: Request):
return Response(content=content, media_type=content_type)
@router.get("/config")
@router.get("/config", dependencies=[Depends(allow_any_authenticated())])
def config(request: Request):
config_obj: FrigateConfig = request.app.frigate_config
config: dict[str, dict[str, Any]] = config_obj.model_dump(
@ -209,7 +213,7 @@ def config_raw_paths(request: Request):
return JSONResponse(content=raw_paths)
@router.get("/config/raw")
@router.get("/config/raw", dependencies=[Depends(allow_any_authenticated())])
def config_raw():
config_file = find_config_file()
@ -452,7 +456,7 @@ def config_set(request: Request, body: AppConfigSetBody):
)
@router.get("/vainfo")
@router.get("/vainfo", dependencies=[Depends(allow_any_authenticated())])
def vainfo():
vainfo = vainfo_hwaccel()
return JSONResponse(
@ -472,12 +476,16 @@ def vainfo():
)
@router.get("/nvinfo")
@router.get("/nvinfo", dependencies=[Depends(allow_any_authenticated())])
def nvinfo():
return JSONResponse(content=get_nvidia_driver_info())
@router.get("/logs/{service}", tags=[Tags.logs])
@router.get(
"/logs/{service}",
tags=[Tags.logs],
dependencies=[Depends(allow_any_authenticated())],
)
async def logs(
service: str = Path(enum=["frigate", "nginx", "go2rtc"]),
download: Optional[str] = None,
@ -585,7 +593,7 @@ def restart():
)
@router.get("/labels")
@router.get("/labels", dependencies=[Depends(allow_any_authenticated())])
def get_labels(camera: str = ""):
try:
if camera:
@ -603,7 +611,7 @@ def get_labels(camera: str = ""):
return JSONResponse(content=labels)
@router.get("/sub_labels")
@router.get("/sub_labels", dependencies=[Depends(allow_any_authenticated())])
def get_sub_labels(split_joined: Optional[int] = None):
try:
events = Event.select(Event.sub_label).distinct()
@ -634,7 +642,7 @@ def get_sub_labels(split_joined: Optional[int] = None):
return JSONResponse(content=sub_labels)
@router.get("/plus/models")
@router.get("/plus/models", dependencies=[Depends(allow_any_authenticated())])
def plusModels(request: Request, filterByCurrentModelDetector: bool = False):
if not request.app.frigate_config.plus_api.is_active():
return JSONResponse(
@ -676,7 +684,9 @@ def plusModels(request: Request, filterByCurrentModelDetector: bool = False):
return JSONResponse(content=validModels)
@router.get("/recognized_license_plates")
@router.get(
"/recognized_license_plates", dependencies=[Depends(allow_any_authenticated())]
)
def get_recognized_license_plates(split_joined: Optional[int] = None):
try:
query = (
@ -710,7 +720,7 @@ def get_recognized_license_plates(split_joined: Optional[int] = None):
return JSONResponse(content=recognized_license_plates)
@router.get("/timeline")
@router.get("/timeline", dependencies=[Depends(allow_any_authenticated())])
def timeline(camera: str = "all", limit: int = 100, source_id: Optional[str] = None):
clauses = []
@ -747,7 +757,7 @@ def timeline(camera: str = "all", limit: int = 100, source_id: Optional[str] = N
return JSONResponse(content=[t for t in timeline])
@router.get("/timeline/hourly")
@router.get("/timeline/hourly", dependencies=[Depends(allow_any_authenticated())])
def hourly_timeline(params: AppTimelineHourlyQueryParameters = Depends()):
"""Get hourly summary for timeline."""
cameras = params.cameras

158
frigate/api/auth.py
View File

@ -32,10 +32,154 @@ from frigate.models import User
logger = logging.getLogger(__name__)
def require_admin_by_default():
"""
Global admin requirement dependency for all endpoints by default.
This is set as the default dependency on the FastAPI app to ensure all
endpoints require admin access unless explicitly overridden with
allow_public(), allow_any_authenticated(), or require_role().
Port 5000 (internal) always has admin role set by the /auth endpoint,
so this check passes automatically for internal requests.
Certain paths are exempted from the global admin check because they must
be accessible before authentication (login, auth) or they have their own
route-level authorization dependencies that handle access control.
"""
# Paths that have route-level auth dependencies and should bypass global admin check
# These paths still have authorization - it's handled by their route-level dependencies
EXEMPT_PATHS = {
# Public auth endpoints (allow_public)
"/auth",
"/auth/first_time_login",
"/login",
# Authenticated user endpoints (allow_any_authenticated)
"/logout",
"/profile",
# Public info endpoints (allow_public)
"/",
"/version",
"/config/schema.json",
"/metrics",
# Authenticated user endpoints (allow_any_authenticated)
"/stats",
"/stats/history",
"/config",
"/config/raw",
"/vainfo",
"/nvinfo",
"/labels",
"/sub_labels",
"/plus/models",
"/recognized_license_plates",
"/timeline",
"/timeline/hourly",
"/events/summary",
"/recordings/storage",
"/recordings/summary",
"/recordings/unavailable",
"/go2rtc/streams",
}
# Path prefixes that should be exempt (for paths with parameters)
EXEMPT_PREFIXES = (
"/logs/", # /logs/{service}
"/review", # /review, /review/{id}, /review_ids, /review/summary, etc.
"/reviews/", # /reviews/viewed, /reviews/delete
"/events/", # /events/{id}/thumbnail, etc. (camera-scoped)
"/go2rtc/streams/", # /go2rtc/streams/{camera}
"/users/", # /users/{username}/password (has own auth)
"/preview/", # /preview/{file}/thumbnail.jpg
)
async def admin_checker(request: Request):
path = request.url.path
# Check exact path matches
if path in EXEMPT_PATHS:
return
# Check prefix matches for parameterized paths
if path.startswith(EXEMPT_PREFIXES):
return
# For all other paths, require admin role
# Port 5000 (internal) requests have admin role set automatically
role = request.headers.get("remote-role")
if role == "admin":
return
raise HTTPException(
status_code=403,
detail="Admin role required for this endpoint",
)
return admin_checker
def _is_authenticated(request: Request) -> bool:
"""
Helper to determine if a request is from an authenticated user.
Returns True if the request has a valid authenticated user (not anonymous).
Port 5000 internal requests are considered anonymous despite having admin role.
"""
username = request.headers.get("remote-user")
return username is not None and username != "anonymous"
def allow_public():
"""
Override dependency to allow unauthenticated access to an endpoint.
Use this for endpoints that should be publicly accessible without
authentication, such as login page, health checks, or pre-auth info.
Example:
@router.get("/public-endpoint", dependencies=[Depends(allow_public())])
"""
async def public_checker(request: Request):
return # Always allow
return public_checker
def allow_any_authenticated():
"""
Override dependency to allow any authenticated user (bypass admin requirement).
Allows:
- Port 5000 internal requests (have admin role despite anonymous user)
- Any authenticated user with a real username (not "anonymous")
Rejects:
- Port 8971 requests with anonymous user (auth disabled, no proxy auth)
Example:
@router.get("/authenticated-endpoint", dependencies=[Depends(allow_any_authenticated())])
"""
async def auth_checker(request: Request):
# Port 5000 requests have admin role and should be allowed
role = request.headers.get("remote-role")
if role == "admin":
return
# Otherwise require a real authenticated user (not anonymous)
if not _is_authenticated(request):
raise HTTPException(status_code=401, detail="Authentication required")
return
return auth_checker
router = APIRouter(tags=[Tags.auth])
@router.get("/auth/first_time_login")
@router.get("/auth/first_time_login", dependencies=[Depends(allow_public())])
def first_time_login(request: Request):
"""Return whether the admin first-time login help flag is set in config.
@ -352,7 +496,7 @@ def resolve_role(
# Endpoints
@router.get("/auth")
@router.get("/auth", dependencies=[Depends(allow_public())])
def auth(request: Request):
auth_config: AuthConfig = request.app.frigate_config.auth
proxy_config: ProxyConfig = request.app.frigate_config.proxy
@ -478,7 +622,7 @@ def auth(request: Request):
return fail_response
@router.get("/profile")
@router.get("/profile", dependencies=[Depends(allow_any_authenticated())])
def profile(request: Request):
username = request.headers.get("remote-user", "anonymous")
role = request.headers.get("remote-role", "viewer")
@ -492,7 +636,7 @@ def profile(request: Request):
)
@router.get("/logout")
@router.get("/logout", dependencies=[Depends(allow_any_authenticated())])
def logout(request: Request):
auth_config: AuthConfig = request.app.frigate_config.auth
response = RedirectResponse("/login", status_code=303)
@ -503,7 +647,7 @@ def logout(request: Request):
limiter = Limiter(key_func=get_remote_addr)
@router.post("/login")
@router.post("/login", dependencies=[Depends(allow_public())])
@limiter.limit(limit_value=rateLimiter.get_limit)
def login(request: Request, body: AppPostLoginBody):
JWT_COOKIE_NAME = request.app.frigate_config.auth.cookie_name
@ -590,7 +734,9 @@ def delete_user(request: Request, username: str):
return JSONResponse(content={"success": True})
@router.put("/users/{username}/password")
@router.put(
"/users/{username}/password", dependencies=[Depends(allow_any_authenticated())]
)
async def update_password(
request: Request,
username: str,

14
frigate/api/camera.py
View File

@ -15,7 +15,11 @@ from onvif import ONVIFCamera, ONVIFError
from zeep.exceptions import Fault, TransportError
from zeep.transports import AsyncTransport
from frigate.api.auth import require_role
from frigate.api.auth import (
allow_any_authenticated,
require_camera_access,
require_role,
)
from frigate.api.defs.tags import Tags
from frigate.config.config import FrigateConfig
from frigate.util.builtin import clean_camera_user_pass
@ -50,7 +54,7 @@ def _is_valid_host(host: str) -> bool:
return False
@router.get("/go2rtc/streams")
@router.get("/go2rtc/streams", dependencies=[Depends(allow_any_authenticated())])
def go2rtc_streams():
r = requests.get("http://127.0.0.1:1984/api/streams")
if not r.ok:
@ -66,7 +70,9 @@ def go2rtc_streams():
return JSONResponse(content=stream_data)
@router.get("/go2rtc/streams/{camera_name}")
@router.get(
"/go2rtc/streams/{camera_name}", dependencies=[Depends(require_camera_access)]
)
def go2rtc_camera_stream(request: Request, camera_name: str):
r = requests.get(
f"http://127.0.0.1:1984/api/streams?src={camera_name}&video=all&audio=all&microphone"
@ -161,7 +167,7 @@ def go2rtc_delete_stream(stream_name: str):
)
@router.get("/ffprobe")
@router.get("/ffprobe", dependencies=[Depends(require_role(["admin"]))])
def ffprobe(request: Request, paths: str = "", detailed: bool = False):
path_param = paths

3
frigate/api/event.py
View File

@ -22,6 +22,7 @@ from peewee import JOIN, DoesNotExist, fn, operator
from playhouse.shortcuts import model_to_dict
from frigate.api.auth import (
allow_any_authenticated,
get_allowed_cameras_for_filter,
require_camera_access,
require_role,
@ -808,7 +809,7 @@ def events_search(
return JSONResponse(content=processed_events)
@router.get("/events/summary")
@router.get("/events/summary", dependencies=[Depends(allow_any_authenticated())])
def events_summary(
params: EventsSummaryQueryParams = Depends(),
allowed_cameras: List[str] = Depends(get_allowed_cameras_for_filter),

8
frigate/api/fastapi_app.py
View File

@ -2,7 +2,7 @@ import logging
import re
from typing import Optional
from fastapi import FastAPI, Request
from fastapi import Depends, FastAPI, Request
from fastapi.responses import JSONResponse
from joserfc.jwk import OctKey
from playhouse.sqliteq import SqliteQueueDatabase
@ -24,7 +24,7 @@ from frigate.api import (
preview,
review,
)
from frigate.api.auth import get_jwt_secret, limiter
from frigate.api.auth import get_jwt_secret, limiter, require_admin_by_default
from frigate.comms.event_metadata_updater import (
EventMetadataPublisher,
)
@ -62,11 +62,15 @@ def create_fastapi_app(
stats_emitter: StatsEmitter,
event_metadata_updater: EventMetadataPublisher,
config_publisher: CameraConfigUpdatePublisher,
enforce_default_admin: bool = True,
):
logger.info("Starting FastAPI app")
app = FastAPI(
debug=False,
swagger_ui_parameters={"apisSorter": "alpha", "operationsSorter": "alpha"},
dependencies=[Depends(require_admin_by_default())]
if enforce_default_admin
else [],
)
# update the request_address with the x-forwarded-for header from nginx

44
frigate/api/media.py
View File

@ -22,7 +22,11 @@ from pathvalidate import sanitize_filename
from peewee import DoesNotExist, fn, operator
from tzlocal import get_localzone_name
from frigate.api.auth import get_allowed_cameras_for_filter, require_camera_access
from frigate.api.auth import (
allow_any_authenticated,
get_allowed_cameras_for_filter,
require_camera_access,
)
from frigate.api.defs.query.media_query_parameters import (
Extension,
MediaEventsSnapshotQueryParams,
@ -393,7 +397,7 @@ async def submit_recording_snapshot_to_plus(
)
@router.get("/recordings/storage")
@router.get("/recordings/storage", dependencies=[Depends(allow_any_authenticated())])
def get_recordings_storage_usage(request: Request):
recording_stats = request.app.stats_emitter.get_latest_stats()["service"][
"storage"
@ -417,7 +421,7 @@ def get_recordings_storage_usage(request: Request):
return JSONResponse(content=camera_usages)
@router.get("/recordings/summary")
@router.get("/recordings/summary", dependencies=[Depends(allow_any_authenticated())])
def all_recordings_summary(
request: Request,
params: MediaRecordingsSummaryQueryParams = Depends(),
@ -635,7 +639,11 @@ async def recordings(
return JSONResponse(content=list(recordings))
@router.get("/recordings/unavailable", response_model=list[dict])
@router.get(
"/recordings/unavailable",
response_model=list[dict],
dependencies=[Depends(allow_any_authenticated())],
)
async def no_recordings(
request: Request,
params: MediaRecordingsAvailabilityQueryParams = Depends(),
@ -1053,7 +1061,10 @@ async def event_snapshot(
)
@router.get("/events/{event_id}/thumbnail.{extension}")
@router.get(
"/events/{event_id}/thumbnail.{extension}",
dependencies=[Depends(require_camera_access)],
)
async def event_thumbnail(
request: Request,
event_id: str,
@ -1251,7 +1262,10 @@ def grid_snapshot(
)
@router.get("/events/{event_id}/snapshot-clean.webp")
@router.get(
"/events/{event_id}/snapshot-clean.webp",
dependencies=[Depends(require_camera_access)],
)
def event_snapshot_clean(request: Request, event_id: str, download: bool = False):
webp_bytes = None
try:
@ -1375,7 +1389,9 @@ def event_snapshot_clean(request: Request, event_id: str, download: bool = False
)
@router.get("/events/{event_id}/clip.mp4")
@router.get(
"/events/{event_id}/clip.mp4", dependencies=[Depends(require_camera_access)]
)
async def event_clip(
request: Request,
event_id: str,
@ -1403,7 +1419,9 @@ async def event_clip(
)
@router.get("/events/{event_id}/preview.gif")
@router.get(
"/events/{event_id}/preview.gif", dependencies=[Depends(require_camera_access)]
)
def event_preview(request: Request, event_id: str):
try:
event: Event = Event.get(Event.id == event_id)
@ -1756,7 +1774,7 @@ def preview_mp4(
)
@router.get("/review/{event_id}/preview")
@router.get("/review/{event_id}/preview", dependencies=[Depends(require_camera_access)])
def review_preview(
request: Request,
event_id: str,
@ -1782,8 +1800,12 @@ def review_preview(
return preview_mp4(request, review.camera, start_ts, end_ts)
@router.get("/preview/{file_name}/thumbnail.jpg")
@router.get("/preview/{file_name}/thumbnail.webp")
@router.get(
"/preview/{file_name}/thumbnail.jpg", dependencies=[Depends(require_camera_access)]
)
@router.get(
"/preview/{file_name}/thumbnail.webp", dependencies=[Depends(require_camera_access)]
)
def preview_thumbnail(file_name: str):
"""Get a thumbnail from the cached preview frames."""
if len(file_name) > 1000:

43
frigate/api/review.py
View File

@ -14,6 +14,7 @@ from peewee import Case, DoesNotExist, IntegrityError, fn, operator
from playhouse.shortcuts import model_to_dict
from frigate.api.auth import (
allow_any_authenticated,
get_allowed_cameras_for_filter,
get_current_user,
require_camera_access,
@ -43,7 +44,11 @@ logger = logging.getLogger(__name__)
router = APIRouter(tags=[Tags.review])
@router.get("/review", response_model=list[ReviewSegmentResponse])
@router.get(
"/review",
response_model=list[ReviewSegmentResponse],
dependencies=[Depends(allow_any_authenticated())],
)
async def review(
params: ReviewQueryParams = Depends(),
current_user: dict = Depends(get_current_user),
@ -152,7 +157,11 @@ async def review(
return JSONResponse(content=[r for r in review_query])
@router.get("/review_ids", response_model=list[ReviewSegmentResponse])
@router.get(
"/review_ids",
response_model=list[ReviewSegmentResponse],
dependencies=[Depends(allow_any_authenticated())],
)
async def review_ids(request: Request, ids: str):
ids = ids.split(",")
@ -186,7 +195,11 @@ async def review_ids(request: Request, ids: str):
)
@router.get("/review/summary", response_model=ReviewSummaryResponse)
@router.get(
"/review/summary",
response_model=ReviewSummaryResponse,
dependencies=[Depends(allow_any_authenticated())],
)
async def review_summary(
params: ReviewSummaryQueryParams = Depends(),
current_user: dict = Depends(get_current_user),
@ -461,7 +474,11 @@ async def review_summary(
return JSONResponse(content=data)
@router.post("/reviews/viewed", response_model=GenericResponse)
@router.post(
"/reviews/viewed",
response_model=GenericResponse,
dependencies=[Depends(allow_any_authenticated())],
)
async def set_multiple_reviewed(
request: Request,
body: ReviewModifyMultipleBody,
@ -644,7 +661,11 @@ def motion_activity(
return JSONResponse(content=normalized)
@router.get("/review/event/{event_id}", response_model=ReviewSegmentResponse)
@router.get(
"/review/event/{event_id}",
response_model=ReviewSegmentResponse,
dependencies=[Depends(allow_any_authenticated())],
)
async def get_review_from_event(request: Request, event_id: str):
try:
review = ReviewSegment.get(
@ -659,7 +680,11 @@ async def get_review_from_event(request: Request, event_id: str):
)
@router.get("/review/{review_id}", response_model=ReviewSegmentResponse)
@router.get(
"/review/{review_id}",
response_model=ReviewSegmentResponse,
dependencies=[Depends(allow_any_authenticated())],
)
async def get_review(request: Request, review_id: str):
try:
review = ReviewSegment.get(ReviewSegment.id == review_id)
@ -672,7 +697,11 @@ async def get_review(request: Request, review_id: str):
)
@router.delete("/review/{review_id}/viewed", response_model=GenericResponse)
@router.delete(
"/review/{review_id}/viewed",
response_model=GenericResponse,
dependencies=[Depends(allow_any_authenticated())],
)
async def set_not_reviewed(
review_id: str,
current_user: dict = Depends(get_current_user),

45
frigate/test/http_api/base_http_test.py
View File

@ -3,6 +3,8 @@ import logging
import os
import unittest
from fastapi import Request
from fastapi.testclient import TestClient
from peewee_migrate import Router
from playhouse.sqlite_ext import SqliteExtDatabase
from playhouse.sqliteq import SqliteQueueDatabase
@ -16,6 +18,20 @@ from frigate.review.types import SeverityEnum
from frigate.test.const import TEST_DB, TEST_DB_CLEANUPS
class AuthTestClient(TestClient):
"""TestClient that automatically adds auth headers to all requests."""
def request(self, *args, **kwargs):
# Add default auth headers if not already present
headers = kwargs.get("headers") or {}
if "remote-user" not in headers:
headers["remote-user"] = "admin"
if "remote-role" not in headers:
headers["remote-role"] = "admin"
kwargs["headers"] = headers
return super().request(*args, **kwargs)
class BaseTestHttp(unittest.TestCase):
def setUp(self, models):
# setup clean database for each test run
@ -113,7 +129,9 @@ class BaseTestHttp(unittest.TestCase):
pass
def create_app(self, stats=None, event_metadata_publisher=None):
return create_fastapi_app(
from frigate.api.auth import get_allowed_cameras_for_filter, get_current_user
app = create_fastapi_app(
FrigateConfig(**self.minimal_config),
self.db,
None,
@ -123,8 +141,33 @@ class BaseTestHttp(unittest.TestCase):
stats,
event_metadata_publisher,
None,
enforce_default_admin=False,
)
# Default test mocks for authentication
# Tests can override these in their setUp if needed
# This mock uses headers set by AuthTestClient
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}
async def mock_get_allowed_cameras_for_filter(request: Request):
return list(self.minimal_config.get("cameras", {}).keys())
app.dependency_overrides[get_current_user] = mock_get_current_user
app.dependency_overrides[get_allowed_cameras_for_filter] = (
mock_get_allowed_cameras_for_filter
)
return app
def insert_mock_event(
self,
id: str,

6
frigate/test/http_api/test_http_app.py
View File

@ -1,10 +1,8 @@
from unittest.mock import Mock
from fastapi.testclient import TestClient
from frigate.models import Event, Recordings, ReviewSegment
from frigate.stats.emitter import StatsEmitter
from frigate.test.http_api.base_http_test import BaseTestHttp
from frigate.test.http_api.base_http_test import AuthTestClient, BaseTestHttp
class TestHttpApp(BaseTestHttp):
@ -20,7 +18,7 @@ class TestHttpApp(BaseTestHttp):
stats.get_latest_stats.return_value = self.test_stats
app = super().create_app(stats)
with TestClient(app) as client:
with AuthTestClient(app) as client:
response = client.get("/stats")
response_json = response.json()
assert response_json == self.test_stats

113
frigate/test/http_api/test_http_camera_access.py
View File

@ -1,14 +1,13 @@
from unittest.mock import patch
from fastapi import HTTPException, Request
from fastapi.testclient import TestClient
from frigate.api.auth import (
get_allowed_cameras_for_filter,
get_current_user,
)
from frigate.models import Event, Recordings, ReviewSegment
from frigate.test.http_api.base_http_test import BaseTestHttp
from frigate.test.http_api.base_http_test import AuthTestClient, BaseTestHttp
class TestCameraAccessEventReview(BaseTestHttp):
@ -16,9 +15,17 @@ class TestCameraAccessEventReview(BaseTestHttp):
super().setUp([Event, ReviewSegment, Recordings])
self.app = super().create_app()
# Mock get_current_user to return valid user for all tests
async def mock_get_current_user():
return {"username": "test_user", "role": "user"}
# 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
@ -30,21 +37,25 @@ class TestCameraAccessEventReview(BaseTestHttp):
super().insert_mock_event("event1", camera="front_door")
super().insert_mock_event("event2", camera="back_door")
self.app.dependency_overrides[get_allowed_cameras_for_filter] = lambda: [
"front_door"
]
with TestClient(self.app) as client:
async def mock_cameras(request: Request):
return ["front_door"]
self.app.dependency_overrides[get_allowed_cameras_for_filter] = mock_cameras
with AuthTestClient(self.app) as client:
resp = client.get("/events")
assert resp.status_code == 200
ids = [e["id"] for e in resp.json()]
assert "event1" in ids
assert "event2" not in ids
self.app.dependency_overrides[get_allowed_cameras_for_filter] = lambda: [
"front_door",
"back_door",
]
with TestClient(self.app) as client:
async def mock_cameras(request: Request):
return [
"front_door",
"back_door",
]
self.app.dependency_overrides[get_allowed_cameras_for_filter] = mock_cameras
with AuthTestClient(self.app) as client:
resp = client.get("/events")
assert resp.status_code == 200
ids = [e["id"] for e in resp.json()]
@ -54,21 +65,25 @@ class TestCameraAccessEventReview(BaseTestHttp):
super().insert_mock_review_segment("rev1", camera="front_door")
super().insert_mock_review_segment("rev2", camera="back_door")
self.app.dependency_overrides[get_allowed_cameras_for_filter] = lambda: [
"front_door"
]
with TestClient(self.app) as client:
async def mock_cameras(request: Request):
return ["front_door"]
self.app.dependency_overrides[get_allowed_cameras_for_filter] = mock_cameras
with AuthTestClient(self.app) as client:
resp = client.get("/review")
assert resp.status_code == 200
ids = [r["id"] for r in resp.json()]
assert "rev1" in ids
assert "rev2" not in ids
self.app.dependency_overrides[get_allowed_cameras_for_filter] = lambda: [
"front_door",
"back_door",
]
with TestClient(self.app) as client:
async def mock_cameras(request: Request):
return [
"front_door",
"back_door",
]
self.app.dependency_overrides[get_allowed_cameras_for_filter] = mock_cameras
with AuthTestClient(self.app) as client:
resp = client.get("/review")
assert resp.status_code == 200
ids = [r["id"] for r in resp.json()]
@ -84,7 +99,7 @@ class TestCameraAccessEventReview(BaseTestHttp):
raise HTTPException(status_code=403, detail="Access denied")
with patch("frigate.api.event.require_camera_access", mock_require_allowed):
with TestClient(self.app) as client:
with AuthTestClient(self.app) as client:
resp = client.get("/events/event1")
assert resp.status_code == 200
assert resp.json()["id"] == "event1"
@ -94,7 +109,7 @@ class TestCameraAccessEventReview(BaseTestHttp):
raise HTTPException(status_code=403, detail="Access denied")
with patch("frigate.api.event.require_camera_access", mock_require_disallowed):
with TestClient(self.app) as client:
with AuthTestClient(self.app) as client:
resp = client.get("/events/event1")
assert resp.status_code == 403
@ -108,7 +123,7 @@ class TestCameraAccessEventReview(BaseTestHttp):
raise HTTPException(status_code=403, detail="Access denied")
with patch("frigate.api.review.require_camera_access", mock_require_allowed):
with TestClient(self.app) as client:
with AuthTestClient(self.app) as client:
resp = client.get("/review/rev1")
assert resp.status_code == 200
assert resp.json()["id"] == "rev1"
@ -118,7 +133,7 @@ class TestCameraAccessEventReview(BaseTestHttp):
raise HTTPException(status_code=403, detail="Access denied")
with patch("frigate.api.review.require_camera_access", mock_require_disallowed):
with TestClient(self.app) as client:
with AuthTestClient(self.app) as client:
resp = client.get("/review/rev1")
assert resp.status_code == 403
@ -126,21 +141,25 @@ class TestCameraAccessEventReview(BaseTestHttp):
super().insert_mock_event("event1", camera="front_door")
super().insert_mock_event("event2", camera="back_door")
self.app.dependency_overrides[get_allowed_cameras_for_filter] = lambda: [
"front_door"
]
with TestClient(self.app) as client:
async def mock_cameras(request: Request):
return ["front_door"]
self.app.dependency_overrides[get_allowed_cameras_for_filter] = mock_cameras
with AuthTestClient(self.app) as client:
resp = client.get("/events", params={"cameras": "all"})
assert resp.status_code == 200
ids = [e["id"] for e in resp.json()]
assert "event1" in ids
assert "event2" not in ids
self.app.dependency_overrides[get_allowed_cameras_for_filter] = lambda: [
"front_door",
"back_door",
]
with TestClient(self.app) as client:
async def mock_cameras(request: Request):
return [
"front_door",
"back_door",
]
self.app.dependency_overrides[get_allowed_cameras_for_filter] = mock_cameras
with AuthTestClient(self.app) as client:
resp = client.get("/events", params={"cameras": "all"})
assert resp.status_code == 200
ids = [e["id"] for e in resp.json()]
@ -150,20 +169,24 @@ class TestCameraAccessEventReview(BaseTestHttp):
super().insert_mock_event("event1", camera="front_door")
super().insert_mock_event("event2", camera="back_door")
self.app.dependency_overrides[get_allowed_cameras_for_filter] = lambda: [
"front_door"
]
with TestClient(self.app) as client:
async def mock_cameras(request: Request):
return ["front_door"]
self.app.dependency_overrides[get_allowed_cameras_for_filter] = mock_cameras
with AuthTestClient(self.app) as client:
resp = client.get("/events/summary")
assert resp.status_code == 200
summary_list = resp.json()
assert len(summary_list) == 1
self.app.dependency_overrides[get_allowed_cameras_for_filter] = lambda: [
"front_door",
"back_door",
]
with TestClient(self.app) as client:
async def mock_cameras(request: Request):
return [
"front_door",
"back_door",
]
self.app.dependency_overrides[get_allowed_cameras_for_filter] = mock_cameras
with AuthTestClient(self.app) as client:
resp = client.get("/events/summary")
summary_list = resp.json()
assert len(summary_list) == 2

66
frigate/test/http_api/test_http_event.py
View File

@ -2,14 +2,13 @@ from datetime import datetime
from typing import Any
from unittest.mock import Mock
from fastapi.testclient import TestClient
from playhouse.shortcuts import model_to_dict
from frigate.api.auth import get_allowed_cameras_for_filter, get_current_user
from frigate.comms.event_metadata_updater import EventMetadataPublisher
from frigate.models import Event, Recordings, ReviewSegment, Timeline
from frigate.stats.emitter import StatsEmitter
from frigate.test.http_api.base_http_test import BaseTestHttp
from frigate.test.http_api.base_http_test import AuthTestClient, BaseTestHttp, Request
from frigate.test.test_storage import _insert_mock_event
@ -18,14 +17,26 @@ class TestHttpApp(BaseTestHttp):
super().setUp([Event, Recordings, ReviewSegment, Timeline])
self.app = super().create_app()
# Mock auth to bypass camera access for tests
async def mock_get_current_user(request: Any):
return {"username": "test_user", "role": "admin"}
# 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
self.app.dependency_overrides[get_allowed_cameras_for_filter] = lambda: [
"front_door"
]
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):
self.app.dependency_overrides.clear()
@ -35,20 +46,20 @@ class TestHttpApp(BaseTestHttp):
################################### GET /events Endpoint #########################################################
####################################################################################################################
def test_get_event_list_no_events(self):
with TestClient(self.app) as client:
with AuthTestClient(self.app) as client:
events = client.get("/events").json()
assert len(events) == 0
def test_get_event_list_no_match_event_id(self):
id = "123456.random"
with TestClient(self.app) as client:
with AuthTestClient(self.app) as client:
super().insert_mock_event(id)
events = client.get("/events", params={"event_id": "abc"}).json()
assert len(events) == 0
def test_get_event_list_match_event_id(self):
id = "123456.random"
with TestClient(self.app) as client:
with AuthTestClient(self.app) as client:
super().insert_mock_event(id)
events = client.get("/events", params={"event_id": id}).json()
assert len(events) == 1
@ -58,7 +69,7 @@ class TestHttpApp(BaseTestHttp):
now = int(datetime.now().timestamp())
id = "123456.random"
with TestClient(self.app) as client:
with AuthTestClient(self.app) as client:
super().insert_mock_event(id, now, now + 1)
events = client.get(
"/events", params={"max_length": 1, "min_length": 1}
@ -69,7 +80,7 @@ class TestHttpApp(BaseTestHttp):
def test_get_event_list_no_match_max_length(self):
now = int(datetime.now().timestamp())
with TestClient(self.app) as client:
with AuthTestClient(self.app) as client:
id = "123456.random"
super().insert_mock_event(id, now, now + 2)
events = client.get("/events", params={"max_length": 1}).json()
@ -78,7 +89,7 @@ class TestHttpApp(BaseTestHttp):
def test_get_event_list_no_match_min_length(self):
now = int(datetime.now().timestamp())
with TestClient(self.app) as client:
with AuthTestClient(self.app) as client:
id = "123456.random"
super().insert_mock_event(id, now, now + 2)
events = client.get("/events", params={"min_length": 3}).json()
@ -88,7 +99,7 @@ class TestHttpApp(BaseTestHttp):
id = "123456.random"
id2 = "54321.random"
with TestClient(self.app) as client:
with AuthTestClient(self.app) as client:
super().insert_mock_event(id)
events = client.get("/events").json()
assert len(events) == 1
@ -108,14 +119,14 @@ class TestHttpApp(BaseTestHttp):
def test_get_event_list_no_match_has_clip(self):
now = int(datetime.now().timestamp())
with TestClient(self.app) as client:
with AuthTestClient(self.app) as client:
id = "123456.random"
super().insert_mock_event(id, now, now + 2)
events = client.get("/events", params={"has_clip": 0}).json()
assert len(events) == 0
def test_get_event_list_has_clip(self):
with TestClient(self.app) as client:
with AuthTestClient(self.app) as client:
id = "123456.random"
super().insert_mock_event(id, has_clip=True)
events = client.get("/events", params={"has_clip": 1}).json()
@ -123,7 +134,7 @@ class TestHttpApp(BaseTestHttp):
assert events[0]["id"] == id
def test_get_event_list_sort_score(self):
with TestClient(self.app) as client:
with AuthTestClient(self.app) as client:
id = "123456.random"
id2 = "54321.random"
super().insert_mock_event(id, top_score=37, score=37, data={"score": 50})
@ -141,7 +152,7 @@ class TestHttpApp(BaseTestHttp):
def test_get_event_list_sort_start_time(self):
now = int(datetime.now().timestamp())
with TestClient(self.app) as client:
with AuthTestClient(self.app) as client:
id = "123456.random"
id2 = "54321.random"
super().insert_mock_event(id, start_time=now + 3)
@ -159,7 +170,7 @@ class TestHttpApp(BaseTestHttp):
def test_get_good_event(self):
id = "123456.random"
with TestClient(self.app) as client:
with AuthTestClient(self.app) as client:
super().insert_mock_event(id)
event = client.get(f"/events/{id}").json()
@ -171,7 +182,7 @@ class TestHttpApp(BaseTestHttp):
id = "123456.random"
bad_id = "654321.other"
with TestClient(self.app) as client:
with AuthTestClient(self.app) as client:
super().insert_mock_event(id)
event_response = client.get(f"/events/{bad_id}")
assert event_response.status_code == 404
@ -180,7 +191,7 @@ class TestHttpApp(BaseTestHttp):
def test_delete_event(self):
id = "123456.random"
with TestClient(self.app) as client:
with AuthTestClient(self.app) as client:
super().insert_mock_event(id)
event = client.get(f"/events/{id}").json()
assert event
@ -193,7 +204,7 @@ class TestHttpApp(BaseTestHttp):
def test_event_retention(self):
id = "123456.random"
with TestClient(self.app) as client:
with AuthTestClient(self.app) as client:
super().insert_mock_event(id)
client.post(f"/events/{id}/retain", headers={"remote-role": "admin"})
event = client.get(f"/events/{id}").json()
@ -212,12 +223,11 @@ class TestHttpApp(BaseTestHttp):
morning = 1656590400 # 06/30/2022 6 am (GMT)
evening = 1656633600 # 06/30/2022 6 pm (GMT)
with TestClient(self.app) as client:
with AuthTestClient(self.app) as client:
super().insert_mock_event(morning_id, morning)
super().insert_mock_event(evening_id, evening)
# both events come back
events = client.get("/events").json()
print("events!!!", events)
assert events
assert len(events) == 2
# morning event is excluded
@ -248,7 +258,7 @@ class TestHttpApp(BaseTestHttp):
mock_event_updater.publish.side_effect = update_event
with TestClient(app) as client:
with AuthTestClient(app) as client:
super().insert_mock_event(id)
new_sub_label_response = client.post(
f"/events/{id}/sub_label",
@ -285,7 +295,7 @@ class TestHttpApp(BaseTestHttp):
mock_event_updater.publish.side_effect = update_event
with TestClient(app) as client:
with AuthTestClient(app) as client:
super().insert_mock_event(id)
client.post(
f"/events/{id}/sub_label",
@ -301,7 +311,7 @@ class TestHttpApp(BaseTestHttp):
####################################################################################################################
def test_get_metrics(self):
"""ensure correct prometheus metrics api response"""
with TestClient(self.app) as client:
with AuthTestClient(self.app) as client:
ts_start = datetime.now().timestamp()
ts_end = ts_start + 30
_insert_mock_event(

60
frigate/test/http_api/test_http_media.py
View File

@ -1,14 +1,13 @@
"""Unit tests for recordings/media API endpoints."""
from datetime import datetime, timezone
from typing import Any
import pytz
from fastapi.testclient import TestClient
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 BaseTestHttp
from frigate.test.http_api.base_http_test import AuthTestClient, BaseTestHttp
class TestHttpMedia(BaseTestHttp):
@ -19,15 +18,26 @@ class TestHttpMedia(BaseTestHttp):
super().setUp([Recordings])
self.app = super().create_app()
# Mock auth to bypass camera access for tests
async def mock_get_current_user(request: Any):
return {"username": "test_user", "role": "admin"}
# 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
self.app.dependency_overrides[get_allowed_cameras_for_filter] = lambda: [
"front_door",
"back_door",
]
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."""
@ -52,7 +62,7 @@ class TestHttpMedia(BaseTestHttp):
# March 11, 2024 at 12:00 PM EDT (after DST)
march_11_noon = tz.localize(datetime(2024, 3, 11, 12, 0, 0)).timestamp()
with TestClient(self.app) as client:
with AuthTestClient(self.app) as client:
# Insert recordings for each day
Recordings.insert(
id="recording_march_9",
@ -128,7 +138,7 @@ class TestHttpMedia(BaseTestHttp):
# November 4, 2024 at 12:00 PM EST (after DST)
nov_4_noon = tz.localize(datetime(2024, 11, 4, 12, 0, 0)).timestamp()
with TestClient(self.app) as client:
with AuthTestClient(self.app) as client:
# Insert recordings for each day
Recordings.insert(
id="recording_nov_2",
@ -195,7 +205,15 @@ class TestHttpMedia(BaseTestHttp):
# 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 TestClient(self.app) as client:
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",
@ -236,6 +254,14 @@ class TestHttpMedia(BaseTestHttp):
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.
@ -250,7 +276,7 @@ class TestHttpMedia(BaseTestHttp):
# 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 TestClient(self.app) as client:
with AuthTestClient(self.app) as client:
Recordings.insert(
id="recording_during_transition",
path="/media/recordings/transition.mp4",
@ -283,7 +309,7 @@ class TestHttpMedia(BaseTestHttp):
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 TestClient(self.app) as client:
with AuthTestClient(self.app) as client:
Recordings.insert(
id="recording_march_9_utc",
path="/media/recordings/march_9_utc.mp4",
@ -325,7 +351,7 @@ class TestHttpMedia(BaseTestHttp):
"""
Test recordings summary when no recordings exist.
"""
with TestClient(self.app) as client:
with AuthTestClient(self.app) as client:
response = client.get(
"/recordings/summary",
params={"timezone": "America/New_York", "cameras": "all"},
@ -342,7 +368,7 @@ class TestHttpMedia(BaseTestHttp):
tz = pytz.timezone("America/New_York")
march_10_noon = tz.localize(datetime(2024, 3, 10, 12, 0, 0)).timestamp()
with TestClient(self.app) as client:
with AuthTestClient(self.app) as client:
# Insert recordings for both cameras
Recordings.insert(
id="front_recording",

88
frigate/test/http_api/test_http_review.py
View File

@ -1,12 +1,12 @@
from datetime import datetime, timedelta
from fastapi.testclient import TestClient
from fastapi import Request
from peewee import DoesNotExist
from frigate.api.auth import get_allowed_cameras_for_filter, get_current_user
from frigate.models import Event, Recordings, ReviewSegment, UserReviewStatus
from frigate.review.types import SeverityEnum
from frigate.test.http_api.base_http_test import BaseTestHttp
from frigate.test.http_api.base_http_test import AuthTestClient, BaseTestHttp
class TestHttpReview(BaseTestHttp):
@ -16,14 +16,26 @@ class TestHttpReview(BaseTestHttp):
self.user_id = "admin"
# Mock get_current_user for all tests
async def mock_get_current_user():
return {"username": self.user_id, "role": "admin"}
# This mock uses headers set by AuthTestClient
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
self.app.dependency_overrides[get_allowed_cameras_for_filter] = lambda: [
"front_door"
]
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):
self.app.dependency_overrides.clear()
@ -57,7 +69,7 @@ class TestHttpReview(BaseTestHttp):
but ends after is included in the results."""
now = datetime.now().timestamp()
with TestClient(self.app) as client:
with AuthTestClient(self.app) as client:
super().insert_mock_review_segment("123456.random", now, now + 2)
response = client.get("/review")
assert response.status_code == 200
@ -67,7 +79,7 @@ class TestHttpReview(BaseTestHttp):
def test_get_review_no_filters(self):
now = datetime.now().timestamp()
with TestClient(self.app) as client:
with AuthTestClient(self.app) as client:
id = "123456.random"
super().insert_mock_review_segment(id, now - 2, now - 1)
response = client.get("/review")
@ -81,7 +93,7 @@ class TestHttpReview(BaseTestHttp):
"""Test that review items outside the range are not returned."""
now = datetime.now().timestamp()
with TestClient(self.app) as client:
with AuthTestClient(self.app) as client:
id = "123456.random"
super().insert_mock_review_segment(id, now - 2, now - 1)
super().insert_mock_review_segment(f"{id}2", now + 4, now + 5)
@ -97,7 +109,7 @@ class TestHttpReview(BaseTestHttp):
def test_get_review_with_time_filter(self):
now = datetime.now().timestamp()
with TestClient(self.app) as client:
with AuthTestClient(self.app) as client:
id = "123456.random"
super().insert_mock_review_segment(id, now, now + 2)
params = {
@ -113,7 +125,7 @@ class TestHttpReview(BaseTestHttp):
def test_get_review_with_limit_filter(self):
now = datetime.now().timestamp()
with TestClient(self.app) as client:
with AuthTestClient(self.app) as client:
id = "123456.random"
id2 = "654321.random"
super().insert_mock_review_segment(id, now, now + 2)
@ -132,7 +144,7 @@ class TestHttpReview(BaseTestHttp):
def test_get_review_with_severity_filters_no_matches(self):
now = datetime.now().timestamp()
with TestClient(self.app) as client:
with AuthTestClient(self.app) as client:
id = "123456.random"
super().insert_mock_review_segment(id, now, now + 2, SeverityEnum.detection)
params = {
@ -149,7 +161,7 @@ class TestHttpReview(BaseTestHttp):
def test_get_review_with_severity_filters(self):
now = datetime.now().timestamp()
with TestClient(self.app) as client:
with AuthTestClient(self.app) as client:
id = "123456.random"
super().insert_mock_review_segment(id, now, now + 2, SeverityEnum.detection)
params = {
@ -165,7 +177,7 @@ class TestHttpReview(BaseTestHttp):
def test_get_review_with_all_filters(self):
now = datetime.now().timestamp()
with TestClient(self.app) as client:
with AuthTestClient(self.app) as client:
id = "123456.random"
super().insert_mock_review_segment(id, now, now + 2)
params = {
@ -188,7 +200,7 @@ class TestHttpReview(BaseTestHttp):
################################### GET /review/summary Endpoint #################################################
####################################################################################################################
def test_get_review_summary_all_filters(self):
with TestClient(self.app) as client:
with AuthTestClient(self.app) as client:
super().insert_mock_review_segment("123456.random")
params = {
"cameras": "front_door",
@ -219,7 +231,7 @@ class TestHttpReview(BaseTestHttp):
self.assertEqual(response_json, expected_response)
def test_get_review_summary_no_filters(self):
with TestClient(self.app) as client:
with AuthTestClient(self.app) as client:
super().insert_mock_review_segment("123456.random")
response = client.get("/review/summary")
assert response.status_code == 200
@ -247,7 +259,7 @@ class TestHttpReview(BaseTestHttp):
now = datetime.now()
five_days_ago = datetime.today() - timedelta(days=5)
with TestClient(self.app) as client:
with AuthTestClient(self.app) as client:
super().insert_mock_review_segment(
"123456.random", now.timestamp() - 2, now.timestamp() - 1
)
@ -291,7 +303,7 @@ class TestHttpReview(BaseTestHttp):
now = datetime.now()
five_days_ago = datetime.today() - timedelta(days=5)
with TestClient(self.app) as client:
with AuthTestClient(self.app) as client:
super().insert_mock_review_segment("123456.random", now.timestamp())
five_days_ago_ts = five_days_ago.timestamp()
for i in range(20):
@ -342,7 +354,7 @@ class TestHttpReview(BaseTestHttp):
def test_get_review_summary_multiple_in_same_day_with_reviewed(self):
five_days_ago = datetime.today() - timedelta(days=5)
with TestClient(self.app) as client:
with AuthTestClient(self.app) as client:
five_days_ago_ts = five_days_ago.timestamp()
for i in range(10):
id = f"123456_{i}.random_alert_not_reviewed"
@ -393,14 +405,14 @@ class TestHttpReview(BaseTestHttp):
####################################################################################################################
def test_post_reviews_viewed_no_body(self):
with TestClient(self.app) as client:
with AuthTestClient(self.app) as client:
super().insert_mock_review_segment("123456.random")
response = client.post("/reviews/viewed")
# Missing ids
assert response.status_code == 422
def test_post_reviews_viewed_no_body_ids(self):
with TestClient(self.app) as client:
with AuthTestClient(self.app) as client:
super().insert_mock_review_segment("123456.random")
body = {"ids": [""]}
response = client.post("/reviews/viewed", json=body)
@ -408,7 +420,7 @@ class TestHttpReview(BaseTestHttp):
assert response.status_code == 422
def test_post_reviews_viewed_non_existent_id(self):
with TestClient(self.app) as client:
with AuthTestClient(self.app) as client:
id = "123456.random"
super().insert_mock_review_segment(id)
body = {"ids": ["1"]}
@ -425,7 +437,7 @@ class TestHttpReview(BaseTestHttp):
)
def test_post_reviews_viewed(self):
with TestClient(self.app) as client:
with AuthTestClient(self.app) as client:
id = "123456.random"
super().insert_mock_review_segment(id)
body = {"ids": [id]}
@ -445,14 +457,14 @@ class TestHttpReview(BaseTestHttp):
################################### POST reviews/delete Endpoint ################################################
####################################################################################################################
def test_post_reviews_delete_no_body(self):
with TestClient(self.app) as client:
with AuthTestClient(self.app) as client:
super().insert_mock_review_segment("123456.random")
response = client.post("/reviews/delete", headers={"remote-role": "admin"})
# Missing ids
assert response.status_code == 422
def test_post_reviews_delete_no_body_ids(self):
with TestClient(self.app) as client:
with AuthTestClient(self.app) as client:
super().insert_mock_review_segment("123456.random")
body = {"ids": [""]}
response = client.post(
@ -462,7 +474,7 @@ class TestHttpReview(BaseTestHttp):
assert response.status_code == 422
def test_post_reviews_delete_non_existent_id(self):
with TestClient(self.app) as client:
with AuthTestClient(self.app) as client:
id = "123456.random"
super().insert_mock_review_segment(id)
body = {"ids": ["1"]}
@ -479,7 +491,7 @@ class TestHttpReview(BaseTestHttp):
assert review_ids_in_db_after[0].id == id
def test_post_reviews_delete(self):
with TestClient(self.app) as client:
with AuthTestClient(self.app) as client:
id = "123456.random"
super().insert_mock_review_segment(id)
body = {"ids": [id]}
@ -495,7 +507,7 @@ class TestHttpReview(BaseTestHttp):
assert len(review_ids_in_db_after) == 0
def test_post_reviews_delete_many(self):
with TestClient(self.app) as client:
with AuthTestClient(self.app) as client:
ids = ["123456.random", "654321.random"]
for id in ids:
super().insert_mock_review_segment(id)
@ -527,7 +539,7 @@ class TestHttpReview(BaseTestHttp):
def test_review_activity_motion_no_data_for_time_range(self):
now = datetime.now().timestamp()
with TestClient(self.app) as client:
with AuthTestClient(self.app) as client:
params = {
"after": now,
"before": now + 3,
@ -540,7 +552,7 @@ class TestHttpReview(BaseTestHttp):
def test_review_activity_motion(self):
now = int(datetime.now().timestamp())
with TestClient(self.app) as client:
with AuthTestClient(self.app) as client:
one_m = int((datetime.now() + timedelta(minutes=1)).timestamp())
id = "123456.random"
id2 = "123451.random"
@ -573,7 +585,7 @@ class TestHttpReview(BaseTestHttp):
################################### GET /review/event/{event_id} Endpoint #######################################
####################################################################################################################
def test_review_event_not_found(self):
with TestClient(self.app) as client:
with AuthTestClient(self.app) as client:
response = client.get("/review/event/123456.random")
assert response.status_code == 404
response_json = response.json()
@ -585,7 +597,7 @@ class TestHttpReview(BaseTestHttp):
def test_review_event_not_found_in_data(self):
now = datetime.now().timestamp()
with TestClient(self.app) as client:
with AuthTestClient(self.app) as client:
id = "123456.random"
super().insert_mock_review_segment(id, now + 1, now + 2)
response = client.get(f"/review/event/{id}")
@ -599,7 +611,7 @@ class TestHttpReview(BaseTestHttp):
def test_review_get_specific_event(self):
now = datetime.now().timestamp()
with TestClient(self.app) as client:
with AuthTestClient(self.app) as client:
event_id = "123456.event.random"
super().insert_mock_event(event_id)
review_id = "123456.review.random"
@ -626,7 +638,7 @@ class TestHttpReview(BaseTestHttp):
################################### GET /review/{review_id} Endpoint #######################################
####################################################################################################################
def test_review_not_found(self):
with TestClient(self.app) as client:
with AuthTestClient(self.app) as client:
response = client.get("/review/123456.random")
assert response.status_code == 404
response_json = response.json()
@ -638,7 +650,7 @@ class TestHttpReview(BaseTestHttp):
def test_get_review(self):
now = datetime.now().timestamp()
with TestClient(self.app) as client:
with AuthTestClient(self.app) as client:
review_id = "123456.review.random"
super().insert_mock_review_segment(review_id, now + 1, now + 2)
response = client.get(f"/review/{review_id}")
@ -662,7 +674,7 @@ class TestHttpReview(BaseTestHttp):
####################################################################################################################
def test_delete_review_viewed_review_not_found(self):
with TestClient(self.app) as client:
with AuthTestClient(self.app) as client:
review_id = "123456.random"
response = client.delete(f"/review/{review_id}/viewed")
assert response.status_code == 404
@ -675,7 +687,7 @@ class TestHttpReview(BaseTestHttp):
def test_delete_review_viewed(self):
now = datetime.now().timestamp()
with TestClient(self.app) as client:
with AuthTestClient(self.app) as client:
review_id = "123456.review.random"
super().insert_mock_review_segment(review_id, now + 1, now + 2)
self._insert_user_review_status(review_id, reviewed=True)