mirror of
https://github.com/blakeblackshear/frigate.git
synced 2024-10-15 20:07:23 +02:00
3c4b1fb6f2
* Update flask * Update flask and break apart different sections into different files * formatting * Fix test and add safety check
428 lines
12 KiB
Python
428 lines
12 KiB
Python
import datetime
|
|
import json
|
|
import logging
|
|
import os
|
|
import unittest
|
|
from unittest.mock import Mock
|
|
|
|
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.app import create_app
|
|
from frigate.config import FrigateConfig
|
|
from frigate.models import Event, Recordings
|
|
from frigate.plus import PlusApi
|
|
from frigate.stats.emitter import StatsEmitter
|
|
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]
|
|
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 = {
|
|
"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,
|
|
},
|
|
"/media/frigate/clips": {
|
|
"free": 42429.9,
|
|
"mount_type": "ext4",
|
|
"total": 244529.7,
|
|
"used": 189607.0,
|
|
},
|
|
"/media/frigate/recordings": {
|
|
"free": 0.2,
|
|
"mount_type": "ext4",
|
|
"total": 8.0,
|
|
"used": 7.8,
|
|
},
|
|
"/tmp/cache": {
|
|
"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 test_get_event_list(self):
|
|
app = create_app(
|
|
FrigateConfig(**self.minimal_config),
|
|
self.db,
|
|
None,
|
|
None,
|
|
None,
|
|
None,
|
|
PlusApi(),
|
|
None,
|
|
)
|
|
id = "123456.random"
|
|
id2 = "7890.random"
|
|
|
|
with app.test_client() 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",
|
|
query_string={"limit": 1},
|
|
).json
|
|
assert events
|
|
assert len(events) == 1
|
|
events = client.get(
|
|
"/events",
|
|
query_string={"has_clip": 0},
|
|
).json
|
|
assert not events
|
|
|
|
def test_get_good_event(self):
|
|
app = create_app(
|
|
FrigateConfig(**self.minimal_config),
|
|
self.db,
|
|
None,
|
|
None,
|
|
None,
|
|
None,
|
|
PlusApi(),
|
|
None,
|
|
)
|
|
id = "123456.random"
|
|
|
|
with app.test_client() as client:
|
|
_insert_mock_event(id)
|
|
event = client.get(f"/events/{id}").json
|
|
|
|
assert event
|
|
assert event["id"] == id
|
|
assert event == model_to_dict(Event.get(Event.id == id))
|
|
|
|
def test_get_bad_event(self):
|
|
app = create_app(
|
|
FrigateConfig(**self.minimal_config),
|
|
self.db,
|
|
None,
|
|
None,
|
|
None,
|
|
None,
|
|
PlusApi(),
|
|
None,
|
|
)
|
|
id = "123456.random"
|
|
bad_id = "654321.other"
|
|
|
|
with app.test_client() as client:
|
|
_insert_mock_event(id)
|
|
event = client.get(f"/events/{bad_id}").json
|
|
|
|
assert not event
|
|
|
|
def test_delete_event(self):
|
|
app = create_app(
|
|
FrigateConfig(**self.minimal_config),
|
|
self.db,
|
|
None,
|
|
None,
|
|
None,
|
|
None,
|
|
PlusApi(),
|
|
None,
|
|
)
|
|
id = "123456.random"
|
|
|
|
with app.test_client() as client:
|
|
_insert_mock_event(id)
|
|
event = client.get(f"/events/{id}").json
|
|
assert event
|
|
assert event["id"] == id
|
|
client.delete(f"/events/{id}")
|
|
event = client.get(f"/events/{id}").json
|
|
assert not event
|
|
|
|
def test_event_retention(self):
|
|
app = create_app(
|
|
FrigateConfig(**self.minimal_config),
|
|
self.db,
|
|
None,
|
|
None,
|
|
None,
|
|
None,
|
|
PlusApi(),
|
|
None,
|
|
)
|
|
id = "123456.random"
|
|
|
|
with app.test_client() as client:
|
|
_insert_mock_event(id)
|
|
client.post(f"/events/{id}/retain")
|
|
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")
|
|
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 = create_app(
|
|
FrigateConfig(**self.minimal_config),
|
|
self.db,
|
|
None,
|
|
None,
|
|
None,
|
|
None,
|
|
PlusApi(),
|
|
None,
|
|
)
|
|
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 app.test_client() 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",
|
|
query_string={"time_range": "07:00,24:00"},
|
|
).json
|
|
assert events
|
|
# assert len(events) == 1
|
|
# evening event is excluded
|
|
events = client.get(
|
|
"/events",
|
|
query_string={"time_range": "00:00,18:00"},
|
|
).json
|
|
assert events
|
|
assert len(events) == 1
|
|
|
|
def test_set_delete_sub_label(self):
|
|
app = create_app(
|
|
FrigateConfig(**self.minimal_config),
|
|
self.db,
|
|
None,
|
|
None,
|
|
None,
|
|
None,
|
|
PlusApi(),
|
|
None,
|
|
)
|
|
id = "123456.random"
|
|
sub_label = "sub"
|
|
|
|
with app.test_client() as client:
|
|
_insert_mock_event(id)
|
|
client.post(
|
|
f"/events/{id}/sub_label",
|
|
data=json.dumps({"subLabel": sub_label}),
|
|
content_type="application/json",
|
|
)
|
|
event = client.get(f"/events/{id}").json
|
|
assert event
|
|
assert event["id"] == id
|
|
assert event["sub_label"] == sub_label
|
|
client.post(
|
|
f"/events/{id}/sub_label",
|
|
data=json.dumps({"subLabel": ""}),
|
|
content_type="application/json",
|
|
)
|
|
event = client.get(f"/events/{id}").json
|
|
assert event
|
|
assert event["id"] == id
|
|
assert event["sub_label"] == ""
|
|
|
|
def test_sub_label_list(self):
|
|
app = create_app(
|
|
FrigateConfig(**self.minimal_config),
|
|
self.db,
|
|
None,
|
|
None,
|
|
None,
|
|
None,
|
|
PlusApi(),
|
|
None,
|
|
)
|
|
id = "123456.random"
|
|
sub_label = "sub"
|
|
|
|
with app.test_client() as client:
|
|
_insert_mock_event(id)
|
|
client.post(
|
|
f"/events/{id}/sub_label",
|
|
data=json.dumps({"subLabel": sub_label}),
|
|
content_type="application/json",
|
|
)
|
|
sub_labels = client.get("/sub_labels").json
|
|
assert sub_labels
|
|
assert sub_labels == [sub_label]
|
|
|
|
def test_config(self):
|
|
app = create_app(
|
|
FrigateConfig(**self.minimal_config).runtime_config(),
|
|
self.db,
|
|
None,
|
|
None,
|
|
None,
|
|
None,
|
|
PlusApi(),
|
|
None,
|
|
)
|
|
|
|
with app.test_client() as client:
|
|
config = client.get("/config").json
|
|
assert config
|
|
assert config["cameras"]["front_door"]
|
|
|
|
def test_recordings(self):
|
|
app = create_app(
|
|
FrigateConfig(**self.minimal_config).runtime_config(),
|
|
self.db,
|
|
None,
|
|
None,
|
|
None,
|
|
None,
|
|
PlusApi(),
|
|
None,
|
|
)
|
|
id = "123456.random"
|
|
|
|
with app.test_client() as client:
|
|
_insert_mock_recording(id)
|
|
recording = client.get("/front_door/recordings").json
|
|
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_app(
|
|
FrigateConfig(**self.minimal_config).runtime_config(),
|
|
self.db,
|
|
None,
|
|
None,
|
|
None,
|
|
None,
|
|
PlusApi(),
|
|
stats,
|
|
)
|
|
|
|
with app.test_client() as client:
|
|
full_stats = client.get("/stats").json
|
|
assert full_stats == self.test_stats
|
|
|
|
|
|
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() - 50,
|
|
end_time=datetime.datetime.now().timestamp() - 60,
|
|
duration=10,
|
|
motion=True,
|
|
objects=True,
|
|
).execute()
|