mirror of
https://github.com/blakeblackshear/frigate.git
synced 2025-12-06 13:34:13 +03:00
cd606ad240
Some checks are pending
CI / AMD64 Extra Build (push) Blocked by required conditions
CI / ARM Extra Build (push) Blocked by required conditions
CI / AMD64 Build (push) Waiting to run
CI / ARM Build (push) Waiting to run
CI / Jetson Jetpack 6 (push) Waiting to run
CI / Synaptics Build (push) Blocked by required conditions
CI / Assemble and push default build (push) Blocked by required conditions
* 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
711 lines
30 KiB
Python
711 lines
30 KiB
Python
from datetime import datetime, timedelta
|
||
|
||
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 AuthTestClient, 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
|
||
# 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
|
||
|
||
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()
|
||
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 AuthTestClient(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 AuthTestClient(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 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)
|
||
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 AuthTestClient(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 AuthTestClient(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 AuthTestClient(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 AuthTestClient(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 AuthTestClient(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 AuthTestClient(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 AuthTestClient(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 AuthTestClient(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 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):
|
||
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 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"
|
||
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 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 AuthTestClient(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 AuthTestClient(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"] == "Marked multiple items as reviewed"
|
||
# 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 AuthTestClient(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"] == "Marked multiple items as reviewed"
|
||
# 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 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 AuthTestClient(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 AuthTestClient(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 AuthTestClient(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 AuthTestClient(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 AuthTestClient(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 AuthTestClient(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 AuthTestClient(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 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}")
|
||
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 AuthTestClient(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 AuthTestClient(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 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}")
|
||
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 AuthTestClient(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 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)
|
||
# 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,
|
||
)
|