from datetime import datetime, timedelta from fastapi.testclient import TestClient from peewee import DoesNotExist from frigate.api.auth import 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 class TestHttpReview(BaseTestHttp): def setUp(self): super().setUp([Event, Recordings, ReviewSegment, UserReviewStatus]) self.app = super().create_app() self.user_id = "admin" # Mock get_current_user for all tests async def mock_get_current_user(): return {"username": self.user_id, "role": "admin"} self.app.dependency_overrides[get_current_user] = mock_get_current_user def tearDown(self): self.app.dependency_overrides.clear() super().tearDown() def _get_reviews(self, ids: list[str]): return list( ReviewSegment.select(ReviewSegment.id) .where(ReviewSegment.id.in_(ids)) .execute() ) def _get_recordings(self, ids: list[str]): return list( Recordings.select(Recordings.id).where(Recordings.id.in_(ids)).execute() ) def _insert_user_review_status(self, review_id: str, reviewed: bool = True): UserReviewStatus.create( user_id=self.user_id, review_segment=ReviewSegment.get(ReviewSegment.id == review_id), has_been_reviewed=reviewed, ) #################################################################################################################### ################################### GET /review Endpoint ######################################################## #################################################################################################################### def test_get_review_that_overlaps_default_period(self): """Test that a review item that starts during the default period but ends after is included in the results.""" now = datetime.now().timestamp() with TestClient(self.app) as client: super().insert_mock_review_segment("123456.random", now, now + 2) response = client.get("/review") assert response.status_code == 200 response_json = response.json() assert len(response_json) == 1 def test_get_review_no_filters(self): now = datetime.now().timestamp() with TestClient(self.app) as client: id = "123456.random" super().insert_mock_review_segment(id, now - 2, now - 1) response = client.get("/review") assert response.status_code == 200 response_json = response.json() assert len(response_json) == 1 assert response_json[0]["id"] == id assert response_json[0]["has_been_reviewed"] == False def test_get_review_with_time_filter_no_matches(self): """Test that review items outside the range are not returned.""" now = datetime.now().timestamp() with TestClient(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) params = { "after": now, "before": now + 3, } response = client.get("/review", params=params) assert response.status_code == 200 response_json = response.json() assert len(response_json) == 0 def test_get_review_with_time_filter(self): now = datetime.now().timestamp() with TestClient(self.app) as client: id = "123456.random" super().insert_mock_review_segment(id, now, now + 2) params = { "after": now - 1, "before": now + 3, } response = client.get("/review", params=params) assert response.status_code == 200 response_json = response.json() assert len(response_json) == 1 assert response_json[0]["id"] == id def test_get_review_with_limit_filter(self): now = datetime.now().timestamp() with TestClient(self.app) as client: id = "123456.random" id2 = "654321.random" super().insert_mock_review_segment(id, now, now + 2) super().insert_mock_review_segment(id2, now + 1, now + 2) params = { "limit": 1, "after": now, "before": now + 3, } response = client.get("/review", params=params) assert response.status_code == 200 response_json = response.json() assert len(response_json) == 1 assert response_json[0]["id"] == id2 def test_get_review_with_severity_filters_no_matches(self): now = datetime.now().timestamp() with TestClient(self.app) as client: id = "123456.random" super().insert_mock_review_segment(id, now, now + 2, SeverityEnum.detection) params = { "severity": "detection", "after": now - 1, "before": now + 3, } response = client.get("/review", params=params) assert response.status_code == 200 response_json = response.json() assert len(response_json) == 1 assert response_json[0]["id"] == id def test_get_review_with_severity_filters(self): now = datetime.now().timestamp() with TestClient(self.app) as client: id = "123456.random" super().insert_mock_review_segment(id, now, now + 2, SeverityEnum.detection) params = { "severity": "alert", "after": now - 1, "before": now + 3, } response = client.get("/review", params=params) assert response.status_code == 200 response_json = response.json() assert len(response_json) == 0 def test_get_review_with_all_filters(self): now = datetime.now().timestamp() with TestClient(self.app) as client: id = "123456.random" super().insert_mock_review_segment(id, now, now + 2) params = { "cameras": "front_door", "labels": "all", "zones": "all", "reviewed": 0, "limit": 1, "severity": "alert", "after": now - 1, "before": now + 3, } response = client.get("/review", params=params) assert response.status_code == 200 response_json = response.json() assert len(response_json) == 1 assert response_json[0]["id"] == id #################################################################################################################### ################################### GET /review/summary Endpoint ################################################# #################################################################################################################### def test_get_review_summary_all_filters(self): with TestClient(self.app) as client: super().insert_mock_review_segment("123456.random") params = { "cameras": "front_door", "labels": "all", "zones": "all", "timezone": "utc", } response = client.get("/review/summary", params=params) assert response.status_code == 200 response_json = response.json() # e.g. '2024-11-24' today_formatted = datetime.today().strftime("%Y-%m-%d") expected_response = { "last24Hours": { "reviewed_alert": 0, "reviewed_detection": 0, "total_alert": 1, "total_detection": 0, }, today_formatted: { "day": today_formatted, "reviewed_alert": 0, "reviewed_detection": 0, "total_alert": 1, "total_detection": 0, }, } self.assertEqual(response_json, expected_response) def test_get_review_summary_no_filters(self): with TestClient(self.app) as client: super().insert_mock_review_segment("123456.random") response = client.get("/review/summary") assert response.status_code == 200 response_json = response.json() # e.g. '2024-11-24' today_formatted = datetime.today().strftime("%Y-%m-%d") expected_response = { "last24Hours": { "reviewed_alert": 0, "reviewed_detection": 0, "total_alert": 1, "total_detection": 0, }, today_formatted: { "day": today_formatted, "reviewed_alert": 0, "reviewed_detection": 0, "total_alert": 1, "total_detection": 0, }, } self.assertEqual(response_json, expected_response) def test_get_review_summary_multiple_days(self): now = datetime.now() five_days_ago = datetime.today() - timedelta(days=5) with TestClient(self.app) as client: super().insert_mock_review_segment( "123456.random", now.timestamp() - 2, now.timestamp() - 1 ) super().insert_mock_review_segment( "654321.random", five_days_ago.timestamp(), five_days_ago.timestamp() + 1, ) response = client.get("/review/summary") assert response.status_code == 200 response_json = response.json() # e.g. '2024-11-24' today_formatted = now.strftime("%Y-%m-%d") # e.g. '2024-11-19' five_days_ago_formatted = five_days_ago.strftime("%Y-%m-%d") expected_response = { "last24Hours": { "reviewed_alert": 0, "reviewed_detection": 0, "total_alert": 1, "total_detection": 0, }, today_formatted: { "day": today_formatted, "reviewed_alert": 0, "reviewed_detection": 0, "total_alert": 1, "total_detection": 0, }, five_days_ago_formatted: { "day": five_days_ago_formatted, "reviewed_alert": 0, "reviewed_detection": 0, "total_alert": 1, "total_detection": 0, }, } self.assertEqual(response_json, expected_response) def test_get_review_summary_multiple_in_same_day(self): now = datetime.now() five_days_ago = datetime.today() - timedelta(days=5) with TestClient(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): super().insert_mock_review_segment( f"123456_{i}.random_alert", five_days_ago_ts, five_days_ago_ts, SeverityEnum.alert, ) for i in range(15): super().insert_mock_review_segment( f"123456_{i}.random_detection", five_days_ago_ts, five_days_ago_ts, SeverityEnum.detection, ) response = client.get("/review/summary") assert response.status_code == 200 response_json = response.json() # e.g. '2024-11-24' today_formatted = now.strftime("%Y-%m-%d") # e.g. '2024-11-19' five_days_ago_formatted = five_days_ago.strftime("%Y-%m-%d") expected_response = { "last24Hours": { "reviewed_alert": 0, "reviewed_detection": 0, "total_alert": 1, "total_detection": 0, }, today_formatted: { "day": today_formatted, "reviewed_alert": 0, "reviewed_detection": 0, "total_alert": 1, "total_detection": 0, }, five_days_ago_formatted: { "day": five_days_ago_formatted, "reviewed_alert": 0, "reviewed_detection": 0, "total_alert": 20, "total_detection": 15, }, } self.assertEqual(response_json, expected_response) 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: five_days_ago_ts = five_days_ago.timestamp() for i in range(10): id = f"123456_{i}.random_alert_not_reviewed" super().insert_mock_review_segment( id, five_days_ago_ts, five_days_ago_ts, SeverityEnum.alert ) for i in range(10): id = f"123456_{i}.random_alert_reviewed" super().insert_mock_review_segment( id, five_days_ago_ts, five_days_ago_ts, SeverityEnum.alert ) self._insert_user_review_status(id, reviewed=True) for i in range(10): id = f"123456_{i}.random_detection_not_reviewed" super().insert_mock_review_segment( id, five_days_ago_ts, five_days_ago_ts, SeverityEnum.detection ) for i in range(5): id = f"123456_{i}.random_detection_reviewed" super().insert_mock_review_segment( id, five_days_ago_ts, five_days_ago_ts, SeverityEnum.detection ) self._insert_user_review_status(id, reviewed=True) response = client.get("/review/summary") assert response.status_code == 200 response_json = response.json() # e.g. '2024-11-19' five_days_ago_formatted = five_days_ago.strftime("%Y-%m-%d") expected_response = { "last24Hours": { "reviewed_alert": None, "reviewed_detection": None, "total_alert": None, "total_detection": None, }, five_days_ago_formatted: { "day": five_days_ago_formatted, "reviewed_alert": 10, "reviewed_detection": 5, "total_alert": 20, "total_detection": 15, }, } self.assertEqual(response_json, expected_response) #################################################################################################################### ################################### POST reviews/viewed Endpoint ################################################ #################################################################################################################### def test_post_reviews_viewed_no_body(self): with TestClient(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: super().insert_mock_review_segment("123456.random") body = {"ids": [""]} response = client.post("/reviews/viewed", json=body) # Missing ids assert response.status_code == 422 def test_post_reviews_viewed_non_existent_id(self): with TestClient(self.app) as client: id = "123456.random" super().insert_mock_review_segment(id) body = {"ids": ["1"]} response = client.post("/reviews/viewed", json=body) assert response.status_code == 200 response = response.json() assert response["success"] == True assert response["message"] == "Reviewed multiple items" # Verify that in DB the review segment was not changed with self.assertRaises(DoesNotExist): UserReviewStatus.get( UserReviewStatus.user_id == self.user_id, UserReviewStatus.review_segment == "1", ) def test_post_reviews_viewed(self): with TestClient(self.app) as client: id = "123456.random" super().insert_mock_review_segment(id) body = {"ids": [id]} response = client.post("/reviews/viewed", json=body) assert response.status_code == 200 response_json = response.json() assert response_json["success"] == True assert response_json["message"] == "Reviewed multiple items" # Verify UserReviewStatus was created user_review = UserReviewStatus.get( UserReviewStatus.user_id == self.user_id, UserReviewStatus.review_segment == id, ) assert user_review.has_been_reviewed == True #################################################################################################################### ################################### POST reviews/delete Endpoint ################################################ #################################################################################################################### def test_post_reviews_delete_no_body(self): with TestClient(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: super().insert_mock_review_segment("123456.random") body = {"ids": [""]} response = client.post( "/reviews/delete", json=body, headers={"remote-role": "admin"} ) # Missing ids assert response.status_code == 422 def test_post_reviews_delete_non_existent_id(self): with TestClient(self.app) as client: id = "123456.random" super().insert_mock_review_segment(id) body = {"ids": ["1"]} response = client.post( "/reviews/delete", json=body, headers={"remote-role": "admin"} ) assert response.status_code == 200 response_json = response.json() assert response_json["success"] == True assert response_json["message"] == "Deleted review items." # Verify that in DB the review segment was not deleted review_ids_in_db_after = self._get_reviews([id]) assert len(review_ids_in_db_after) == 1 assert review_ids_in_db_after[0].id == id def test_post_reviews_delete(self): with TestClient(self.app) as client: id = "123456.random" super().insert_mock_review_segment(id) body = {"ids": [id]} response = client.post( "/reviews/delete", json=body, headers={"remote-role": "admin"} ) assert response.status_code == 200 response_json = response.json() assert response_json["success"] == True assert response_json["message"] == "Deleted review items." # Verify that in DB the review segment was deleted review_ids_in_db_after = self._get_reviews([id]) assert len(review_ids_in_db_after) == 0 def test_post_reviews_delete_many(self): with TestClient(self.app) as client: ids = ["123456.random", "654321.random"] for id in ids: super().insert_mock_review_segment(id) super().insert_mock_recording(id) review_ids_in_db_before = self._get_reviews(ids) recordings_ids_in_db_before = self._get_recordings(ids) assert len(review_ids_in_db_before) == 2 assert len(recordings_ids_in_db_before) == 2 body = {"ids": ids} response = client.post( "/reviews/delete", json=body, headers={"remote-role": "admin"} ) assert response.status_code == 200 response_json = response.json() assert response_json["success"] == True assert response_json["message"] == "Deleted review items." # Verify that in DB all review segments and recordings that were passed were deleted review_ids_in_db_after = self._get_reviews(ids) recording_ids_in_db_after = self._get_recordings(ids) assert len(review_ids_in_db_after) == 0 assert len(recording_ids_in_db_after) == 0 #################################################################################################################### ################################### GET /review/activity/motion Endpoint ######################################## #################################################################################################################### def test_review_activity_motion_no_data_for_time_range(self): now = datetime.now().timestamp() with TestClient(self.app) as client: params = { "after": now, "before": now + 3, } response = client.get("/review/activity/motion", params=params) assert response.status_code == 200 response_json = response.json() assert len(response_json) == 0 def test_review_activity_motion(self): now = int(datetime.now().timestamp()) with TestClient(self.app) as client: one_m = int((datetime.now() + timedelta(minutes=1)).timestamp()) id = "123456.random" id2 = "123451.random" super().insert_mock_recording(id, now + 1, now + 2, motion=101) super().insert_mock_recording(id2, one_m + 1, one_m + 2, motion=200) params = { "after": now, "before": one_m + 3, "scale": 1, } response = client.get("/review/activity/motion", params=params) assert response.status_code == 200 response_json = response.json() assert len(response_json) == 61 self.assertDictEqual( {"motion": 50.5, "camera": "front_door", "start_time": now + 1}, response_json[0], ) for item in response_json[1:-1]: self.assertDictEqual( {"motion": 0.0, "camera": "", "start_time": item["start_time"]}, item, ) self.assertDictEqual( {"motion": 100.0, "camera": "front_door", "start_time": one_m + 1}, response_json[len(response_json) - 1], ) #################################################################################################################### ################################### GET /review/event/{event_id} Endpoint ####################################### #################################################################################################################### def test_review_event_not_found(self): with TestClient(self.app) as client: response = client.get("/review/event/123456.random") assert response.status_code == 404 response_json = response.json() self.assertDictEqual( {"success": False, "message": "Review item not found"}, response_json, ) def test_review_event_not_found_in_data(self): now = datetime.now().timestamp() with TestClient(self.app) as client: id = "123456.random" super().insert_mock_review_segment(id, now + 1, now + 2) response = client.get(f"/review/event/{id}") assert response.status_code == 404 response_json = response.json() self.assertDictEqual( {"success": False, "message": "Review item not found"}, response_json, ) def test_review_get_specific_event(self): now = datetime.now().timestamp() with TestClient(self.app) as client: event_id = "123456.event.random" super().insert_mock_event(event_id) review_id = "123456.review.random" super().insert_mock_review_segment( review_id, now + 1, now + 2, data={"detections": {"event_id": event_id}} ) response = client.get(f"/review/event/{event_id}") assert response.status_code == 200 response_json = response.json() self.assertDictEqual( { "id": review_id, "camera": "front_door", "start_time": now + 1, "end_time": now + 2, "severity": "alert", "thumb_path": "False", "data": {"detections": {"event_id": event_id}}, }, response_json, ) #################################################################################################################### ################################### GET /review/{review_id} Endpoint ####################################### #################################################################################################################### def test_review_not_found(self): with TestClient(self.app) as client: response = client.get("/review/123456.random") assert response.status_code == 404 response_json = response.json() self.assertDictEqual( {"success": False, "message": "Review item not found"}, response_json, ) def test_get_review(self): now = datetime.now().timestamp() with TestClient(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}") assert response.status_code == 200 response_json = response.json() self.assertDictEqual( { "id": review_id, "camera": "front_door", "start_time": now + 1, "end_time": now + 2, "severity": "alert", "thumb_path": "False", "data": {}, }, response_json, ) #################################################################################################################### ################################### DELETE /review/{review_id}/viewed Endpoint ################################## #################################################################################################################### def test_delete_review_viewed_review_not_found(self): with TestClient(self.app) as client: review_id = "123456.random" response = client.delete(f"/review/{review_id}/viewed") assert response.status_code == 404 response_json = response.json() self.assertDictEqual( {"success": False, "message": f"Review {review_id} not found"}, response_json, ) def test_delete_review_viewed(self): now = datetime.now().timestamp() with TestClient(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) # Verify it’s reviewed before response = client.get(f"/review/{review_id}") response = client.delete(f"/review/{review_id}/viewed") assert response.status_code == 200 response_json = response.json() self.assertDictEqual( {"success": True, "message": f"Set Review {review_id} as not viewed"}, response_json, ) # Verify it’s unreviewed after with self.assertRaises(DoesNotExist): UserReviewStatus.get( UserReviewStatus.user_id == self.user_id, UserReviewStatus.review_segment == review_id, )