blakeblackshear.frigate/frigate/test/test_http.py

332 lines
10 KiB
Python
Raw Normal View History

import datetime
import json
import logging
import os
import unittest
from unittest.mock import patch
from peewee_migrate import Router
from playhouse.sqlite_ext import SqliteExtDatabase
from playhouse.sqliteq import SqliteQueueDatabase
from playhouse.shortcuts import model_to_dict
from frigate.config import FrigateConfig
from frigate.http import create_app
from frigate.models import Event, Recordings
2022-07-22 14:56:28 +02:00
from frigate.plus import PlusApi
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, PlusApi()
)
id = "123456.random"
id2 = "7890.random"
with app.test_client() as client:
_insert_mock_event(id)
events = client.get(f"/events").json
assert events
assert len(events) == 1
assert events[0]["id"] == id
_insert_mock_event(id2)
events = client.get(f"/events").json
assert events
assert len(events) == 2
events = client.get(
f"/events",
query_string={"limit": 1},
).json
assert events
assert len(events) == 1
events = client.get(
f"/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, PlusApi()
)
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, PlusApi()
)
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, PlusApi()
)
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, PlusApi()
)
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"] == True
client.delete(f"/events/{id}/retain")
event = client.get(f"/events/{id}").json
assert event
assert event["id"] == id
assert event["retain_indefinitely"] == False
def test_set_delete_sub_label(self):
app = create_app(
FrigateConfig(**self.minimal_config), self.db, None, None, None, PlusApi()
)
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, PlusApi()
)
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,
2022-07-22 14:56:28 +02:00
PlusApi(),
)
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,
2022-07-22 14:56:28 +02:00
PlusApi(),
)
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
@patch("frigate.http.stats_snapshot")
def test_stats(self, mock_stats):
app = create_app(
FrigateConfig(**self.minimal_config).runtime_config,
self.db,
None,
None,
None,
2022-07-22 14:56:28 +02:00
PlusApi(),
)
mock_stats.return_value = self.test_stats
with app.test_client() as client:
stats = client.get("/stats").json
assert stats == self.test_stats
def _insert_mock_event(id: str) -> Event:
"""Inserts a basic event model with a given id."""
return Event.insert(
id=id,
label="Mock",
camera="front_door",
start_time=datetime.datetime.now().timestamp(),
end_time=datetime.datetime.now().timestamp() + 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()