Mirrors/blakeblackshear.frigate
1
0
Fork 0
mirror of https://github.com/blakeblackshear/frigate.git synced 2025-10-27 10:52:11 +01:00
Code Issues Projects Releases Wiki Activity

Finish unit tests for review controller and started for event controller (#15955)

Browse Source 
* Started unit tests for the review controller

* Revert "Started unit tests for the review controller"

This reverts commit 7746eb146f.

* Started unit tests for GET /review/activity/motion Endpoint

* Started unit tests for GET /review/event/{event_id} Endpoint

* Continued unit tests for GET /review/event/{event_id} Endpoint

* Continued unit tests for GET /review/{event_id} Endpoint

* Continued unit tests for GET /review/{review_id} Endpoint

* Added unit tests for GET /review/{review_id}/viewed Endpoint

* Added unit tests for GET /stats Endpoint

* Added unit tests for GET /events Endpoint

* Updated unit tests for GET /events Endpoint

* Deleted unit tests for /events from test_http (updated tests are now in test_http_event.py)

* Removed duplicated test for GET /review/activity/motion Endpoint
This commit is contained in:
Rui Alves 2025-02-04 13:28:14 +00:00 committed by GitHub
parent 0645dc70a5
commit df840b7cd5
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
6 changed files with 354 additions and 66 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
frigate/api/review.py
View File

@ -490,8 +490,6 @@ def set_not_reviewed(review_id: str):
review.save()
return JSONResponse(
content=(
{"success": True, "message": "Set Review " + review_id + " as not viewed"}
),
content=({"success": True, "message": f"Set Review {review_id} as not viewed"}),
status_code=200,
)

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

@ -6,6 +6,7 @@ import unittest
from peewee_migrate import Router
from playhouse.sqlite_ext import SqliteExtDatabase
from playhouse.sqliteq import SqliteQueueDatabase
from pydantic import Json
from frigate.api.fastapi_app import create_fastapi_app
from frigate.config import FrigateConfig
@ -123,7 +124,12 @@ class BaseTestHttp(unittest.TestCase):
def insert_mock_event(
self,
id: str,
start_time: datetime.datetime = datetime.datetime.now().timestamp(),
start_time: float = datetime.datetime.now().timestamp(),
end_time: float = datetime.datetime.now().timestamp() + 20,
has_clip: bool = True,
top_score: int = 100,
score: int = 0,
data: Json = {},
) -> Event:
"""Inserts a basic event model with a given id."""
return Event.insert(
@ -131,16 +137,18 @@ class BaseTestHttp(unittest.TestCase):
label="Mock",
camera="front_door",
start_time=start_time,
end_time=start_time + 20,
top_score=100,
end_time=end_time,
top_score=top_score,
score=score,
false_positive=False,
zones=list(),
thumbnail="",
region=[],
box=[],
area=0,
has_clip=True,
has_clip=has_clip,
has_snapshot=True,
data=data,
).execute()
def insert_mock_review_segment(
@ -150,6 +158,7 @@ class BaseTestHttp(unittest.TestCase):
end_time: float = datetime.datetime.now().timestamp() + 20,
severity: SeverityEnum = SeverityEnum.alert,
has_been_reviewed: bool = False,
data: Json = {},
) -> Event:
"""Inserts a review segment model with a given id."""
return ReviewSegment.insert(
@ -160,7 +169,7 @@ class BaseTestHttp(unittest.TestCase):
has_been_reviewed=has_been_reviewed,
severity=severity,
thumb_path=False,
data={},
data=data,
).execute()
def insert_mock_recording(
@ -168,6 +177,7 @@ class BaseTestHttp(unittest.TestCase):
id: str,
start_time: float = datetime.datetime.now().timestamp(),
end_time: float = datetime.datetime.now().timestamp() + 20,
motion: int = 0,
) -> Event:
"""Inserts a recording model with a given id."""
return Recordings.insert(
@ -177,4 +187,5 @@ class BaseTestHttp(unittest.TestCase):
start_time=start_time,
end_time=end_time,
duration=end_time - start_time,
motion=motion,
).execute()

26
frigate/test/http_api/test_http_app.py Normal file
View File

@ -0,0 +1,26 @@
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
class TestHttpApp(BaseTestHttp):
def setUp(self):
super().setUp([Event, Recordings, ReviewSegment])
self.app = super().create_app()
####################################################################################################################
################################### GET /stats Endpoint #########################################################
####################################################################################################################
def test_stats_endpoint(self):
stats = Mock(spec=StatsEmitter)
stats.get_latest_stats.return_value = self.test_stats
app = super().create_app(stats)
with TestClient(app) as client:
response = client.get("/stats")
response_json = response.json()
assert response_json == self.test_stats

137
frigate/test/http_api/test_http_event.py Normal file
View File

@ -0,0 +1,137 @@
from datetime import datetime
from fastapi.testclient import TestClient
from frigate.models import Event, Recordings, ReviewSegment
from frigate.test.http_api.base_http_test import BaseTestHttp
class TestHttpApp(BaseTestHttp):
def setUp(self):
super().setUp([Event, Recordings, ReviewSegment])
self.app = super().create_app()
####################################################################################################################
################################### GET /events Endpoint #########################################################
####################################################################################################################
def test_get_event_list_no_events(self):
with TestClient(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:
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:
super().insert_mock_event(id)
events = client.get("/events", params={"event_id": id}).json()
assert len(events) == 1
assert events[0]["id"] == id
def test_get_event_list_match_length(self):
now = int(datetime.now().timestamp())
id = "123456.random"
with TestClient(self.app) as client:
super().insert_mock_event(id, now, now + 1)
events = client.get(
"/events", params={"max_length": 1, "min_length": 1}
).json()
assert len(events) == 1
assert events[0]["id"] == id
def test_get_event_list_no_match_max_length(self):
now = int(datetime.now().timestamp())
with TestClient(self.app) as client:
id = "123456.random"
super().insert_mock_event(id, now, now + 2)
events = client.get("/events", params={"max_length": 1}).json()
assert len(events) == 0
def test_get_event_list_no_match_min_length(self):
now = int(datetime.now().timestamp())
with TestClient(self.app) as client:
id = "123456.random"
super().insert_mock_event(id, now, now + 2)
events = client.get("/events", params={"min_length": 3}).json()
assert len(events) == 0
def test_get_event_list_limit(self):
id = "123456.random"
id2 = "54321.random"
with TestClient(self.app) as client:
super().insert_mock_event(id)
events = client.get("/events").json()
assert len(events) == 1
assert events[0]["id"] == id
super().insert_mock_event(id2)
events = client.get("/events").json()
assert len(events) == 2
events = client.get("/events", params={"limit": 1}).json()
assert len(events) == 1
assert events[0]["id"] == id
events = client.get("/events", params={"limit": 3}).json()
assert len(events) == 2
def test_get_event_list_no_match_has_clip(self):
now = int(datetime.now().timestamp())
with TestClient(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:
id = "123456.random"
super().insert_mock_event(id, has_clip=True)
events = client.get("/events", params={"has_clip": 1}).json()
assert len(events) == 1
assert events[0]["id"] == id
def test_get_event_list_sort_score(self):
with TestClient(self.app) as client:
id = "123456.random"
id2 = "54321.random"
super().insert_mock_event(id, top_score=37, score=37, data={"score": 50})
super().insert_mock_event(id2, top_score=47, score=47, data={"score": 20})
events = client.get("/events", params={"sort": "score_asc"}).json()
assert len(events) == 2
assert events[0]["id"] == id2
assert events[1]["id"] == id
events = client.get("/events", params={"sort": "score_des"}).json()
assert len(events) == 2
assert events[0]["id"] == id
assert events[1]["id"] == id2
def test_get_event_list_sort_start_time(self):
now = int(datetime.now().timestamp())
with TestClient(self.app) as client:
id = "123456.random"
id2 = "54321.random"
super().insert_mock_event(id, start_time=now + 3)
super().insert_mock_event(id2, start_time=now)
events = client.get("/events", params={"sort": "date_asc"}).json()
assert len(events) == 2
assert events[0]["id"] == id2
assert events[1]["id"] == id
events = client.get("/events", params={"sort": "date_desc"}).json()
assert len(events) == 2
assert events[0]["id"] == id
assert events[1]["id"] == id2

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

@ -569,3 +569,177 @@ class TestHttpReview(BaseTestHttp):
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,
"has_been_reviewed": False,
"severity": SeverityEnum.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,
"has_been_reviewed": False,
"severity": SeverityEnum.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, has_been_reviewed=True
)
review_before = ReviewSegment.get(ReviewSegment.id == review_id)
assert review_before.has_been_reviewed == True
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,
)
review_after = ReviewSegment.get(ReviewSegment.id == review_id)
assert review_after.has_been_reviewed == False

58
frigate/test/test_http.py
View File

@ -2,7 +2,6 @@ import datetime
import logging
import os
import unittest
from unittest.mock import Mock
from fastapi.testclient import TestClient
from peewee_migrate import Router
@ -13,7 +12,6 @@ from playhouse.sqliteq import SqliteQueueDatabase
from frigate.api.fastapi_app import create_fastapi_app
from frigate.config import FrigateConfig
from frigate.models import Event, Recordings, Timeline
from frigate.stats.emitter import StatsEmitter
from frigate.test.const import TEST_DB, TEST_DB_CLEANUPS
@ -111,43 +109,6 @@ class TestHttp(unittest.TestCase):
except OSError:
pass
def test_get_event_list(self):
app = create_fastapi_app(
FrigateConfig(**self.minimal_config),
self.db,
None,
None,
None,
None,
None,
None,
None,
)
id = "123456.random"
id2 = "7890.random"
with TestClient(app) as client:
_insert_mock_event(id)
events = client.get("/events").json()
assert events
assert len(events) == 1
assert events[0]["id"] == id
_insert_mock_event(id2)
events = client.get("/events").json()
assert events
assert len(events) == 2
events = client.get(
"/events",
params={"limit": 1},
).json()
assert events
assert len(events) == 1
events = client.get(
"/events",
params={"has_clip": 0},
).json()
assert not events
def test_get_good_event(self):
app = create_fastapi_app(
FrigateConfig(**self.minimal_config),
@ -381,25 +342,6 @@ class TestHttp(unittest.TestCase):
assert recording
assert recording[0]["id"] == id
def test_stats(self):
stats = Mock(spec=StatsEmitter)
stats.get_latest_stats.return_value = self.test_stats
app = create_fastapi_app(
FrigateConfig(**self.minimal_config),
self.db,
None,
None,
None,
None,
None,
stats,
None,
)
with TestClient(app) as client:
full_stats = client.get("/stats").json()
assert full_stats == self.test_stats
def _insert_mock_event(
id: str,