operating-room-monitor-server/backend/tests/test_hls_preview_api.py

"""HLS 预览 API。"""

from __future__ import annotations

from unittest.mock import patch

import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient

from app.api import router as api_router
from app.config import Settings
from app.services.hls_preview import (
    HlsPreviewManager,
    HlsPreviewSession,
    _mediamtx_config_lines,
    append_upstream_query,
    attached_upstream_connect_candidates,
    docker_publish_bind_host,
    normalize_upstream_base,
    resolve_ephemeral_upstream_host,
)


@pytest.fixture
def hls_client(tmp_path) -> TestClient:
    site = tmp_path / "site.json"
    site.write_text(
        '{"video_rtsp_urls":{"or-cam-01":"rtsp://cam/1"},'
        '"voice_or_room_bindings":[]}',
        encoding="utf-8",
    )
    app = FastAPI()
    app.include_router(api_router)
    with patch("app.api.settings", Settings(or_site_config_json_file=str(site))):
        yield TestClient(app)
    HlsPreviewManager.stop()


def test_normalize_upstream_base() -> None:
    assert normalize_upstream_base("http://mediamtx-hls:8888") == "http://mediamtx-hls:8888"
    assert normalize_upstream_base("http://mediamtx-hls:8888/") == "http://mediamtx-hls:8888"


def test_docker_publish_bind_host_maps_docker_internal() -> None:
    assert docker_publish_bind_host("host.docker.internal") == "127.0.0.1"
    assert docker_publish_bind_host("127.0.0.1") == "127.0.0.1"


def test_resolve_upstream_setting() -> None:
    assert HlsPreviewManager.resolve_upstream_setting("http://127.0.0.1:18888") == (
        "http://127.0.0.1:18888"
    )
    assert HlsPreviewManager.resolve_upstream_setting("ephemeral") == ""


def test_hls_preview_ensure_pull(hls_client: TestClient) -> None:
    fake_sess = HlsPreviewSession(
        upstream_base="http://127.0.0.1:18888",
        path_by_camera={"or-cam-01": "or-cam-01"},
        mode="ephemeral",
    )

    def _fake_start(**_kwargs: object) -> HlsPreviewSession:
        HlsPreviewManager._active = fake_sess
        return fake_sess

    with patch.object(HlsPreviewManager, "start", side_effect=_fake_start):
        r = hls_client.post(
            "/internal/demo/hls-preview/ensure",
            json={"camera_ids": ["or-cam-01"]},
        )
    assert r.status_code == 200, r.text
    body = r.json()
    assert body["cameras"][0]["camera_id"] == "or-cam-01"
    assert "index.m3u8" in body["cameras"][0]["playlist_url"]


def test_append_upstream_query() -> None:
    assert (
        append_upstream_query("http://mediamtx-hls:8888/or-cam-01/video1_stream.m3u8", "session=abc")
        == "http://mediamtx-hls:8888/or-cam-01/video1_stream.m3u8?session=abc"
    )
    assert append_upstream_query("http://host/a.m3u8?foo=1", "session=abc") == (
        "http://host/a.m3u8?foo=1&session=abc"
    )


def test_hls_proxy_forwards_session_query(hls_client: TestClient) -> None:
    HlsPreviewManager._active = HlsPreviewSession(
        upstream_base="http://127.0.0.1:18888",
        path_by_camera={"or-cam-01": "or-cam-01"},
        mode="attached",
    )
    captured: list[str] = []

    def _fake_fetch(url: str, *, query: str = "") -> tuple[bytes, str]:
        captured.append(append_upstream_query(url, query))
        return (b"#EXTM3U\n", "application/vnd.apple.mpegurl")

    with patch("app.api.fetch_hls_upstream", side_effect=_fake_fetch):
        r = hls_client.get(
            "/internal/demo/hls-preview/or-cam-01/video1_stream.m3u8?session=abc-123"
        )
    assert r.status_code == 200
    assert captured == [
        "http://127.0.0.1:18888/or-cam-01/video1_stream.m3u8?session=abc-123"
    ]


def test_hls_proxy_rewrites_playlist(hls_client: TestClient) -> None:
    HlsPreviewManager._active = HlsPreviewSession(
        upstream_base="http://127.0.0.1:18888",
        path_by_camera={"or-cam-01": "or-cam-01"},
        mode="attached",
    )
    playlist = (
        "#EXTM3U\n#EXT-X-TARGETDURATION:1\n"
        "http://127.0.0.1:18888/or-cam-01/segment.ts\n"
    ).encode()

    with patch("app.api.fetch_hls_upstream", return_value=(playlist, "application/vnd.apple.mpegurl")):
        r = hls_client.get("/internal/demo/hls-preview/or-cam-01/index.m3u8")
    assert r.status_code == 200
    text = r.text
    assert "/internal/demo/hls-preview/or-cam-01/segment.ts" in text
    assert "127.0.0.1:18888" not in text


def test_mediamtx_config_includes_anonymous_read() -> None:
    text = "\n".join(
        _mediamtx_config_lines({"or-cam-01": "rtsp://127.0.0.1:18554/demo1"})
    )
    assert "authInternalUsers:" in text
    assert "action: read" in text
    assert "action: playback" in text
    assert "or-cam-01:" in text


def test_attached_upstream_connect_candidates_local() -> None:
    with patch("app.services.hls_preview.Path") as path_cls:
        path_cls.return_value.is_file.return_value = False
        got = attached_upstream_connect_candidates("http://127.0.0.1:18888")
    assert got == [("127.0.0.1", 18888)]


def test_attached_upstream_connect_candidates_docker_bridge() -> None:
    with patch("app.services.hls_preview.Path") as path_cls:
        path_cls.return_value.is_file.return_value = True
        got = attached_upstream_connect_candidates("http://127.0.0.1:18888")
    assert got == [("127.0.0.1", 18888), ("mediamtx-hls", 8888)]


def test_resolve_ephemeral_upstream_host_local() -> None:
    with patch("app.services.hls_preview.Path") as path_cls:
        path_cls.return_value.is_file.return_value = False
        assert resolve_ephemeral_upstream_host("127.0.0.1") == "127.0.0.1"