93 lines
3.0 KiB
Python
93 lines
3.0 KiB
Python
"""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,
|
|
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_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_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_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"
|
|
|