mirror of
https://github.com/blakeblackshear/frigate.git
synced 2026-07-20 18:59:01 +03:00
Josh HawkinsandGitHub
6f80bcd19f
CI / AMD64 Build (push) Has been cancelled
CI / ARM Build (push) Has been cancelled
CI / Jetson Jetpack 6 (push) Has been cancelled
CI / AMD64 Extra Build (push) Has been cancelled
CI / ARM Extra Build (push) Has been cancelled
CI / Synaptics Build (push) Has been cancelled
CI / Assemble and push default build (push) Has been cancelled
* resolve saved credential sentinel to the stored api_key in the GenAI probe * add profile faq * center the multi-camera export time range on the current playback position * add faq about preview restart cache * clarify exports bulk download
254 lines
9.6 KiB
Python
254 lines
9.6 KiB
Python
from unittest.mock import Mock, patch
|
|
|
|
import frigate.genai
|
|
from frigate.config import GenAIProviderEnum
|
|
from frigate.const import 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
|
|
|
|
####################################################################################################################
|
|
################################### 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
|