mirror of
https://github.com/blakeblackshear/frigate.git
synced 2026-10-23 23:14:46 +03:00
Josh HawkinsandGitHub
f385065981
* materialize event counts in metrics endpoint The metrics endpoint stored a lazy peewee query on the shared Prometheus collector, and collect() ran it later. Overlapping scrapes run in separate threadpool threads, so two requests could execute and step the same query's sqlite cursor at once. The rows are now read with list() before they reach the collector. * fix fps overflow restarting ffmpeg after 3 seconds instead of 3 retry intervals This is a regression from 0.18. The watchdog has restarted ffmpeg after 3 consecutive fps overflow checks since #13651, which was 3 retry intervals (30s by default) when the loop ran every retry_interval. #21297 moved the loop to 1s ticks without converting the count, so the window shrank to 3s. camera_fps is a 10s average, so one burst of buffered frames after a stream hiccup could keep it over the limit long enough to restart a healthy camera, and the next reconnect burst could repeat it. The threshold is now 3 * retry_interval ticks. When the count filled while restarts were still rate limited, the watchdog logged "Exiting ffmpeg...", marked the camera offline and reset the count without restarting. The check now waits for can_restart, so the log, status and restart always happen together. * keep recording cleanup running after a failed pass An exception anywhere in a cleanup pass, such as an OSError from a read-only or unmounted storage path, ended the recording cleanup thread with nothing logged. Retention stopped until restart, and recordings, previews and database rows kept growing. Each pass now logs the exception and the loop continues. * expose frigate_camera_events as a gauge The metric reports the current count of events in the database, which drops when retention prunes old events. As a counter, each prune read as a counter reset, so increase() and rate() inflated event counts. The series name changes from frigate_camera_events_total to frigate_camera_events. The test's porch/inside assertion was missing its `in event.text` and always passed. * keep uvicorn idle connections open longer than nginx's upstream keepalive uvicorn closed idle connections after its 5s default while nginx kept them pooled for 60s, so a request nginx sent on a connection uvicorn was closing got a 502. nginx retries that on a fresh connection for most methods, but not for POST or PATCH. * snapshot websocket clients under the manager lock in the output loop The output loop iterated ws4py's live websocket dict without its lock, so a client connecting or disconnecting mid-iteration raised "dictionary changed size during iteration" and killed the output loop. It now copies the clients under manager.lock once per frame and skips terminated ones, matching the jsmpeg and birdseye broadcasters. * only cache preview frames when recording is enabled The segment-end check tested the record config object instead of record.enabled, and the first frame after startup was written unconditionally. With recording disabled, the first segment still cached frames and encoded a preview. * fix untranslated unknown resolution text in camera info dialog * detect all AMD GPUs from vainfo The vainfo check only matched "AMD Radeon Graphics", so drivers reporting names like "AMD Radeon Vega 8 Graphics", "AMD Radeon RX 6600", "AMD Custom GPU 0405", or older codename strings fell through and their GPU stats were collected as Intel. It now matches Mesa's "for AMD" driver string. * return a 500 instead of raising when go2rtc is unreachable The two go2rtc stream GETs had no exception handling or timeout, so a go2rtc outage raised a connection error with a traceback, and a hung go2rtc held the request open indefinitely. Both now catch request errors, log the cause, and use a timeout. The per-stream probe gets a 35s read timeout because go2rtc probes each source in turn, and one unresponsive source takes about 10s. * retry camera snapshots after a failed load AutoUpdatingCameraImage only scheduled the next frame from onLoad, so a single failed latest.webp request froze the image until the component remounted. It now retries on error, dropping the periodic cache key so the retry requests a fresh URL. * reuse pooled upstream connections for nginx auth subrequests The /auth subrequest proxied to a variable address, which nginx can't pool, and its location-level headers dropped the inherited Connection header so nginx sent Connection: close. Every authenticated request opened and closed its own loopback connection to the API. It now proxies to the frigate_api upstream with an empty Connection header so the keepalive pool is reused.
583 lines
22 KiB
Python
583 lines
22 KiB
Python
from datetime import datetime
|
|
from typing import Any
|
|
from unittest.mock import Mock
|
|
|
|
from playhouse.shortcuts import model_to_dict
|
|
|
|
from frigate.api.auth import get_allowed_cameras_for_filter, get_current_user
|
|
from frigate.comms.event_metadata_updater import EventMetadataPublisher
|
|
from frigate.models import Event, Recordings, ReviewSegment, Timeline
|
|
from frigate.stats.emitter import StatsEmitter
|
|
from frigate.test.http_api.base_http_test import AuthTestClient, BaseTestHttp, Request
|
|
from frigate.test.test_storage import _insert_mock_event
|
|
|
|
|
|
class TestHttpApp(BaseTestHttp):
|
|
def setUp(self):
|
|
super().setUp([Event, Recordings, ReviewSegment, Timeline])
|
|
self.app = super().create_app()
|
|
|
|
# Mock get_current_user for all tests
|
|
async def mock_get_current_user(request: Request):
|
|
username = request.headers.get("remote-user")
|
|
role = request.headers.get("remote-role")
|
|
if not username or not role:
|
|
from fastapi.responses import JSONResponse
|
|
|
|
return JSONResponse(
|
|
content={"message": "No authorization headers."}, status_code=401
|
|
)
|
|
return {"username": username, "role": role}
|
|
|
|
self.app.dependency_overrides[get_current_user] = mock_get_current_user
|
|
|
|
async def mock_get_allowed_cameras_for_filter(request: Request):
|
|
return ["front_door"]
|
|
|
|
self.app.dependency_overrides[get_allowed_cameras_for_filter] = (
|
|
mock_get_allowed_cameras_for_filter
|
|
)
|
|
|
|
def tearDown(self):
|
|
self.app.dependency_overrides.clear()
|
|
super().tearDown()
|
|
|
|
####################################################################################################################
|
|
################################### GET /events Endpoint #########################################################
|
|
####################################################################################################################
|
|
def test_get_event_list_no_events(self):
|
|
with AuthTestClient(self.app) as client:
|
|
events = client.get("/events").json()
|
|
assert len(events) == 0
|
|
|
|
def test_get_event_list_no_match_event_id(self):
|
|
id = "123456.random"
|
|
with AuthTestClient(self.app) as client:
|
|
super().insert_mock_event(id)
|
|
events = client.get("/events", params={"event_id": "abc"}).json()
|
|
assert len(events) == 0
|
|
|
|
def test_get_event_list_match_event_id(self):
|
|
id = "123456.random"
|
|
with AuthTestClient(self.app) as client:
|
|
super().insert_mock_event(id)
|
|
events = client.get("/events", params={"event_id": id}).json()
|
|
assert len(events) == 1
|
|
assert events[0]["id"] == id
|
|
|
|
def test_get_event_list_match_length(self):
|
|
now = int(datetime.now().timestamp())
|
|
|
|
id = "123456.random"
|
|
with AuthTestClient(self.app) as client:
|
|
super().insert_mock_event(id, now, now + 1)
|
|
events = client.get(
|
|
"/events", params={"max_length": 1, "min_length": 1}
|
|
).json()
|
|
assert len(events) == 1
|
|
assert events[0]["id"] == id
|
|
|
|
def test_get_event_list_no_match_max_length(self):
|
|
now = int(datetime.now().timestamp())
|
|
|
|
with AuthTestClient(self.app) as client:
|
|
id = "123456.random"
|
|
super().insert_mock_event(id, now, now + 2)
|
|
events = client.get("/events", params={"max_length": 1}).json()
|
|
assert len(events) == 0
|
|
|
|
def test_get_event_list_no_match_min_length(self):
|
|
now = int(datetime.now().timestamp())
|
|
|
|
with AuthTestClient(self.app) as client:
|
|
id = "123456.random"
|
|
super().insert_mock_event(id, now, now + 2)
|
|
events = client.get("/events", params={"min_length": 3}).json()
|
|
assert len(events) == 0
|
|
|
|
def test_get_event_list_limit(self):
|
|
now = datetime.now().timestamp()
|
|
id = "123456.random"
|
|
id2 = "54321.random"
|
|
|
|
with AuthTestClient(self.app) as client:
|
|
super().insert_mock_event(id, start_time=now + 1)
|
|
events = client.get("/events").json()
|
|
assert len(events) == 1
|
|
assert events[0]["id"] == id
|
|
|
|
super().insert_mock_event(id2, start_time=now)
|
|
events = client.get("/events").json()
|
|
assert len(events) == 2
|
|
|
|
events = client.get("/events", params={"limit": 1}).json()
|
|
assert len(events) == 1
|
|
assert events[0]["id"] == id
|
|
|
|
events = client.get("/events", params={"limit": 3}).json()
|
|
assert len(events) == 2
|
|
|
|
def test_get_event_list_no_match_has_clip(self):
|
|
now = int(datetime.now().timestamp())
|
|
|
|
with AuthTestClient(self.app) as client:
|
|
id = "123456.random"
|
|
super().insert_mock_event(id, now, now + 2)
|
|
events = client.get("/events", params={"has_clip": 0}).json()
|
|
assert len(events) == 0
|
|
|
|
def test_get_event_list_has_clip(self):
|
|
with AuthTestClient(self.app) as client:
|
|
id = "123456.random"
|
|
super().insert_mock_event(id, has_clip=True)
|
|
events = client.get("/events", params={"has_clip": 1}).json()
|
|
assert len(events) == 1
|
|
assert events[0]["id"] == id
|
|
|
|
def test_get_event_list_sort_score(self):
|
|
with AuthTestClient(self.app) as client:
|
|
id = "123456.random"
|
|
id2 = "54321.random"
|
|
super().insert_mock_event(id, top_score=37, score=37, data={"score": 50})
|
|
super().insert_mock_event(id2, top_score=47, score=47, data={"score": 20})
|
|
events = client.get("/events", params={"sort": "score_asc"}).json()
|
|
assert len(events) == 2
|
|
assert events[0]["id"] == id2
|
|
assert events[1]["id"] == id
|
|
|
|
events = client.get("/events", params={"sort": "score_desc"}).json()
|
|
assert len(events) == 2
|
|
assert events[0]["id"] == id
|
|
assert events[1]["id"] == id2
|
|
|
|
def test_get_event_list_sort_start_time(self):
|
|
now = int(datetime.now().timestamp())
|
|
|
|
with AuthTestClient(self.app) as client:
|
|
id = "123456.random"
|
|
id2 = "54321.random"
|
|
super().insert_mock_event(id, start_time=now + 3)
|
|
super().insert_mock_event(id2, start_time=now)
|
|
events = client.get("/events", params={"sort": "date_asc"}).json()
|
|
assert len(events) == 2
|
|
assert events[0]["id"] == id2
|
|
assert events[1]["id"] == id
|
|
|
|
events = client.get("/events", params={"sort": "date_desc"}).json()
|
|
assert len(events) == 2
|
|
assert events[0]["id"] == id
|
|
assert events[1]["id"] == id2
|
|
|
|
def test_get_event_list_offset_pages_score_sort(self):
|
|
now = datetime.now().timestamp()
|
|
scores = [0.6, 0.9, 0.7, 0.95, 0.8]
|
|
|
|
with AuthTestClient(self.app) as client:
|
|
for i, score in enumerate(scores):
|
|
super().insert_mock_event(
|
|
f"event-{i}", start_time=now + i, data={"score": score}
|
|
)
|
|
|
|
params = {"sort": "score_desc"}
|
|
full = [e["id"] for e in client.get("/events", params=params).json()]
|
|
paged = [
|
|
e["id"]
|
|
for offset in (0, 2, 4)
|
|
for e in client.get(
|
|
"/events", params={**params, "limit": 2, "offset": offset}
|
|
).json()
|
|
]
|
|
|
|
assert full == ["event-3", "event-1", "event-4", "event-2", "event-0"]
|
|
assert paged == full
|
|
|
|
def test_get_event_list_match_multilingual_attribute(self):
|
|
event_id = "123456.zh"
|
|
attribute = "中文标签"
|
|
|
|
with AuthTestClient(self.app) as client:
|
|
super().insert_mock_event(event_id, data={"custom_attr": attribute})
|
|
|
|
events = client.get("/events", params={"attributes": attribute}).json()
|
|
assert len(events) == 1
|
|
assert events[0]["id"] == event_id
|
|
|
|
events = client.get(
|
|
"/events", params={"attributes": "%E4%B8%AD%E6%96%87%E6%A0%87%E7%AD%BE"}
|
|
).json()
|
|
assert len(events) == 1
|
|
assert events[0]["id"] == event_id
|
|
|
|
def test_events_search_match_multilingual_attribute(self):
|
|
event_id = "123456.zh.search"
|
|
attribute = "中文标签"
|
|
mock_embeddings = Mock()
|
|
mock_embeddings.search_thumbnail.return_value = [(event_id, 0.05)]
|
|
|
|
self.app.frigate_config.semantic_search.enabled = True
|
|
self.app.embeddings = mock_embeddings
|
|
|
|
with AuthTestClient(self.app) as client:
|
|
super().insert_mock_event(event_id, data={"custom_attr": attribute})
|
|
|
|
events = client.get(
|
|
"/events/search",
|
|
params={
|
|
"search_type": "similarity",
|
|
"event_id": event_id,
|
|
"attributes": attribute,
|
|
},
|
|
).json()
|
|
assert len(events) == 1
|
|
assert events[0]["id"] == event_id
|
|
|
|
events = client.get(
|
|
"/events/search",
|
|
params={
|
|
"search_type": "similarity",
|
|
"event_id": event_id,
|
|
"attributes": "%E4%B8%AD%E6%96%87%E6%A0%87%E7%AD%BE",
|
|
},
|
|
).json()
|
|
assert len(events) == 1
|
|
assert events[0]["id"] == event_id
|
|
|
|
def test_events_search_offset_pages_score_sort(self):
|
|
now = datetime.now().timestamp()
|
|
scores = [0.6, 0.9, 0.7, 0.95, 0.8]
|
|
ids = [f"event-{i}" for i in range(len(scores))]
|
|
mock_embeddings = Mock()
|
|
mock_embeddings.search_thumbnail.return_value = [
|
|
(event_id, 0.1 * i) for i, event_id in enumerate(ids)
|
|
]
|
|
|
|
self.app.frigate_config.semantic_search.enabled = True
|
|
self.app.embeddings = mock_embeddings
|
|
|
|
with AuthTestClient(self.app) as client:
|
|
for i, score in enumerate(scores):
|
|
super().insert_mock_event(
|
|
ids[i], start_time=now + i, data={"score": score}
|
|
)
|
|
|
|
params = {
|
|
"search_type": "similarity",
|
|
"event_id": ids[0],
|
|
"sort": "score_desc",
|
|
}
|
|
paged = [
|
|
e["id"]
|
|
for offset in (0, 2, 4)
|
|
for e in client.get(
|
|
"/events/search",
|
|
params={**params, "limit": 2, "offset": offset},
|
|
).json()
|
|
]
|
|
|
|
assert paged == ["event-3", "event-1", "event-4", "event-2", "event-0"]
|
|
|
|
def test_events_search_offset_pages_orders_ties_by_id(self):
|
|
now = datetime.now().timestamp()
|
|
ids = ["event-c", "event-a", "event-b"]
|
|
mock_embeddings = Mock()
|
|
mock_embeddings.search_thumbnail.return_value = [
|
|
(event_id, 0.1) for event_id in ids
|
|
]
|
|
|
|
self.app.frigate_config.semantic_search.enabled = True
|
|
self.app.embeddings = mock_embeddings
|
|
|
|
with AuthTestClient(self.app) as client:
|
|
for i, event_id in enumerate(ids):
|
|
super().insert_mock_event(
|
|
event_id, start_time=now + i, data={"score": 0.8}
|
|
)
|
|
|
|
for sort in ("score_desc", "relevance"):
|
|
params = {
|
|
"search_type": "similarity",
|
|
"event_id": ids[0],
|
|
"sort": sort,
|
|
}
|
|
paged = [
|
|
e["id"]
|
|
for offset in (0, 1, 2)
|
|
for e in client.get(
|
|
"/events/search",
|
|
params={**params, "limit": 1, "offset": offset},
|
|
).json()
|
|
]
|
|
|
|
assert paged == ["event-a", "event-b", "event-c"]
|
|
|
|
def test_event_list_rejects_negative_offset(self):
|
|
with AuthTestClient(self.app) as client:
|
|
response = client.get("/events", params={"offset": -5})
|
|
assert response.status_code == 422
|
|
|
|
response = client.get(
|
|
"/events/search",
|
|
params={"query": "car", "offset": -5},
|
|
)
|
|
assert response.status_code == 422
|
|
|
|
def test_similarity_search_hides_unauthorized_anchor_event(self):
|
|
mock_embeddings = Mock()
|
|
self.app.frigate_config.semantic_search.enabled = True
|
|
self.app.embeddings = mock_embeddings
|
|
|
|
with AuthTestClient(self.app) as client:
|
|
super().insert_mock_event("hidden.anchor", camera="back_door")
|
|
response = client.get(
|
|
"/events/search",
|
|
params={
|
|
"search_type": "similarity",
|
|
"event_id": "hidden.anchor",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 404
|
|
assert response.json()["message"] == "Event not found"
|
|
mock_embeddings.search_thumbnail.assert_not_called()
|
|
|
|
def test_get_good_event(self):
|
|
id = "123456.random"
|
|
|
|
with AuthTestClient(self.app) as client:
|
|
super().insert_mock_event(id)
|
|
event = client.get(f"/events/{id}").json()
|
|
|
|
assert event
|
|
assert event["id"] == id
|
|
assert event["id"] == model_to_dict(Event.get(Event.id == id))["id"]
|
|
|
|
def test_get_bad_event(self):
|
|
id = "123456.random"
|
|
bad_id = "654321.other"
|
|
|
|
with AuthTestClient(self.app) as client:
|
|
super().insert_mock_event(id)
|
|
event_response = client.get(f"/events/{bad_id}")
|
|
assert event_response.status_code == 404
|
|
assert event_response.json() == "Event not found"
|
|
|
|
def test_delete_event(self):
|
|
id = "123456.random"
|
|
|
|
with AuthTestClient(self.app) as client:
|
|
super().insert_mock_event(id)
|
|
event = client.get(f"/events/{id}").json()
|
|
assert event
|
|
assert event["id"] == id
|
|
response = client.delete(f"/events/{id}", headers={"remote-role": "admin"})
|
|
assert response.status_code == 200
|
|
event_after_delete = client.get(f"/events/{id}")
|
|
assert event_after_delete.status_code == 404
|
|
|
|
def test_event_retention(self):
|
|
id = "123456.random"
|
|
|
|
with AuthTestClient(self.app) as client:
|
|
super().insert_mock_event(id)
|
|
client.post(f"/events/{id}/retain", headers={"remote-role": "admin"})
|
|
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", headers={"remote-role": "admin"})
|
|
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):
|
|
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 AuthTestClient(self.app) as client:
|
|
super().insert_mock_event(morning_id, morning)
|
|
super().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",
|
|
params={"time_range": "07:00,24:00"},
|
|
).json()
|
|
assert events
|
|
assert len(events) == 1
|
|
# evening event is excluded
|
|
events = client.get(
|
|
"/events",
|
|
params={"time_range": "00:00,18:00"},
|
|
).json()
|
|
assert events
|
|
assert len(events) == 1
|
|
|
|
def test_set_delete_sub_label(self):
|
|
mock_event_updater = Mock(spec=EventMetadataPublisher)
|
|
app = super().create_app(event_metadata_publisher=mock_event_updater)
|
|
id = "123456.random"
|
|
sub_label = "sub"
|
|
|
|
def update_event(payload: Any, topic: str):
|
|
event = Event.get(id=id)
|
|
event.sub_label = payload[1]
|
|
event.save()
|
|
|
|
mock_event_updater.publish.side_effect = update_event
|
|
|
|
with AuthTestClient(app) as client:
|
|
super().insert_mock_event(id)
|
|
new_sub_label_response = client.post(
|
|
f"/events/{id}/sub_label",
|
|
json={"subLabel": sub_label},
|
|
headers={"remote-role": "admin"},
|
|
)
|
|
assert new_sub_label_response.status_code == 200
|
|
event = client.get(f"/events/{id}").json()
|
|
assert event
|
|
assert event["id"] == id
|
|
assert event["sub_label"] == sub_label
|
|
empty_sub_label_response = client.post(
|
|
f"/events/{id}/sub_label",
|
|
json={"subLabel": ""},
|
|
headers={"remote-role": "admin"},
|
|
)
|
|
assert empty_sub_label_response.status_code == 200
|
|
event = client.get(f"/events/{id}").json()
|
|
assert event
|
|
assert event["id"] == id
|
|
assert event["sub_label"] == None
|
|
|
|
def test_sub_label_list(self):
|
|
mock_event_updater = Mock(spec=EventMetadataPublisher)
|
|
app = super().create_app(event_metadata_publisher=mock_event_updater)
|
|
app.event_metadata_publisher = mock_event_updater
|
|
id = "123456.random"
|
|
sub_label = "sub"
|
|
|
|
def update_event(payload: Any, _: str):
|
|
event = Event.get(id=id)
|
|
event.sub_label = payload[1]
|
|
event.save()
|
|
|
|
mock_event_updater.publish.side_effect = update_event
|
|
|
|
with AuthTestClient(app) as client:
|
|
super().insert_mock_event(id)
|
|
client.post(
|
|
f"/events/{id}/sub_label",
|
|
json={"subLabel": sub_label},
|
|
headers={"remote-role": "admin"},
|
|
)
|
|
sub_labels = client.get("/sub_labels").json()
|
|
assert sub_labels
|
|
assert sub_labels == [sub_label]
|
|
|
|
####################################################################################################################
|
|
################################### GET /metrics Endpoint #########################################################
|
|
####################################################################################################################
|
|
def test_get_metrics(self):
|
|
"""ensure correct prometheus metrics api response"""
|
|
with AuthTestClient(self.app) as client:
|
|
ts_start = datetime.now().timestamp()
|
|
ts_end = ts_start + 30
|
|
_insert_mock_event(
|
|
id="abcde.random", start=ts_start, end=ts_end, retain=True
|
|
)
|
|
_insert_mock_event(
|
|
id="01234.random", start=ts_start, end=ts_end, retain=True
|
|
)
|
|
_insert_mock_event(
|
|
id="56789.random", start=ts_start, end=ts_end, retain=True
|
|
)
|
|
_insert_mock_event(
|
|
id="101112.random",
|
|
label="outside",
|
|
start=ts_start,
|
|
end=ts_end,
|
|
retain=True,
|
|
)
|
|
_insert_mock_event(
|
|
id="131415.random",
|
|
label="outside",
|
|
start=ts_start,
|
|
end=ts_end,
|
|
retain=True,
|
|
)
|
|
_insert_mock_event(
|
|
id="161718.random",
|
|
camera="porch",
|
|
start=ts_start,
|
|
end=ts_end,
|
|
retain=True,
|
|
)
|
|
_insert_mock_event(
|
|
id="192021.random",
|
|
camera="porch",
|
|
start=ts_start,
|
|
end=ts_end,
|
|
retain=True,
|
|
)
|
|
_insert_mock_event(
|
|
id="222324.random",
|
|
camera="porch",
|
|
label="inside",
|
|
start=ts_start,
|
|
end=ts_end,
|
|
retain=True,
|
|
)
|
|
_insert_mock_event(
|
|
id="252627.random",
|
|
camera="porch",
|
|
label="inside",
|
|
start=ts_start,
|
|
end=ts_end,
|
|
retain=True,
|
|
)
|
|
_insert_mock_event(
|
|
id="282930.random",
|
|
label="inside",
|
|
start=ts_start,
|
|
end=ts_end,
|
|
retain=True,
|
|
)
|
|
_insert_mock_event(
|
|
id="313233.random",
|
|
label="inside",
|
|
start=ts_start,
|
|
end=ts_end,
|
|
retain=True,
|
|
)
|
|
|
|
stats_emitter = Mock(spec=StatsEmitter)
|
|
stats_emitter.get_latest_stats.return_value = self.test_stats
|
|
self.app.stats_emitter = stats_emitter
|
|
event = client.get("/metrics")
|
|
|
|
assert "# TYPE frigate_detection_total_fps gauge" in event.text
|
|
assert "frigate_detection_total_fps 13.7" in event.text
|
|
assert (
|
|
"# HELP frigate_camera_events Current number of events stored in the database per camera and label"
|
|
in event.text
|
|
)
|
|
assert "# TYPE frigate_camera_events gauge" in event.text
|
|
assert (
|
|
'frigate_camera_events{camera="front_door",label="Mock"} 3.0' in event.text
|
|
)
|
|
assert (
|
|
'frigate_camera_events{camera="front_door",label="inside"} 2.0'
|
|
in event.text
|
|
)
|
|
assert (
|
|
'frigate_camera_events{camera="front_door",label="outside"} 2.0'
|
|
in event.text
|
|
)
|
|
assert 'frigate_camera_events{camera="porch",label="Mock"} 2.0' in event.text
|
|
assert 'frigate_camera_events{camera="porch",label="inside"} 2.0' in event.text
|