administrator/frigate
1
0
Fork 0
mirror of https://github.com/blakeblackshear/frigate.git synced 2026-02-15 07:35:27 +03:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

fix test

Browse Source
This commit is contained in:
Josh Hawkins 2024-09-24 08:39:29 -05:00
parent 1a5f707998
commit ad9d2666ce
1 changed files with 63 additions and 62 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

125
frigate/test/test_http.py
View File

@ -1,18 +1,18 @@
import datetime import datetime
import json
import logging import logging
import os import os
import unittest import unittest
from unittest.mock import Mock from unittest.mock import Mock
from fastapi.testclient import TestClient
from peewee_migrate import Router from peewee_migrate import Router
from playhouse.shortcuts import model_to_dict from playhouse.shortcuts import model_to_dict
from playhouse.sqlite_ext import SqliteExtDatabase from playhouse.sqlite_ext import SqliteExtDatabase
from playhouse.sqliteq import SqliteQueueDatabase from playhouse.sqliteq import SqliteQueueDatabase
from frigate.api.app import create_app from frigate.api.fastapi_app import create_fastapi_app
from frigate.config import FrigateConfig from frigate.config import FrigateConfig
from frigate.models import Event, Recordings from frigate.models import Event, Recordings, Timeline
from frigate.stats.emitter import StatsEmitter from frigate.stats.emitter import StatsEmitter
from frigate.test.const import TEST_DB, TEST_DB_CLEANUPS from frigate.test.const import TEST_DB, TEST_DB_CLEANUPS
@ -26,7 +26,7 @@ class TestHttp(unittest.TestCase):
router.run() router.run()
migrate_db.close() migrate_db.close()
self.db = SqliteQueueDatabase(TEST_DB) self.db = SqliteQueueDatabase(TEST_DB)
models = [Event, Recordings] models = [Event, Recordings, Timeline]
self.db.bind(models) self.db.bind(models)
self.minimal_config = { self.minimal_config = {
@ -112,7 +112,7 @@ class TestHttp(unittest.TestCase):
pass pass
def test_get_event_list(self): def test_get_event_list(self):
app = create_app( app = create_fastapi_app(
FrigateConfig(**self.minimal_config), FrigateConfig(**self.minimal_config),
self.db, self.db,
None, None,
@ -126,30 +126,30 @@ class TestHttp(unittest.TestCase):
id = "123456.random" id = "123456.random"
id2 = "7890.random" id2 = "7890.random"
with app.test_client() as client: with TestClient(app) as client:
_insert_mock_event(id) _insert_mock_event(id)
events = client.get("/events").json events = client.get("/events").json()
assert events assert events
assert len(events) == 1 assert len(events) == 1
assert events[0]["id"] == id assert events[0]["id"] == id
_insert_mock_event(id2) _insert_mock_event(id2)
events = client.get("/events").json events = client.get("/events").json()
assert events assert events
assert len(events) == 2 assert len(events) == 2
events = client.get( events = client.get(
"/events", "/events",
query_string={"limit": 1}, params={"limit": 1},
).json ).json()
assert events assert events
assert len(events) == 1 assert len(events) == 1
events = client.get( events = client.get(
"/events", "/events",
query_string={"has_clip": 0}, params={"has_clip": 0},
).json ).json()
assert not events assert not events
def test_get_good_event(self): def test_get_good_event(self):
app = create_app( app = create_fastapi_app(
FrigateConfig(**self.minimal_config), FrigateConfig(**self.minimal_config),
self.db, self.db,
None, None,
@ -162,16 +162,16 @@ class TestHttp(unittest.TestCase):
) )
id = "123456.random" id = "123456.random"
with app.test_client() as client: with TestClient(app) as client:
_insert_mock_event(id) _insert_mock_event(id)
event = client.get(f"/events/{id}").json event = client.get(f"/events/{id}").json()
assert event assert event
assert event["id"] == id assert event["id"] == id
assert event == model_to_dict(Event.get(Event.id == id)) assert event == model_to_dict(Event.get(Event.id == id))
def test_get_bad_event(self): def test_get_bad_event(self):
app = create_app( app = create_fastapi_app(
FrigateConfig(**self.minimal_config), FrigateConfig(**self.minimal_config),
self.db, self.db,
None, None,
@ -185,14 +185,14 @@ class TestHttp(unittest.TestCase):
id = "123456.random" id = "123456.random"
bad_id = "654321.other" bad_id = "654321.other"
with app.test_client() as client: with TestClient(app) as client:
_insert_mock_event(id) _insert_mock_event(id)
event = client.get(f"/events/{bad_id}").json event_response = client.get(f"/events/{bad_id}")
assert event_response.status_code == 404
assert not event assert event_response.json() == "Event not found"
def test_delete_event(self): def test_delete_event(self):
app = create_app( app = create_fastapi_app(
FrigateConfig(**self.minimal_config), FrigateConfig(**self.minimal_config),
self.db, self.db,
None, None,
@ -205,17 +205,17 @@ class TestHttp(unittest.TestCase):
) )
id = "123456.random" id = "123456.random"
with app.test_client() as client: with TestClient(app) as client:
_insert_mock_event(id) _insert_mock_event(id)
event = client.get(f"/events/{id}").json event = client.get(f"/events/{id}").json()
assert event assert event
assert event["id"] == id assert event["id"] == id
client.delete(f"/events/{id}") client.delete(f"/events/{id}")
event = client.get(f"/events/{id}").json event = client.get(f"/events/{id}").json()
assert not event assert event == "Event not found"
def test_event_retention(self): def test_event_retention(self):
app = create_app( app = create_fastapi_app(
FrigateConfig(**self.minimal_config), FrigateConfig(**self.minimal_config),
self.db, self.db,
None, None,
@ -228,21 +228,21 @@ class TestHttp(unittest.TestCase):
) )
id = "123456.random" id = "123456.random"
with app.test_client() as client: with TestClient(app) as client:
_insert_mock_event(id) _insert_mock_event(id)
client.post(f"/events/{id}/retain") client.post(f"/events/{id}/retain")
event = client.get(f"/events/{id}").json event = client.get(f"/events/{id}").json()
assert event assert event
assert event["id"] == id assert event["id"] == id
assert event["retain_indefinitely"] is True assert event["retain_indefinitely"] is True
client.delete(f"/events/{id}/retain") client.delete(f"/events/{id}/retain")
event = client.get(f"/events/{id}").json event = client.get(f"/events/{id}").json()
assert event assert event
assert event["id"] == id assert event["id"] == id
assert event["retain_indefinitely"] is False assert event["retain_indefinitely"] is False
def test_event_time_filtering(self): def test_event_time_filtering(self):
app = create_app( app = create_fastapi_app(
FrigateConfig(**self.minimal_config), FrigateConfig(**self.minimal_config),
self.db, self.db,
None, None,
@ -258,30 +258,30 @@ class TestHttp(unittest.TestCase):
morning = 1656590400 # 06/30/2022 6 am (GMT) morning = 1656590400 # 06/30/2022 6 am (GMT)
evening = 1656633600 # 06/30/2022 6 pm (GMT) evening = 1656633600 # 06/30/2022 6 pm (GMT)
with app.test_client() as client: with TestClient(app) as client:
_insert_mock_event(morning_id, morning) _insert_mock_event(morning_id, morning)
_insert_mock_event(evening_id, evening) _insert_mock_event(evening_id, evening)
# both events come back # both events come back
events = client.get("/events").json events = client.get("/events").json()
assert events assert events
assert len(events) == 2 assert len(events) == 2
# morning event is excluded # morning event is excluded
events = client.get( events = client.get(
"/events", "/events",
query_string={"time_range": "07:00,24:00"}, params={"time_range": "07:00,24:00"},
).json ).json()
assert events assert events
# assert len(events) == 1 # assert len(events) == 1
# evening event is excluded # evening event is excluded
events = client.get( events = client.get(
"/events", "/events",
query_string={"time_range": "00:00,18:00"}, params={"time_range": "00:00,18:00"},
).json ).json()
assert events assert events
assert len(events) == 1 assert len(events) == 1
def test_set_delete_sub_label(self): def test_set_delete_sub_label(self):
app = create_app( app = create_fastapi_app(
FrigateConfig(**self.minimal_config), FrigateConfig(**self.minimal_config),
self.db, self.db,
None, None,
@ -295,29 +295,29 @@ class TestHttp(unittest.TestCase):
id = "123456.random" id = "123456.random"
sub_label = "sub" sub_label = "sub"
with app.test_client() as client: with TestClient(app) as client:
_insert_mock_event(id) _insert_mock_event(id)
client.post( new_sub_label_response = client.post(
f"/events/{id}/sub_label", f"/events/{id}/sub_label",
data=json.dumps({"subLabel": sub_label}), json={"subLabel": sub_label},
content_type="application/json",
) )
event = client.get(f"/events/{id}").json assert new_sub_label_response.status_code == 200
event = client.get(f"/events/{id}").json()
assert event assert event
assert event["id"] == id assert event["id"] == id
assert event["sub_label"] == sub_label assert event["sub_label"] == sub_label
client.post( empty_sub_label_response = client.post(
f"/events/{id}/sub_label", f"/events/{id}/sub_label",
data=json.dumps({"subLabel": ""}), json={"subLabel": ""},
content_type="application/json",
) )
event = client.get(f"/events/{id}").json assert empty_sub_label_response.status_code == 200
event = client.get(f"/events/{id}").json()
assert event assert event
assert event["id"] == id assert event["id"] == id
assert event["sub_label"] == "" assert event["sub_label"] == ""
def test_sub_label_list(self): def test_sub_label_list(self):
app = create_app( app = create_fastapi_app(
FrigateConfig(**self.minimal_config), FrigateConfig(**self.minimal_config),
self.db, self.db,
None, None,
@ -331,19 +331,18 @@ class TestHttp(unittest.TestCase):
id = "123456.random" id = "123456.random"
sub_label = "sub" sub_label = "sub"
with app.test_client() as client: with TestClient(app) as client:
_insert_mock_event(id) _insert_mock_event(id)
client.post( client.post(
f"/events/{id}/sub_label", f"/events/{id}/sub_label",
data=json.dumps({"subLabel": sub_label}), json={"subLabel": sub_label},
content_type="application/json",
) )
sub_labels = client.get("/sub_labels").json sub_labels = client.get("/sub_labels").json()
assert sub_labels assert sub_labels
assert sub_labels == [sub_label] assert sub_labels == [sub_label]
def test_config(self): def test_config(self):
app = create_app( app = create_fastapi_app(
FrigateConfig(**self.minimal_config), FrigateConfig(**self.minimal_config),
self.db, self.db,
None, None,
@ -355,13 +354,13 @@ class TestHttp(unittest.TestCase):
None, None,
) )
with app.test_client() as client: with TestClient(app) as client:
config = client.get("/config").json config = client.get("/config").json()
assert config assert config
assert config["cameras"]["front_door"] assert config["cameras"]["front_door"]
def test_recordings(self): def test_recordings(self):
app = create_app( app = create_fastapi_app(
FrigateConfig(**self.minimal_config), FrigateConfig(**self.minimal_config),
self.db, self.db,
None, None,
@ -374,16 +373,18 @@ class TestHttp(unittest.TestCase):
) )
id = "123456.random" id = "123456.random"
with app.test_client() as client: with TestClient(app) as client:
_insert_mock_recording(id) _insert_mock_recording(id)
recording = client.get("/front_door/recordings").json response = client.get("/front_door/recordings")
assert response.status_code == 200
recording = response.json()
assert recording assert recording
assert recording[0]["id"] == id assert recording[0]["id"] == id
def test_stats(self): def test_stats(self):
stats = Mock(spec=StatsEmitter) stats = Mock(spec=StatsEmitter)
stats.get_latest_stats.return_value = self.test_stats stats.get_latest_stats.return_value = self.test_stats
app = create_app( app = create_fastapi_app(
FrigateConfig(**self.minimal_config), FrigateConfig(**self.minimal_config),
self.db, self.db,
None, None,
@ -395,8 +396,8 @@ class TestHttp(unittest.TestCase):
None, None,
) )
with app.test_client() as client: with TestClient(app) as client:
full_stats = client.get("/stats").json full_stats = client.get("/stats").json()
assert full_stats == self.test_stats assert full_stats == self.test_stats
@ -429,8 +430,8 @@ def _insert_mock_recording(id: str) -> Event:
id=id, id=id,
camera="front_door", camera="front_door",
path=f"/recordings/{id}", path=f"/recordings/{id}",
start_time=datetime.datetime.now().timestamp() - 50, start_time=datetime.datetime.now().timestamp() - 60,
end_time=datetime.datetime.now().timestamp() - 60, end_time=datetime.datetime.now().timestamp() - 50,
duration=10, duration=10,
motion=True, motion=True,
objects=True, objects=True,