mirror of
https://github.com/blakeblackshear/frigate.git
synced 2026-09-29 03:16:51 +03:00
Josh Hawkins
225e014979
`/api/config` served a Frigate+ model's path as the resolved `/config/model_cache/<id>` file, and since the models list is saved whole, editing any model in the settings UI wrote that cache path back to the config in place of `plus://<id>`. After a restart the model loaded as a custom model with the default labelmap. The config API now reports the `plus://<id>` reference the model was configured with, and the models section drops the fields the Frigate+ model info supplies (size, tensor, pixel format, dtype, and type) instead of pinning them in the config.
294 lines
11 KiB
Python
294 lines
11 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
|
|
|
|
|
|
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_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_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
|