From 552a6e5f101de24059abc09c596722e8e5d9671b Mon Sep 17 00:00:00 2001 From: Josh Hawkins <32435876+hawkeye217@users.noreply.github.com> Date: Tue, 7 Apr 2026 07:08:27 -0500 Subject: [PATCH] add tests --- frigate/test/test_env.py | 136 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 136 insertions(+) diff --git a/frigate/test/test_env.py b/frigate/test/test_env.py index fe2ce8d3e..b2bd635a9 100644 --- a/frigate/test/test_env.py +++ b/frigate/test/test_env.py @@ -2,6 +2,7 @@ import os import unittest +from unittest.mock import MagicMock, patch from frigate.config.env import ( FRIGATE_ENV_VARS, @@ -10,6 +11,71 @@ from frigate.config.env import ( ) +class TestGo2RtcAddStreamSubstitution(unittest.TestCase): + """Covers the API path: PUT /go2rtc/streams/{stream_name}. + + The route shells out to go2rtc via `requests.put`; we mock the HTTP call + and assert that the substituted `src` parameter handles the same mixed + {FRIGATE_*} + literal-brace strings as the config-loading path. + """ + + def setUp(self): + self._original_env_vars = dict(FRIGATE_ENV_VARS) + + def tearDown(self): + FRIGATE_ENV_VARS.clear() + FRIGATE_ENV_VARS.update(self._original_env_vars) + + def _call_route(self, src: str) -> str: + """Invoke go2rtc_add_stream and return the substituted src param.""" + from frigate.api import camera as camera_api + + captured = {} + + def fake_put(url, params=None, timeout=None): + captured["params"] = params + resp = MagicMock() + resp.ok = True + resp.text = "" + resp.status_code = 200 + return resp + + with patch.object(camera_api.requests, "put", side_effect=fake_put): + camera_api.go2rtc_add_stream( + request=MagicMock(), stream_name="cam1", src=src + ) + return captured["params"]["src"] + + def test_mixed_localtime_and_frigate_var(self): + """%{localtime\\:...} alongside {FRIGATE_USER} substitutes only the var.""" + FRIGATE_ENV_VARS["FRIGATE_USER"] = "admin" + src = ( + "ffmpeg:rtsp://host/s#raw=-vf " + "drawtext=text=%{localtime\\:%Y-%m-%d}:user={FRIGATE_USER}" + ) + self.assertEqual( + self._call_route(src), + "ffmpeg:rtsp://host/s#raw=-vf " + "drawtext=text=%{localtime\\:%Y-%m-%d}:user=admin", + ) + + def test_unknown_var_falls_back_to_raw_src(self): + """Existing route behavior: unknown {FRIGATE_*} keeps raw src.""" + src = "rtsp://host/{FRIGATE_NONEXISTENT}/stream" + self.assertEqual(self._call_route(src), src) + + def test_malformed_placeholder_rejected_via_api(self): + """Malformed FRIGATE placeholders raise (not silently passed through). + + Regression: previously camera.py caught any KeyError and fell back + to the raw src, so `{FRIGATE_FOO:>5}` was silently accepted via the + API while config loading rejected it. The helper now raises + ValueError for malformed syntax to keep the two paths consistent. + """ + with self.assertRaises(ValueError): + self._call_route("rtsp://host/{FRIGATE_FOO:>5}/stream") + + class TestEnvString(unittest.TestCase): def setUp(self): self._original_env_vars = dict(FRIGATE_ENV_VARS) @@ -43,6 +109,76 @@ class TestEnvString(unittest.TestCase): with self.assertRaises(KeyError): validate_env_string("{FRIGATE_NONEXISTENT_VAR}") + def test_non_frigate_braces_passthrough(self): + """Braces that are not {FRIGATE_*} placeholders pass through untouched. + + Regression test for ffmpeg drawtext expressions like + "%{localtime\\:%Y-%m-%d}" being mangled by str.format(). + """ + expr = ( + "ffmpeg:rtsp://127.0.0.1/src#raw=-vf " + "drawtext=text=%{localtime\\:%Y-%m-%d_%H\\:%M\\:%S}" + ":x=5:fontcolor=white" + ) + self.assertEqual(validate_env_string(expr), expr) + + def test_double_brace_escape_preserved(self): + """`{{output}}` collapses to `{output}` (documented go2rtc escape).""" + result = validate_env_string( + "exec:ffmpeg -i /media/file.mp4 -f rtsp {{output}}" + ) + self.assertEqual( + result, "exec:ffmpeg -i /media/file.mp4 -f rtsp {output}" + ) + + def test_double_brace_around_frigate_var(self): + """`{{FRIGATE_FOO}}` stays literal — escape takes precedence.""" + FRIGATE_ENV_VARS["FRIGATE_FOO"] = "bar" + self.assertEqual( + validate_env_string("{{FRIGATE_FOO}}"), "{FRIGATE_FOO}" + ) + + def test_mixed_frigate_var_and_braces(self): + """A FRIGATE_ var alongside literal single braces substitutes only the var.""" + FRIGATE_ENV_VARS["FRIGATE_USER"] = "admin" + result = validate_env_string( + "drawtext=text=%{localtime}:user={FRIGATE_USER}:x=5" + ) + self.assertEqual(result, "drawtext=text=%{localtime}:user=admin:x=5") + + def test_triple_braces_around_frigate_var(self): + """`{{{FRIGATE_FOO}}}` collapses like str.format(): `{bar}`.""" + FRIGATE_ENV_VARS["FRIGATE_FOO"] = "bar" + self.assertEqual(validate_env_string("{{{FRIGATE_FOO}}}"), "{bar}") + + def test_trailing_double_brace_after_var(self): + """`{FRIGATE_FOO}}}` collapses like str.format(): `bar}`.""" + FRIGATE_ENV_VARS["FRIGATE_FOO"] = "bar" + self.assertEqual(validate_env_string("{FRIGATE_FOO}}}"), "bar}") + + def test_leading_double_brace_then_var(self): + """`{{{FRIGATE_FOO}` collapses like str.format(): `{bar`.""" + FRIGATE_ENV_VARS["FRIGATE_FOO"] = "bar" + self.assertEqual(validate_env_string("{{{FRIGATE_FOO}"), "{bar") + + def test_malformed_unterminated_placeholder_raises(self): + """`{FRIGATE_FOO` (no closing brace) raises like str.format() did.""" + FRIGATE_ENV_VARS["FRIGATE_FOO"] = "bar" + with self.assertRaises(ValueError): + validate_env_string("prefix-{FRIGATE_FOO") + + def test_malformed_format_spec_raises(self): + """`{FRIGATE_FOO:>5}` (format spec) raises like str.format() did.""" + FRIGATE_ENV_VARS["FRIGATE_FOO"] = "bar" + with self.assertRaises(ValueError): + validate_env_string("{FRIGATE_FOO:>5}") + + def test_malformed_conversion_raises(self): + """`{FRIGATE_FOO!r}` (conversion) raises like str.format() did.""" + FRIGATE_ENV_VARS["FRIGATE_FOO"] = "bar" + with self.assertRaises(ValueError): + validate_env_string("{FRIGATE_FOO!r}") + class TestEnvVars(unittest.TestCase): def setUp(self):