operating-room-monitor-server/tests/test_voice_terminal_binding.py

"""术间绑定与站点配置（OR_SITE_CONFIG_JSON）解析。"""

from __future__ import annotations

import json
from pathlib import Path

import pytest

from app.config import Settings
from app.or_site_config import (
    load_or_site_config_from_path,
    merge_video_rtsp_urls_into_file,
    parse_or_site_config_object,
)
from app.services.voice_terminal_binding import VoiceTerminalBindingIndex


def test_parse_and_resolve() -> None:
    idx = VoiceTerminalBindingIndex.from_binding_list(
        [
            {
                "or_room_id": "OR-05",
                "camera_ids": ["or-cam-01", "or-cam-02"],
                "voice_terminal_id": "T-05",
            }
        ]
    )
    assert idx is not None
    assert idx.resolve_terminal(["or-cam-02", "or-cam-01"]) == "T-05"
    assert idx.resolve_terminal(["or-cam-01"]) == "T-05"


def test_resolve_subset_uses_smallest_superset_when_ambiguous() -> None:
    idx = VoiceTerminalBindingIndex.from_binding_list(
        [
            {
                "or_room_id": "A",
                "camera_ids": ["c1", "c2"],
                "voice_terminal_id": "T-AB",
            },
            {
                "or_room_id": "B",
                "camera_ids": ["c1", "c2", "c3"],
                "voice_terminal_id": "T-ABC",
            },
        ]
    )
    assert idx is not None
    assert idx.resolve_terminal(["c1"]) == "T-AB"


def test_duplicate_terminal_rejected() -> None:
    assert (
        VoiceTerminalBindingIndex.from_binding_list(
            [
                {"or_room_id": "A", "camera_ids": ["c1"], "voice_terminal_id": "T"},
                {"or_room_id": "B", "camera_ids": ["c2"], "voice_terminal_id": "T"},
            ]
        )
        is None
    )


def test_duplicate_camera_set_rejected() -> None:
    assert (
        VoiceTerminalBindingIndex.from_binding_list(
            [
                {"or_room_id": "A", "camera_ids": ["c1"], "voice_terminal_id": "T1"},
                {"or_room_id": "B", "camera_ids": ["c1"], "voice_terminal_id": "T2"},
            ]
        )
        is None
    )


def test_from_binding_list_empty() -> None:
    idx = VoiceTerminalBindingIndex.from_binding_list([])
    assert idx is not None
    assert idx.resolve_terminal(["x"]) is None


def test_parse_site_config_rejects_unknown_keys() -> None:
    with pytest.raises(ValueError, match="unknown top-level"):
        parse_or_site_config_object(
            {
                "video_rtsp_urls": {},
                "voice_or_room_bindings": [],
                "extra": 1,
            }
        )


def test_parse_site_config_requires_both_sections() -> None:
    with pytest.raises(ValueError, match="must include"):
        parse_or_site_config_object({"video_rtsp_urls": {}})


def test_load_from_path_roundtrip(tmp_path: Path) -> None:
    p = tmp_path / "site.json"
    p.write_text(
        '{"video_rtsp_urls":{"c1":"rtsp://h"},"voice_or_room_bindings":['
        '{"or_room_id":"R","camera_ids":["a"],"voice_terminal_id":"T"}]}',
        encoding="utf-8",
    )
    cfg = load_or_site_config_from_path(p)
    assert cfg.video_rtsp_urls == {"c1": "rtsp://h"}
    assert cfg.voice_bindings.resolve_terminal(["a"]) == "T"


def test_settings_video_map(tmp_path: Path) -> None:
    p = tmp_path / "site.json"
    p.write_text(
        '{"video_rtsp_urls":{"c1":"rtsp://x"},"voice_or_room_bindings":[]}',
        encoding="utf-8",
    )
    s = Settings(or_site_config_json_file=str(p))
    assert s.video_rtsp_url_map() == {"c1": "rtsp://x"}


def test_merge_preserves_bindings(tmp_path: Path) -> None:
    p = tmp_path / "site.json"
    p.write_text(
        '{"video_rtsp_urls":{"old":"rtsp://old"},'
        '"voice_or_room_bindings":[{"or_room_id":"R","camera_ids":["x"],'
        '"voice_terminal_id":"T"}]}',
        encoding="utf-8",
    )
    merge_video_rtsp_urls_into_file(
        p, {"x": "rtsp://127.0.0.1:1/p"}, replace_host="127.0.0.1"
    )
    cfg = load_or_site_config_from_path(p)
    assert cfg.video_rtsp_urls == {"x": "rtsp://127.0.0.1:1/p"}
    assert cfg.voice_bindings.resolve_terminal(["x"]) == "T"