From c93f53d84c4cdf04f035ef189327d2a46f11337a Mon Sep 17 00:00:00 2001 From: Josh Hawkins <32435876+hawkeye217@users.noreply.github.com> Date: Wed, 10 Sep 2025 14:50:58 -0500 Subject: [PATCH] camera access tests --- .../test/http_api/test_http_camera_access.py | 169 ++++++++++++++++++ 1 file changed, 169 insertions(+) create mode 100644 frigate/test/http_api/test_http_camera_access.py diff --git a/frigate/test/http_api/test_http_camera_access.py b/frigate/test/http_api/test_http_camera_access.py new file mode 100644 index 000000000..db5446bff --- /dev/null +++ b/frigate/test/http_api/test_http_camera_access.py @@ -0,0 +1,169 @@ +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 + + +class TestCameraAccessEventReview(BaseTestHttp): + def setUp(self): + 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"} + + self.app.dependency_overrides[get_current_user] = mock_get_current_user + + def tearDown(self): + self.app.dependency_overrides.clear() + super().tearDown() + + def test_event_camera_access(self): + 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: + 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: + resp = client.get("/events") + assert resp.status_code == 200 + ids = [e["id"] for e in resp.json()] + assert "event1" in ids and "event2" in ids + + def test_review_camera_access(self): + 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: + 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: + resp = client.get("/review") + assert resp.status_code == 200 + ids = [r["id"] for r in resp.json()] + assert "rev1" in ids and "rev2" in ids + + def test_event_single_access(self): + super().insert_mock_event("event1", camera="front_door") + + # Allowed + async def mock_require_allowed(camera: str, request: Request = None): + if camera == "front_door": + return + 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: + resp = client.get("/events/event1") + assert resp.status_code == 200 + assert resp.json()["id"] == "event1" + + # Disallowed + async def mock_require_disallowed(camera: str, request: Request = None): + 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: + resp = client.get("/events/event1") + assert resp.status_code == 403 + + def test_review_single_access(self): + super().insert_mock_review_segment("rev1", camera="front_door") + + # Allowed + async def mock_require_allowed(camera: str, request: Request = None): + if camera == "front_door": + return + 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: + resp = client.get("/review/rev1") + assert resp.status_code == 200 + assert resp.json()["id"] == "rev1" + + # Disallowed + async def mock_require_disallowed(camera: str, request: Request = None): + 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: + resp = client.get("/review/rev1") + assert resp.status_code == 403 + + def test_event_search_access(self): + 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: + 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: + resp = client.get("/events", params={"cameras": "all"}) + assert resp.status_code == 200 + ids = [e["id"] for e in resp.json()] + assert "event1" in ids and "event2" in ids + + def test_event_summary_access(self): + 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: + 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: + resp = client.get("/events/summary") + summary_list = resp.json() + assert len(summary_list) == 2