mirror of
https://github.com/blakeblackshear/frigate.git
synced 2024-10-15 20:07:23 +02:00
ab50d0b006
* Add isort and ruff linter Both linters are pretty common among modern python code bases. The isort tool provides stable sorting and grouping, as well as pruning of unused imports. Ruff is a modern linter, that is very fast due to being written in rust. It can detect many common issues in a python codebase. Removes the pylint dev requirement, since ruff replaces it. * treewide: fix issues detected by ruff * treewide: fix bare except clauses * .devcontainer: Set up isort * treewide: optimize imports * treewide: apply black * treewide: make regex patterns raw strings This is necessary for escape sequences to be properly recognized.
386 lines
11 KiB
Python
386 lines
11 KiB
Python
import datetime
|
|
import json
|
|
import logging
|
|
import os
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
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.config import FrigateConfig
|
|
from frigate.http import create_app
|
|
from frigate.models import Event, Recordings
|
|
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,
|
|
None,
|
|
None,
|
|
PlusApi(),
|
|
)
|
|
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,
|
|
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,
|
|
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,
|
|
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,
|
|
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"] 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_set_delete_sub_label(self):
|
|
app = create_app(
|
|
FrigateConfig(**self.minimal_config),
|
|
self.db,
|
|
None,
|
|
None,
|
|
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,
|
|
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,
|
|
None,
|
|
None,
|
|
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,
|
|
None,
|
|
None,
|
|
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,
|
|
None,
|
|
None,
|
|
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()
|