Enhance user roles to limit camera access (#20024)

* update config for roles and add validator * ensure admin and viewer are never overridden * add class method to user to retrieve all allowed cameras * enforce config roles in auth api endpoints * add camera access api dependency functions * protect review endpoints * protect preview endpoints * rename param name for better fastapi injection matching * remove unneeded * protect export endpoints * protect event endpoints * protect media endpoints * update auth hook for allowed cameras * update default app view * ensure anonymous user always returns all cameras * limit cameras in explore * cameras is already a list * limit cameras in review/history * limit cameras in live view * limit cameras in camera groups * only show face library and classification in sidebar for admin * remove check in delete reviews since admin role is required, no need to check camera access. fixes failing test * pass request with camera access for tests * more async * camera access tests * fix proxy auth tests * allowed cameras for review tests * combine event tests and refactor for camera access * fix post validation for roles * don't limit roles in create user dialog * fix triggers endpoints no need to run require camera access dep since the required role is admin * fix type * create and edit role dialogs * delete role dialog * fix role change dialog * update settings view for roles * i18n changes * minor spacing tweaks * docs * use badges and camera name label component * clarify docs * display all cameras badge for admin and viewer * i18n fix * use validator to prevent reserved and empty roles from being assigned * split users and roles into separate tabs in settings * tweak docs * clarify docs * change icon * don't memoize roles always recalculate on component render
2026-08-02 17:12:16 +03:00 · 2025-09-12 05:19:29 -06:00
parent ba650af6f2
commit ed1e3a7c9a
41 changed files with 2286 additions and 739 deletions
@@ -112,7 +112,7 @@ class BaseTestHttp(unittest.TestCase):
        except OSError:
            pass

-    def create_app(self, stats=None):
+    def create_app(self, stats=None, event_metadata_publisher=None):
        return create_fastapi_app(
            FrigateConfig(**self.minimal_config),
            self.db,
@@ -121,7 +121,7 @@ class BaseTestHttp(unittest.TestCase):
            None,
            None,
            stats,
-            None,
+            event_metadata_publisher,
            None,
        )

@@ -134,12 +134,13 @@ class BaseTestHttp(unittest.TestCase):
        top_score: int = 100,
        score: int = 0,
        data: Json = {},
+        camera: str = "front_door",
    ) -> Event:
        """Inserts a basic event model with a given id."""
        return Event.insert(
            id=id,
            label="Mock",
-            camera="front_door",
+            camera=camera,
            start_time=start_time,
            end_time=end_time,
            top_score=top_score,
@@ -158,15 +159,23 @@ class BaseTestHttp(unittest.TestCase):
    def insert_mock_review_segment(
        self,
        id: str,
-        start_time: float = datetime.datetime.now().timestamp(),
-        end_time: float = datetime.datetime.now().timestamp() + 20,
+        start_time: float | None = None,
+        end_time: float | None = None,
        severity: SeverityEnum = SeverityEnum.alert,
-        data: Json = {},
+        data: dict | None = None,
+        camera: str = "front_door",
    ) -> ReviewSegment:
        """Inserts a review segment model with a given id."""
+        if start_time is None:
+            start_time = datetime.datetime.now().timestamp()
+        if end_time is None:
+            end_time = start_time + 20
+        if data is None:
+            data = {}
+
        return ReviewSegment.insert(
            id=id,
-            camera="front_door",
+            camera=camera,
            start_time=start_time,
            end_time=end_time,
            severity=severity,
@@ -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
@@ -1,16 +1,34 @@
 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.models import Event, Recordings, ReviewSegment
+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.test.http_api.base_http_test import BaseTestHttp


 class TestHttpApp(BaseTestHttp):
    def setUp(self):
-        super().setUp([Event, Recordings, ReviewSegment])
+        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"}
+
+        self.app.dependency_overrides[get_current_user] = mock_get_current_user
+        self.app.dependency_overrides[get_allowed_cameras_for_filter] = lambda: [
+            "front_door"
+        ]
+
+    def tearDown(self):
+        self.app.dependency_overrides.clear()
+        super().tearDown()
+
    ####################################################################################################################
    ###################################  GET /events Endpoint   #########################################################
    ####################################################################################################################
@@ -135,3 +153,143 @@ class TestHttpApp(BaseTestHttp):
            assert len(events) == 2
            assert events[0]["id"] == id
            assert events[1]["id"] == id2
+
+    def test_get_good_event(self):
+        id = "123456.random"
+
+        with TestClient(self.app) as client:
+            super().insert_mock_event(id)
+            event = client.get(f"/events/{id}").json()
+
+        assert event
+        assert event["id"] == id
+        assert event["id"] == model_to_dict(Event.get(Event.id == id))["id"]
+
+    def test_get_bad_event(self):
+        id = "123456.random"
+        bad_id = "654321.other"
+
+        with TestClient(self.app) as client:
+            super().insert_mock_event(id)
+            event_response = client.get(f"/events/{bad_id}")
+            assert event_response.status_code == 404
+            assert event_response.json() == "Event not found"
+
+    def test_delete_event(self):
+        id = "123456.random"
+
+        with TestClient(self.app) as client:
+            super().insert_mock_event(id)
+            event = client.get(f"/events/{id}").json()
+            assert event
+            assert event["id"] == id
+            response = client.delete(f"/events/{id}", headers={"remote-role": "admin"})
+            assert response.status_code == 200
+            event_after_delete = client.get(f"/events/{id}")
+            assert event_after_delete.status_code == 404
+
+    def test_event_retention(self):
+        id = "123456.random"
+
+        with TestClient(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()
+            assert event
+            assert event["id"] == id
+            assert event["retain_indefinitely"] is True
+            client.delete(f"/events/{id}/retain", headers={"remote-role": "admin"})
+            event = client.get(f"/events/{id}").json()
+            assert event
+            assert event["id"] == id
+            assert event["retain_indefinitely"] is False
+
+    def test_event_time_filtering(self):
+        morning_id = "123456.random"
+        evening_id = "654321.random"
+        morning = 1656590400  # 06/30/2022 6 am (GMT)
+        evening = 1656633600  # 06/30/2022 6 pm (GMT)
+
+        with TestClient(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
+            events = client.get(
+                "/events",
+                params={"time_range": "07:00,24:00"},
+            ).json()
+            assert events
+            assert len(events) == 1
+            # evening event is excluded
+            events = client.get(
+                "/events",
+                params={"time_range": "00:00,18:00"},
+            ).json()
+            assert events
+            assert len(events) == 1
+
+    def test_set_delete_sub_label(self):
+        mock_event_updater = Mock(spec=EventMetadataPublisher)
+        app = super().create_app(event_metadata_publisher=mock_event_updater)
+        id = "123456.random"
+        sub_label = "sub"
+
+        def update_event(payload: Any, topic: str):
+            event = Event.get(id=id)
+            event.sub_label = payload[1]
+            event.save()
+
+        mock_event_updater.publish.side_effect = update_event
+
+        with TestClient(app) as client:
+            super().insert_mock_event(id)
+            new_sub_label_response = client.post(
+                f"/events/{id}/sub_label",
+                json={"subLabel": sub_label},
+                headers={"remote-role": "admin"},
+            )
+            assert new_sub_label_response.status_code == 200
+            event = client.get(f"/events/{id}").json()
+            assert event
+            assert event["id"] == id
+            assert event["sub_label"] == sub_label
+            empty_sub_label_response = client.post(
+                f"/events/{id}/sub_label",
+                json={"subLabel": ""},
+                headers={"remote-role": "admin"},
+            )
+            assert empty_sub_label_response.status_code == 200
+            event = client.get(f"/events/{id}").json()
+            assert event
+            assert event["id"] == id
+            assert event["sub_label"] == None
+
+    def test_sub_label_list(self):
+        mock_event_updater = Mock(spec=EventMetadataPublisher)
+        app = super().create_app(event_metadata_publisher=mock_event_updater)
+        app.event_metadata_publisher = mock_event_updater
+        id = "123456.random"
+        sub_label = "sub"
+
+        def update_event(payload: Any, _: str):
+            event = Event.get(id=id)
+            event.sub_label = payload[1]
+            event.save()
+
+        mock_event_updater.publish.side_effect = update_event
+
+        with TestClient(app) as client:
+            super().insert_mock_event(id)
+            client.post(
+                f"/events/{id}/sub_label",
+                json={"subLabel": sub_label},
+                headers={"remote-role": "admin"},
+            )
+            sub_labels = client.get("/sub_labels").json()
+            assert sub_labels
+            assert sub_labels == [sub_label]
@@ -3,7 +3,7 @@ from datetime import datetime, timedelta
 from fastapi.testclient import TestClient
 from peewee import DoesNotExist

-from frigate.api.auth import get_current_user
+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
@@ -21,6 +21,10 @@ class TestHttpReview(BaseTestHttp):

        self.app.dependency_overrides[get_current_user] = mock_get_current_user

+        self.app.dependency_overrides[get_allowed_cameras_for_filter] = lambda: [
+            "front_door"
+        ]
+
    def tearDown(self):
        self.app.dependency_overrides.clear()
        super().tearDown()
@@ -1,329 +0,0 @@
-import datetime
-import logging
-import os
-import unittest
-from typing import Any
-from unittest.mock import Mock
-
-from fastapi.testclient import TestClient
-from peewee_migrate import Router
-from playhouse.shortcuts import model_to_dict
-from playhouse.sqlite_ext import SqliteExtDatabase
-from playhouse.sqliteq import SqliteQueueDatabase
-
-from frigate.api.fastapi_app import create_fastapi_app
-from frigate.comms.event_metadata_updater import EventMetadataPublisher
-from frigate.config import FrigateConfig
-from frigate.const import BASE_DIR, CACHE_DIR
-from frigate.models import Event, Recordings, Timeline
-from frigate.test.const import TEST_DB, TEST_DB_CLEANUPS
-
-
-class TestHttp(unittest.TestCase):
-    def setUp(self):
-        # setup clean database for each test run
-        migrate_db = SqliteExtDatabase("test.db")
-        del logging.getLogger("peewee_migrate").handlers[:]
-        router = Router(migrate_db)
-        router.run()
-        migrate_db.close()
-        self.db = SqliteQueueDatabase(TEST_DB)
-        models = [Event, Recordings, Timeline]
-        self.db.bind(models)
-
-        self.minimal_config = {
-            "mqtt": {"host": "mqtt"},
-            "cameras": {
-                "front_door": {
-                    "ffmpeg": {
-                        "inputs": [
-                            {"path": "rtsp://10.0.0.1:554/video", "roles": ["detect"]}
-                        ]
-                    },
-                    "detect": {
-                        "height": 1080,
-                        "width": 1920,
-                        "fps": 5,
-                    },
-                }
-            },
-        }
-        self.test_stats = {
-            "camera_fps": 5.0,
-            "process_fps": 5.0,
-            "skipped_fps": 0.0,
-            "detection_fps": 13.7,
-            "detectors": {
-                "cpu1": {
-                    "detection_start": 0.0,
-                    "inference_speed": 91.43,
-                    "pid": 42,
-                },
-                "cpu2": {
-                    "detection_start": 0.0,
-                    "inference_speed": 84.99,
-                    "pid": 44,
-                },
-            },
-            "front_door": {
-                "camera_fps": 0.0,
-                "capture_pid": 53,
-                "detection_fps": 0.0,
-                "pid": 52,
-                "process_fps": 0.0,
-                "skipped_fps": 0.0,
-            },
-            "service": {
-                "storage": {
-                    "/dev/shm": {
-                        "free": 50.5,
-                        "mount_type": "tmpfs",
-                        "total": 67.1,
-                        "used": 16.6,
-                    },
-                    os.path.join(BASE_DIR, "clips"): {
-                        "free": 42429.9,
-                        "mount_type": "ext4",
-                        "total": 244529.7,
-                        "used": 189607.0,
-                    },
-                    os.path.join(BASE_DIR, "recordings"): {
-                        "free": 0.2,
-                        "mount_type": "ext4",
-                        "total": 8.0,
-                        "used": 7.8,
-                    },
-                    CACHE_DIR: {
-                        "free": 976.8,
-                        "mount_type": "tmpfs",
-                        "total": 1000.0,
-                        "used": 23.2,
-                    },
-                },
-                "uptime": 101113,
-                "version": "0.10.1",
-                "latest_version": "0.11",
-            },
-        }
-
-    def tearDown(self):
-        if not self.db.is_closed():
-            self.db.close()
-
-        try:
-            for file in TEST_DB_CLEANUPS:
-                os.remove(file)
-        except OSError:
-            pass
-
-    def __init_app(self, updater: Any | None = None) -> Any:
-        return create_fastapi_app(
-            FrigateConfig(**self.minimal_config),
-            self.db,
-            None,
-            None,
-            None,
-            None,
-            None,
-            updater,
-            None,
-        )
-
-    def test_get_good_event(self):
-        app = self.__init_app()
-        id = "123456.random"
-
-        with TestClient(app) as client:
-            _insert_mock_event(id)
-            event = client.get(f"/events/{id}").json()
-
-        assert event
-        assert event["id"] == id
-        assert event["id"] == model_to_dict(Event.get(Event.id == id))["id"]
-
-    def test_get_bad_event(self):
-        app = self.__init_app()
-        id = "123456.random"
-        bad_id = "654321.other"
-
-        with TestClient(app) as client:
-            _insert_mock_event(id)
-            event_response = client.get(f"/events/{bad_id}")
-            assert event_response.status_code == 404
-            assert event_response.json() == "Event not found"
-
-    def test_delete_event(self):
-        app = self.__init_app()
-        id = "123456.random"
-
-        with TestClient(app) as client:
-            _insert_mock_event(id)
-            event = client.get(f"/events/{id}").json()
-            assert event
-            assert event["id"] == id
-            client.delete(f"/events/{id}", headers={"remote-role": "admin"})
-            event = client.get(f"/events/{id}").json()
-            assert event == "Event not found"
-
-    def test_event_retention(self):
-        app = self.__init_app()
-        id = "123456.random"
-
-        with TestClient(app) as client:
-            _insert_mock_event(id)
-            client.post(f"/events/{id}/retain", headers={"remote-role": "admin"})
-            event = client.get(f"/events/{id}").json()
-            assert event
-            assert event["id"] == id
-            assert event["retain_indefinitely"] is True
-            client.delete(f"/events/{id}/retain", headers={"remote-role": "admin"})
-            event = client.get(f"/events/{id}").json()
-            assert event
-            assert event["id"] == id
-            assert event["retain_indefinitely"] is False
-
-    def test_event_time_filtering(self):
-        app = self.__init_app()
-        morning_id = "123456.random"
-        evening_id = "654321.random"
-        morning = 1656590400  # 06/30/2022 6 am (GMT)
-        evening = 1656633600  # 06/30/2022 6 pm (GMT)
-
-        with TestClient(app) as client:
-            _insert_mock_event(morning_id, morning)
-            _insert_mock_event(evening_id, evening)
-            # both events come back
-            events = client.get("/events").json()
-            assert events
-            assert len(events) == 2
-            # morning event is excluded
-            events = client.get(
-                "/events",
-                params={"time_range": "07:00,24:00"},
-            ).json()
-            assert events
-            # assert len(events) == 1
-            # evening event is excluded
-            events = client.get(
-                "/events",
-                params={"time_range": "00:00,18:00"},
-            ).json()
-            assert events
-            assert len(events) == 1
-
-    def test_set_delete_sub_label(self):
-        mock_event_updater = Mock(spec=EventMetadataPublisher)
-        app = app = self.__init_app(updater=mock_event_updater)
-        id = "123456.random"
-        sub_label = "sub"
-
-        def update_event(payload: Any, topic: str):
-            event = Event.get(id=id)
-            event.sub_label = payload[1]
-            event.save()
-
-        mock_event_updater.publish.side_effect = update_event
-
-        with TestClient(app) as client:
-            _insert_mock_event(id)
-            new_sub_label_response = client.post(
-                f"/events/{id}/sub_label",
-                json={"subLabel": sub_label},
-                headers={"remote-role": "admin"},
-            )
-            assert new_sub_label_response.status_code == 200
-            event = client.get(f"/events/{id}").json()
-            assert event
-            assert event["id"] == id
-            assert event["sub_label"] == sub_label
-            empty_sub_label_response = client.post(
-                f"/events/{id}/sub_label",
-                json={"subLabel": ""},
-                headers={"remote-role": "admin"},
-            )
-            assert empty_sub_label_response.status_code == 200
-            event = client.get(f"/events/{id}").json()
-            assert event
-            assert event["id"] == id
-            assert event["sub_label"] == None
-
-    def test_sub_label_list(self):
-        mock_event_updater = Mock(spec=EventMetadataPublisher)
-        app = self.__init_app(updater=mock_event_updater)
-        id = "123456.random"
-        sub_label = "sub"
-
-        def update_event(payload: Any, _: str):
-            event = Event.get(id=id)
-            event.sub_label = payload[1]
-            event.save()
-
-        mock_event_updater.publish.side_effect = update_event
-
-        with TestClient(app) as client:
-            _insert_mock_event(id)
-            client.post(
-                f"/events/{id}/sub_label",
-                json={"subLabel": sub_label},
-                headers={"remote-role": "admin"},
-            )
-            sub_labels = client.get("/sub_labels").json()
-            assert sub_labels
-            assert sub_labels == [sub_label]
-
-    def test_config(self):
-        app = self.__init_app()
-
-        with TestClient(app) as client:
-            config = client.get("/config").json()
-            assert config
-            assert config["cameras"]["front_door"]
-
-    def test_recordings(self):
-        app = self.__init_app()
-        id = "123456.random"
-
-        with TestClient(app) as client:
-            _insert_mock_recording(id)
-            response = client.get("/front_door/recordings")
-            assert response.status_code == 200
-            recording = response.json()
-            assert recording
-            assert recording[0]["id"] == id
-
-
-def _insert_mock_event(
-    id: str,
-    start_time: datetime.datetime = datetime.datetime.now().timestamp(),
-) -> Event:
-    """Inserts a basic event model with a given id."""
-    return Event.insert(
-        id=id,
-        label="Mock",
-        camera="front_door",
-        start_time=start_time,
-        end_time=start_time + 20,
-        top_score=100,
-        false_positive=False,
-        zones=list(),
-        thumbnail="",
-        region=[],
-        box=[],
-        area=0,
-        has_clip=True,
-        has_snapshot=True,
-    ).execute()
-
-
-def _insert_mock_recording(id: str) -> Event:
-    """Inserts a basic recording model with a given id."""
-    return Recordings.insert(
-        id=id,
-        camera="front_door",
-        path=f"/recordings/{id}",
-        start_time=datetime.datetime.now().timestamp() - 60,
-        end_time=datetime.datetime.now().timestamp() - 50,
-        duration=10,
-        motion=True,
-        objects=True,
-    ).execute()
@@ -19,61 +19,60 @@ class TestProxyRoleResolution(unittest.TestCase):
                },
            ),
        )
+        self.config_roles = list(["admin", "viewer"])

    def test_role_map_single_group_match(self):
        headers = {"x-remote-role": "group_admin"}
-        role = resolve_role(headers, self.proxy_config)
+        role = resolve_role(headers, self.proxy_config, self.config_roles)
        self.assertEqual(role, "admin")

    def test_role_map_multiple_groups(self):
-        headers = {"x-remote-role": "group_viewer|group_admin"}
-        role = resolve_role(headers, self.proxy_config)
-        # admin should win since VALID_ROLES priority puts it before viewer
+        headers = {"x-remote-role": "group_admin|group_viewer"}
+        role = resolve_role(headers, self.proxy_config, self.config_roles)
        self.assertEqual(role, "admin")

    def test_direct_role_header_with_separator(self):
        config = self.proxy_config
        config.header_map.role_map = None  # disable role_map
-        headers = {"x-remote-role": "viewer|admin"}
-        role = resolve_role(headers, config)
-        # admin should be chosen since it appears in VALID_ROLES
+        headers = {"x-remote-role": "admin|viewer"}
+        role = resolve_role(headers, config, self.config_roles)
        self.assertEqual(role, "admin")

    def test_invalid_role_header(self):
        config = self.proxy_config
        config.header_map.role_map = None
        headers = {"x-remote-role": "notarole"}
-        role = resolve_role(headers, config)
+        role = resolve_role(headers, config, self.config_roles)
        self.assertEqual(role, config.default_role)

    def test_missing_role_header(self):
        headers = {}
-        role = resolve_role(headers, self.proxy_config)
+        role = resolve_role(headers, self.proxy_config, self.config_roles)
        self.assertEqual(role, self.proxy_config.default_role)

    def test_empty_role_header(self):
        headers = {"x-remote-role": ""}
-        role = resolve_role(headers, self.proxy_config)
+        role = resolve_role(headers, self.proxy_config, self.config_roles)
        self.assertEqual(role, self.proxy_config.default_role)

    def test_whitespace_groups(self):
        headers = {"x-remote-role": "   | group_admin |   "}
-        role = resolve_role(headers, self.proxy_config)
+        role = resolve_role(headers, self.proxy_config, self.config_roles)
        self.assertEqual(role, "admin")

    def test_mixed_valid_and_invalid_groups(self):
        headers = {"x-remote-role": "bogus|group_viewer"}
-        role = resolve_role(headers, self.proxy_config)
+        role = resolve_role(headers, self.proxy_config, self.config_roles)
        self.assertEqual(role, "viewer")

    def test_case_insensitive_role_direct(self):
        config = self.proxy_config
        config.header_map.role_map = None
        headers = {"x-remote-role": "AdMiN"}
-        role = resolve_role(headers, config)
+        role = resolve_role(headers, config, self.config_roles)
        self.assertEqual(role, "admin")

    def test_role_map_no_match_falls_back(self):
        headers = {"x-remote-role": "group_unknown"}
-        role = resolve_role(headers, self.proxy_config)
+        role = resolve_role(headers, self.proxy_config, self.config_roles)
        self.assertEqual(role, self.proxy_config.default_role)