silver_pose/v1/tests/test_config.py

import json

import pytest

from v1.config import ConfigError, load_config


def _write_config(path, source):
    path.write_text(
        json.dumps(
            {
                "source": source,
                "model": {
                    "path": "models/best.pt",
                    "sha256": "a" * 64,
                    "confidence_threshold": 0.25,
                },
                "event": {
                    "keypoint_confidence_threshold": 0.4,
                    "suspect_window_seconds": 0.5,
                    "confirm_window_seconds": 1.8,
                    "recovery_window_seconds": 2.0,
                    "cooldown_seconds": 10.0,
                },
                "artifacts": {"event_dir": "artifacts/events"},
            },
            ensure_ascii=False,
        ),
        encoding="utf-8",
    )


def test_load_config_resolves_source_from_environment_variable(tmp_path, monkeypatch):
    config_file = tmp_path / "config.json"
    _write_config(
        config_file,
        {"id": "lobby-camera-01", "rtsp_url_env": "SILVER_POSE_RTSP_URL"},
    )
    monkeypatch.setenv("SILVER_POSE_RTSP_URL", "rtsp://demo.invalid/live")

    config = load_config(config_file)

    assert config.source_id == "lobby-camera-01"
    assert config.source_url == "rtsp://demo.invalid/live"
    assert config.event.confirm_window_seconds == 1.8


def test_load_config_rejects_embedded_source_address(tmp_path):
    config_file = tmp_path / "config.json"
    _write_config(
        config_file,
        {"id": "lobby-camera-01", "url": "rtsp://demo.invalid/live"},
    )

    with pytest.raises(ConfigError, match="rtsp_url_env"):
        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_write_local_camera_source_round_trips_through_load_config(tmp_path):
    from v1.config import write_local_camera_source

    config_file = tmp_path / "config.local.json"
    _write_config(config_file, {"id": "seed", "rtsp_url_env": "SILVER_POSE_RTSP_URL"})

    write_local_camera_source(
        config_file,
        source_id="hik",
        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_mode == "stream"
    assert config.source_id == "hik"


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(
        config_file,
        {"id": "lobby-camera-01", "rtsp_url_env": "SILVER_POSE_RTSP_URL"},
    )
    monkeypatch.setenv("SILVER_POSE_RTSP_URL", "rtsp://demo.invalid/live")

    assert load_config(config_file).source_mode == "stream"


def test_source_mode_replay_is_parsed(tmp_path, monkeypatch):
    config_file = tmp_path / "config.json"
    _write_config(
        config_file,
        {"id": "lobby-camera-01", "rtsp_url_env": "SILVER_POSE_RTSP_URL", "mode": "replay"},
    )
    monkeypatch.setenv("SILVER_POSE_RTSP_URL", "rtsp://demo.invalid/live")

    assert load_config(config_file).source_mode == "replay"


def test_invalid_source_mode_is_rejected(tmp_path, monkeypatch):
    config_file = tmp_path / "config.json"
    _write_config(
        config_file,
        {"id": "lobby-camera-01", "rtsp_url_env": "SILVER_POSE_RTSP_URL", "mode": "loop"},
    )
    monkeypatch.setenv("SILVER_POSE_RTSP_URL", "rtsp://demo.invalid/live")

    with pytest.raises(ConfigError, match="mode"):
        load_config(config_file)


def test_runtime_config_version_is_stable_and_excludes_rtsp_address(tmp_path, monkeypatch):
    config_file = tmp_path / "config.json"
    _write_config(
        config_file,
        {"id": "lobby-camera-01", "rtsp_url_env": "SILVER_POSE_RTSP_URL"},
    )
    monkeypatch.setenv("SILVER_POSE_RTSP_URL", "rtsp://operator:secret@camera-a/live")
    first = load_config(config_file)
    monkeypatch.setenv("SILVER_POSE_RTSP_URL", "rtsp://operator:other-secret@camera-b/live")
    second = load_config(config_file)

    assert first.runtime_config_version == second.runtime_config_version
    assert first.runtime_config_version.startswith("cfg-")