mirror of
https://github.com/blakeblackshear/frigate.git
synced 2026-11-13 17:13:01 +03:00
Josh HawkinsandGitHub
e61ad52b67
CI / AMD64 Build (push) Canceled after 0s
CI / AMD64 Smoke Test (push) Canceled after 0s
CI / ARM Build (push) Canceled after 0s
CI / Jetson Jetpack 6 (push) Canceled after 0s
CI / AMD64 Extra Build (push) Canceled after 0s
CI / ARM Extra Build (push) Canceled after 0s
CI / Synaptics Build (push) Canceled after 0s
CI / Assemble and push default build (push) Canceled after 0s
go2rtc's create_config.py imported frigate.config.env, which ran the frigate.config package init and loaded every detector plugin. Since the onnxruntime 1.30 bump, importing onnxruntime takes about 5s on some hosts, so go2rtc started after Frigate's cameras and every ffmpeg process failed to connect to the restream until it came up. The env helpers now live in frigate.util.env, so go2rtc's config script no longer imports onnxruntime.
338 lines
13 KiB
Python
338 lines
13 KiB
Python
import json
|
|
import os
|
|
from unittest.mock import Mock, patch
|
|
|
|
import frigate.genai
|
|
from frigate.config import GenAIProviderEnum
|
|
from frigate.const import MODEL_CACHE_DIR, REDACTED_CREDENTIAL_SENTINEL
|
|
from frigate.genai import GenAIClient
|
|
from frigate.models import Event, Recordings, ReviewSegment
|
|
from frigate.stats.emitter import StatsEmitter
|
|
from frigate.test.http_api.base_http_test import AuthTestClient, BaseTestHttp
|
|
from frigate.util.env import FRIGATE_ENV_VARS
|
|
|
|
|
|
class TestHttpApp(BaseTestHttp):
|
|
def setUp(self):
|
|
super().setUp([Event, Recordings, ReviewSegment])
|
|
self.app = super().create_app()
|
|
|
|
####################################################################################################################
|
|
################################### GET /stats Endpoint #########################################################
|
|
####################################################################################################################
|
|
def test_stats_endpoint(self):
|
|
stats = Mock(spec=StatsEmitter)
|
|
stats.get_latest_stats.return_value = self.test_stats
|
|
app = super().create_app(stats)
|
|
|
|
with AuthTestClient(app) as client:
|
|
response = client.get("/stats")
|
|
response_json = response.json()
|
|
assert response_json == self.test_stats
|
|
|
|
def test_recordings_storage_requires_admin(self):
|
|
stats = Mock(spec=StatsEmitter)
|
|
stats.get_latest_stats.return_value = self.test_stats
|
|
app = super().create_app(stats)
|
|
app.storage_maintainer = Mock()
|
|
app.storage_maintainer.calculate_camera_usages.return_value = {
|
|
"front_door": {"usage": 2.0},
|
|
}
|
|
|
|
with AuthTestClient(app) as client:
|
|
response = client.get(
|
|
"/recordings/storage",
|
|
headers={"remote-user": "viewer", "remote-role": "viewer"},
|
|
)
|
|
assert response.status_code == 403
|
|
|
|
response = client.get("/recordings/storage")
|
|
assert response.status_code == 200
|
|
assert response.json()["front_door"]["usage_percent"] == 25.0
|
|
|
|
def test_camera_name_collision_keeps_admin_default(self):
|
|
self.minimal_config["cameras"]["faces"] = self.minimal_config["cameras"].pop(
|
|
"front_door"
|
|
)
|
|
app = super().create_app(enforce_default_admin=True)
|
|
viewer = {"remote-user": "viewer", "remote-role": "viewer"}
|
|
|
|
with AuthTestClient(app) as client:
|
|
assert client.get("/faces", headers=viewer).status_code == 403
|
|
assert client.get("/faces").status_code == 200
|
|
assert (
|
|
client.post("/faces/train/person/classify", headers=viewer).status_code
|
|
== 403
|
|
)
|
|
|
|
# Camera routes for the same name stay reachable by viewers
|
|
response = client.get("/faces/recordings/summary", headers=viewer)
|
|
assert response.status_code == 200
|
|
|
|
def test_config_set_in_memory_replaces_objects_track_list(self):
|
|
self.minimal_config["cameras"]["front_door"]["objects"] = {
|
|
"track": ["person", "car"],
|
|
}
|
|
app = super().create_app()
|
|
app.config_publisher = Mock()
|
|
|
|
with AuthTestClient(app) as client:
|
|
response = client.put(
|
|
"/config/set",
|
|
json={
|
|
"requires_restart": 0,
|
|
"skip_save": True,
|
|
"update_topic": "config/cameras/front_door/objects",
|
|
"config_data": {
|
|
"cameras": {
|
|
"front_door": {
|
|
"objects": {
|
|
"track": ["person"],
|
|
}
|
|
}
|
|
}
|
|
},
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert app.frigate_config.cameras["front_door"].objects.track == ["person"]
|
|
|
|
####################################################################################################################
|
|
################################### Credential redaction sentinel ################################################
|
|
####################################################################################################################
|
|
def test_config_response_redacts_mqtt_password_with_sentinel(self):
|
|
self.minimal_config["mqtt"]["user"] = "mqttuser"
|
|
self.minimal_config["mqtt"]["password"] = "supersecret"
|
|
app = super().create_app()
|
|
|
|
with AuthTestClient(app) as client:
|
|
response = client.get("/config")
|
|
assert response.status_code == 200
|
|
mqtt = response.json()["mqtt"]
|
|
assert mqtt["password"] == REDACTED_CREDENTIAL_SENTINEL
|
|
|
|
def test_config_response_hides_notification_email_from_viewers(self):
|
|
self.minimal_config["notifications"] = {"email": "{FRIGATE_TEST_EMAIL}"}
|
|
|
|
with patch.dict(FRIGATE_ENV_VARS, {"FRIGATE_TEST_EMAIL": "me@example.com"}):
|
|
app = super().create_app()
|
|
|
|
assert app.frigate_config.notifications.email == "me@example.com"
|
|
|
|
with AuthTestClient(app) as client:
|
|
response = client.get(
|
|
"/config",
|
|
headers={"remote-user": "viewer", "remote-role": "viewer"},
|
|
)
|
|
assert response.status_code == 200
|
|
config = response.json()
|
|
assert config["notifications"]["email"] == REDACTED_CREDENTIAL_SENTINEL
|
|
assert (
|
|
config["cameras"]["front_door"]["notifications"]["email"]
|
|
== REDACTED_CREDENTIAL_SENTINEL
|
|
)
|
|
|
|
response = client.get("/config")
|
|
assert response.json()["notifications"]["email"] == "me@example.com"
|
|
|
|
def test_config_response_keeps_plus_model_reference(self):
|
|
model_id = "test_plus_reference"
|
|
model_path = os.path.join(MODEL_CACHE_DIR, model_id)
|
|
os.makedirs(MODEL_CACHE_DIR, exist_ok=True)
|
|
|
|
with open(model_path, "w") as f:
|
|
f.write("model")
|
|
|
|
with open(f"{model_path}.json", "w") as f:
|
|
json.dump(
|
|
{
|
|
"id": model_id,
|
|
"type": "ssd",
|
|
"supportedDetectors": ["cpu"],
|
|
"width": 320,
|
|
"height": 320,
|
|
"inputShape": "nhwc",
|
|
"pixelFormat": "rgb",
|
|
"labelMap": {"0": "person"},
|
|
},
|
|
f,
|
|
)
|
|
|
|
self.addCleanup(os.remove, model_path)
|
|
self.addCleanup(os.remove, f"{model_path}.json")
|
|
self.minimal_config["models"] = [
|
|
{"path": f"plus://{model_id}", "devices": ["cpu"]}
|
|
]
|
|
app = super().create_app()
|
|
|
|
with AuthTestClient(app) as client:
|
|
response = client.get("/config")
|
|
assert response.status_code == 200
|
|
assert response.json()["models"][0]["path"] == f"plus://{model_id}"
|
|
|
|
# detection still loads the resolved cache file
|
|
assert app.frigate_config.models[0].path == model_path
|
|
|
|
####################################################################################################################
|
|
################################### POST /genai/probe Endpoint ##################################################
|
|
####################################################################################################################
|
|
def test_genai_probe_requires_admin(self):
|
|
app = super().create_app()
|
|
|
|
with AuthTestClient(app) as client:
|
|
response = client.post(
|
|
"/genai/probe",
|
|
json={"provider": "openai"},
|
|
headers={"remote-user": "viewer", "remote-role": "viewer"},
|
|
)
|
|
assert response.status_code == 403
|
|
|
|
def test_genai_probe_returns_models_from_transient_client(self):
|
|
class FakeClient(GenAIClient):
|
|
def list_models(self):
|
|
return ["fake-model-a", "fake-model-b"]
|
|
|
|
app = super().create_app()
|
|
|
|
with (
|
|
AuthTestClient(app) as client,
|
|
patch.dict(
|
|
frigate.genai.PROVIDERS,
|
|
{GenAIProviderEnum.openai: FakeClient},
|
|
),
|
|
):
|
|
response = client.post(
|
|
"/genai/probe",
|
|
json={
|
|
"provider": "openai",
|
|
"api_key": "sk-test",
|
|
"base_url": "https://example.invalid",
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json() == {
|
|
"success": True,
|
|
"models": ["fake-model-a", "fake-model-b"],
|
|
}
|
|
|
|
def test_genai_probe_resolves_sentinel_to_saved_api_key(self):
|
|
# After a save the UI's api_key field holds the redaction sentinel;
|
|
# the probe must substitute the saved key for the named entry instead
|
|
# of sending the literal sentinel to the provider (GH discussion 23754).
|
|
probed_keys: list[str | None] = []
|
|
|
|
class CapturingClient(GenAIClient):
|
|
def list_models(self):
|
|
probed_keys.append(self.genai_config.api_key)
|
|
return ["fake-model"]
|
|
|
|
self.minimal_config["genai"] = {
|
|
"llm": {
|
|
"provider": "openai",
|
|
"api_key": "sk-saved",
|
|
"base_url": "https://example.invalid",
|
|
"model": "fake-model",
|
|
}
|
|
}
|
|
app = super().create_app()
|
|
|
|
with (
|
|
AuthTestClient(app) as client,
|
|
patch.dict(
|
|
frigate.genai.PROVIDERS,
|
|
{GenAIProviderEnum.openai: CapturingClient},
|
|
),
|
|
):
|
|
response = client.post(
|
|
"/genai/probe",
|
|
json={
|
|
"provider": "openai",
|
|
"name": "llm",
|
|
"api_key": REDACTED_CREDENTIAL_SENTINEL,
|
|
"base_url": "https://example.invalid",
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json()["success"] is True
|
|
assert probed_keys == ["sk-saved"]
|
|
|
|
def test_genai_probe_sentinel_without_saved_entry_sends_no_key(self):
|
|
# If the sentinel arrives for an entry that has no saved config, the
|
|
# probe must drop the key entirely rather than leak the sentinel.
|
|
probed_keys: list[str | None] = []
|
|
|
|
class CapturingClient(GenAIClient):
|
|
def list_models(self):
|
|
probed_keys.append(self.genai_config.api_key)
|
|
return ["fake-model"]
|
|
|
|
app = super().create_app()
|
|
|
|
with (
|
|
AuthTestClient(app) as client,
|
|
patch.dict(
|
|
frigate.genai.PROVIDERS,
|
|
{GenAIProviderEnum.openai: CapturingClient},
|
|
),
|
|
):
|
|
response = client.post(
|
|
"/genai/probe",
|
|
json={
|
|
"provider": "openai",
|
|
"name": "llm",
|
|
"api_key": REDACTED_CREDENTIAL_SENTINEL,
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
assert probed_keys == [None]
|
|
|
|
def test_genai_probe_empty_list_is_treated_as_failure(self):
|
|
# The plugin's list_models() returns [] on connection failure rather
|
|
# than raising. The endpoint should surface that as success=false so
|
|
# the UI can show a meaningful error.
|
|
class EmptyClient(GenAIClient):
|
|
def list_models(self):
|
|
return []
|
|
|
|
app = super().create_app()
|
|
|
|
with (
|
|
AuthTestClient(app) as client,
|
|
patch.dict(
|
|
frigate.genai.PROVIDERS,
|
|
{GenAIProviderEnum.openai: EmptyClient},
|
|
),
|
|
):
|
|
response = client.post(
|
|
"/genai/probe",
|
|
json={"provider": "openai"},
|
|
)
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload["success"] is False
|
|
assert "message" in payload
|
|
|
|
def test_genai_probe_handles_provider_failure(self):
|
|
class FailingClient(GenAIClient):
|
|
def list_models(self):
|
|
raise RuntimeError("provider unreachable")
|
|
|
|
app = super().create_app()
|
|
|
|
with (
|
|
AuthTestClient(app) as client,
|
|
patch.dict(
|
|
frigate.genai.PROVIDERS,
|
|
{GenAIProviderEnum.openai: FailingClient},
|
|
),
|
|
):
|
|
response = client.post(
|
|
"/genai/probe",
|
|
json={"provider": "openai"},
|
|
)
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload["success"] is False
|
|
assert "message" in payload
|