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

fix fastapi

Browse Source
This commit is contained in:
Josh Hawkins 2024-09-24 08:38:04 -05:00
parent 3eb2208a74
commit 1a5f707998
2 changed files with 67 additions and 63 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
frigate/api/fastapi_app.py
View File

@ -13,6 +13,9 @@ from starlette_context.plugins import Plugin
from frigate.api import app as main_app from frigate.api import app as main_app
from frigate.api import auth, event, export, media, notification, preview, review from frigate.api import auth, event, export, media, notification, preview, review
from frigate.api.auth import get_jwt_secret, limiter from frigate.api.auth import get_jwt_secret, limiter
from frigate.comms.event_metadata_updater import (
EventMetadataPublisher,
)
from frigate.config import FrigateConfig from frigate.config import FrigateConfig
from frigate.embeddings import EmbeddingsContext from frigate.embeddings import EmbeddingsContext
from frigate.events.external import ExternalEventProcessor from frigate.events.external import ExternalEventProcessor
@ -47,6 +50,7 @@ def create_fastapi_app(
onvif: OnvifController, onvif: OnvifController,
external_processor: ExternalEventProcessor, external_processor: ExternalEventProcessor,
stats_emitter: StatsEmitter, stats_emitter: StatsEmitter,
event_metadata_updater: EventMetadataPublisher,
): ):
logger.info("Starting FastAPI app") logger.info("Starting FastAPI app")
app = FastAPI( app = FastAPI(
@ -102,6 +106,7 @@ def create_fastapi_app(
app.camera_error_image = None app.camera_error_image = None
app.onvif = onvif app.onvif = onvif
app.stats_emitter = stats_emitter app.stats_emitter = stats_emitter
app.event_metadata_updater = event_metadata_updater
app.external_processor = external_processor app.external_processor = external_processor
app.jwt_token = get_jwt_secret() if frigate_config.auth.enabled else None app.jwt_token = get_jwt_secret() if frigate_config.auth.enabled else None

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.fastapi_app import create_fastapi_app from frigate.api.app import create_app
from frigate.config import FrigateConfig from frigate.config import FrigateConfig
from frigate.models import Event, Recordings, Timeline from frigate.models import Event, Recordings
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, Timeline] models = [Event, Recordings]
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_fastapi_app( app = create_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 TestClient(app) as client: with app.test_client() 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",
params={"limit": 1}, query_string={"limit": 1},
).json() ).json
assert events assert events
assert len(events) == 1 assert len(events) == 1
events = client.get( events = client.get(
"/events", "/events",
params={"has_clip": 0}, query_string={"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_fastapi_app( app = create_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 TestClient(app) as client: with app.test_client() 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_fastapi_app( app = create_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 TestClient(app) as client: with app.test_client() as client:
_insert_mock_event(id) _insert_mock_event(id)
event_response = client.get(f"/events/{bad_id}") event = client.get(f"/events/{bad_id}").json
assert event_response.status_code == 404
assert event_response.json() == "Event not found" assert not event
def test_delete_event(self): def test_delete_event(self):
app = create_fastapi_app( app = create_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 TestClient(app) as client: with app.test_client() 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 event == "Event not found" assert not event
def test_event_retention(self): def test_event_retention(self):
app = create_fastapi_app( app = create_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 TestClient(app) as client: with app.test_client() 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_fastapi_app( app = create_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 TestClient(app) as client: with app.test_client() 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",
params={"time_range": "07:00,24:00"}, query_string={"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",
params={"time_range": "00:00,18:00"}, query_string={"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_fastapi_app( app = create_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 TestClient(app) as client: with app.test_client() as client:
_insert_mock_event(id) _insert_mock_event(id)
new_sub_label_response = client.post( client.post(
f"/events/{id}/sub_label", f"/events/{id}/sub_label",
json={"subLabel": sub_label}, data=json.dumps({"subLabel": sub_label}),
content_type="application/json",
) )
assert new_sub_label_response.status_code == 200 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["sub_label"] == sub_label assert event["sub_label"] == sub_label
empty_sub_label_response = client.post( client.post(
f"/events/{id}/sub_label", f"/events/{id}/sub_label",
json={"subLabel": ""}, data=json.dumps({"subLabel": ""}),
content_type="application/json",
) )
assert empty_sub_label_response.status_code == 200 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["sub_label"] == "" assert event["sub_label"] == ""
def test_sub_label_list(self): def test_sub_label_list(self):
app = create_fastapi_app( app = create_app(
FrigateConfig(**self.minimal_config), FrigateConfig(**self.minimal_config),
self.db, self.db,
None, None,
@ -331,18 +331,19 @@ class TestHttp(unittest.TestCase):
id = "123456.random" id = "123456.random"
sub_label = "sub" sub_label = "sub"
with TestClient(app) as client: with app.test_client() as client:
_insert_mock_event(id) _insert_mock_event(id)
client.post( client.post(
f"/events/{id}/sub_label", f"/events/{id}/sub_label",
json={"subLabel": sub_label}, data=json.dumps({"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_fastapi_app( app = create_app(
FrigateConfig(**self.minimal_config), FrigateConfig(**self.minimal_config),
self.db, self.db,
None, None,
@ -354,13 +355,13 @@ class TestHttp(unittest.TestCase):
None, None,
) )
with TestClient(app) as client: with app.test_client() 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_fastapi_app( app = create_app(
FrigateConfig(**self.minimal_config), FrigateConfig(**self.minimal_config),
self.db, self.db,
None, None,
@ -373,18 +374,16 @@ class TestHttp(unittest.TestCase):
) )
id = "123456.random" id = "123456.random"
with TestClient(app) as client: with app.test_client() as client:
_insert_mock_recording(id) _insert_mock_recording(id)
response = client.get("/front_door/recordings") recording = client.get("/front_door/recordings").json
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_fastapi_app( app = create_app(
FrigateConfig(**self.minimal_config), FrigateConfig(**self.minimal_config),
self.db, self.db,
None, None,
@ -396,8 +395,8 @@ class TestHttp(unittest.TestCase):
None, None,
) )
with TestClient(app) as client: with app.test_client() 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
@ -430,8 +429,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() - 60, start_time=datetime.datetime.now().timestamp() - 50,
end_time=datetime.datetime.now().timestamp() - 50, end_time=datetime.datetime.now().timestamp() - 60,
duration=10, duration=10,
motion=True, motion=True,
objects=True, objects=True,