feat(v1): structured camera config and RTSP url builder

Config accepts structured host/port/channel/username/password (or the
existing rtsp_url_env) and builds the RTSP URL with RFC 3986 percent-
encoded credentials, so passwords containing @/:/ no longer break parsing.
Credentials are excluded from config_version; a guard test keeps the
public example credential-free. probe_stream does a bounded connection
test. 59 tests pass.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
ila
2026-07-21 22:38:21 +08:00
co-authored by Claude Opus 4.8
parent 59925b8e30
commit ab574f1a27
5 changed files with 217 additions and 8 deletions
+49
View File
@@ -0,0 +1,49 @@
import numpy as np
from v1.camera import build_rtsp_url, probe_stream
class _FakeCapture:
def __init__(self, frames):
self._frames = list(frames) if frames is not None else None
def isOpened(self):
return self._frames is not None
def read(self):
if self._frames:
return True, self._frames.pop(0)
return False, None
def release(self):
pass
def test_build_rtsp_url_percent_encodes_special_characters_in_password():
# Example TEST-NET address and a synthetic password (never a real credential).
url = build_rtsp_url("192.0.2.10", 554, "admin", "p@ss/w:d", "102")
assert url == "rtsp://admin:p%40ss%2Fw%3Ad@192.0.2.10:554/Streaming/Channels/102"
def test_build_rtsp_url_defaults_to_main_channel():
url = build_rtsp_url("192.0.2.10", 554, "admin", "secret")
assert url.endswith("/Streaming/Channels/101")
def test_probe_stream_returns_a_frame_on_success():
frame = np.zeros((480, 640, 3), dtype=np.uint8)
result = probe_stream("rtsp://x", capture_factory=lambda _u: _FakeCapture([frame]))
assert result.ok is True
assert "640x480" in result.message
assert result.frame is not None
def test_probe_stream_reports_failure_without_raising():
result = probe_stream("rtsp://x", attempts=3, capture_factory=lambda _u: _FakeCapture(None))
assert result.ok is False
assert result.frame is None
+62
View File
@@ -56,6 +56,68 @@ def test_load_config_rejects_embedded_source_address(tmp_path):
load_config(config_file)
def test_structured_source_builds_percent_encoded_rtsp_url(tmp_path):
config_file = tmp_path / "config.json"
_write_config(
config_file,
{
"id": "hik-ipcamera-101",
"host": "192.0.2.10",
"port": 554,
"channel": "102",
"username": "admin",
"password": "p@ss/w:d",
},
)
config = load_config(config_file)
assert config.source_url == "rtsp://admin:p%40ss%2Fw%3Ad@192.0.2.10:554/Streaming/Channels/102"
assert config.source_id == "hik-ipcamera-101"
assert config.source_mode == "stream"
def test_structured_source_requires_password(tmp_path):
config_file = tmp_path / "config.json"
_write_config(
config_file,
{"id": "hik-ipcamera-101", "host": "192.0.2.10", "username": "admin"},
)
with pytest.raises(ConfigError, match="password"):
load_config(config_file)
def test_structured_credentials_are_excluded_from_config_version(tmp_path):
first_file = tmp_path / "a.json"
second_file = tmp_path / "b.json"
_write_config(
first_file,
{"id": "hik-101", "host": "192.0.2.10", "username": "admin", "password": "secret-a"},
)
_write_config(
second_file,
{"id": "hik-101", "host": "192.0.2.20", "username": "operator", "password": "secret-b"},
)
first = load_config(first_file)
second = load_config(second_file)
assert first.source_url != second.source_url
assert first.runtime_config_version == second.runtime_config_version
def test_public_example_config_has_no_embedded_credentials():
from pathlib import Path
example = json.loads(Path("v1/config.example.json").read_text(encoding="utf-8"))
source = example["source"]
for forbidden in ("host", "username", "password", "url", "rtsp_url"):
assert forbidden not in source
assert "rtsp_url_env" in source
def test_source_mode_defaults_to_stream(tmp_path, monkeypatch):
config_file = tmp_path / "config.json"
_write_config(