mirror of
https://github.com/blakeblackshear/frigate.git
synced 2026-11-09 15:13:00 +03:00
Blake BlackshearandClaude Opus 5.5
b10c8b7498
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
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01E4iDiYSpWJD13YQUdNKC5o
89 lines
3.3 KiB
Python
89 lines
3.3 KiB
Python
"""Tests for the go2rtc ffmpeg and ngrok options applied at startup."""
|
|
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from frigate.util.services import (
|
|
apply_go2rtc_ffmpeg_config,
|
|
is_restricted_go2rtc_api_source,
|
|
is_restricted_go2rtc_source,
|
|
)
|
|
|
|
FFMPEG_PATH = "/usr/lib/ffmpeg/7.0/bin/ffmpeg"
|
|
|
|
|
|
class TestApplyGo2rtcFfmpegConfig(unittest.TestCase):
|
|
def setUp(self):
|
|
# the override is snapshotted from the environment at import time
|
|
exec_patch = patch("frigate.util.services._GO2RTC_ARBITRARY_EXEC_ENV", None)
|
|
exec_patch.start()
|
|
self.addCleanup(exec_patch.stop)
|
|
|
|
def test_sets_bin_when_missing(self):
|
|
config = {}
|
|
assert apply_go2rtc_ffmpeg_config(config, FFMPEG_PATH) == []
|
|
assert config["ffmpeg"] == {"bin": FFMPEG_PATH}
|
|
|
|
def test_forces_bin_override(self):
|
|
config = {"ffmpeg": {"bin": "/bin/sh", "global": "-c 'id' x"}}
|
|
assert apply_go2rtc_ffmpeg_config(config, FFMPEG_PATH) == ["ffmpeg.bin"]
|
|
assert config["ffmpeg"]["bin"] == FFMPEG_PATH
|
|
|
|
def test_keeps_saved_templates(self):
|
|
config = {"ffmpeg": {"global": "-vaapi_device /dev/dri/renderD128"}}
|
|
assert apply_go2rtc_ffmpeg_config(config, FFMPEG_PATH) == []
|
|
assert config["ffmpeg"]["global"] == "-vaapi_device /dev/dri/renderD128"
|
|
|
|
def test_drops_ngrok(self):
|
|
config = {"ngrok": {"command": "/bin/sh -c id"}}
|
|
assert apply_go2rtc_ffmpeg_config(config, FFMPEG_PATH) == ["ngrok"]
|
|
assert "ngrok" not in config
|
|
|
|
def test_allows_bin_and_ngrok_when_override_set(self):
|
|
config = {
|
|
"ffmpeg": {"bin": "/usr/bin/ffmpeg"},
|
|
"ngrok": {"command": "ngrok tcp 8555"},
|
|
}
|
|
with patch("frigate.util.services._GO2RTC_ARBITRARY_EXEC_ENV", "true"):
|
|
assert apply_go2rtc_ffmpeg_config(config, FFMPEG_PATH) == []
|
|
assert config["ffmpeg"]["bin"] == "/usr/bin/ffmpeg"
|
|
assert config["ngrok"] == {"command": "ngrok tcp 8555"}
|
|
|
|
|
|
class TestRestrictedGo2rtcVarExpansion(unittest.TestCase):
|
|
"""go2rtc expands `${VAR:default}` over the raw config before parsing, and
|
|
the default can name an exec source, so a `${` source is restricted."""
|
|
|
|
def setUp(self):
|
|
exec_patch = patch(
|
|
"frigate.util.services.is_go2rtc_arbitrary_exec_allowed",
|
|
return_value=False,
|
|
)
|
|
self.mock_exec_allowed = exec_patch.start()
|
|
self.addCleanup(exec_patch.stop)
|
|
|
|
def test_var_expansion_source_is_restricted(self):
|
|
for source in (
|
|
"${NOPE:exec:touch /tmp/x}",
|
|
"rtsp://host/${CAM}",
|
|
" ${NOPE:exec:id}",
|
|
):
|
|
self.assertTrue(is_restricted_go2rtc_source(source), source)
|
|
self.assertTrue(is_restricted_go2rtc_api_source(source), source)
|
|
|
|
def test_plain_sources_are_not_restricted(self):
|
|
for source in (
|
|
"rtsp://user:pass@host/stream",
|
|
"ffmpeg:front#video=h264#height=720",
|
|
):
|
|
self.assertFalse(is_restricted_go2rtc_source(source), source)
|
|
self.assertFalse(is_restricted_go2rtc_api_source(source), source)
|
|
|
|
def test_var_expansion_allowed_when_override_set(self):
|
|
self.mock_exec_allowed.return_value = True
|
|
self.assertFalse(is_restricted_go2rtc_source("${NOPE:exec:id}"))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|