mirror of
https://github.com/blakeblackshear/frigate.git
synced 2024-11-21 19:07:46 +01:00
e8763b3697
* Ignore entire __pycache__ folder instead of individual *.pyc files * Ignore .mypy_cache in git * Rework config YAML parsing to use only ruamel.yaml PyYAML silently overrides keys when encountering duplicates, but ruamel raises and exception by default. Since we're already using it elsewhere, dropping PyYAML is an easy choice to make. * Added EnvString in config to slim down runtime_config() * Added gitlens to devcontainer * Automatically call FrigateConfig.runtime_config() runtime_config needed to be called manually before. Now, it's been removed, but the same code is run by a pydantic validator. * Fix handling of missing -segment_time * Removed type annotation on FrigateConfig's parse I'd like to keep them, but then mypy complains about some fundamental errors with how the pydantic model is structured. I'd like to fix it, but I'd rather work towards moving some of this config to the database.
439 lines
12 KiB
Python
439 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,
|
|
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,
|
|
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,
|
|
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,
|
|
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,
|
|
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,
|
|
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,
|
|
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,
|
|
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),
|
|
self.db,
|
|
None,
|
|
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),
|
|
self.db,
|
|
None,
|
|
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),
|
|
self.db,
|
|
None,
|
|
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()
|