"""Last-good emergency video cache: honest label, atomic replacement and fail-closed source.""" from __future__ import annotations import sys from datetime import UTC, datetime from io import BytesIO from pathlib import Path import pytest sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "live")) import emergency_video_cache as C # noqa: E402 def test_priority_discovery_uses_only_the_authenticated_loopback_relay(): assert C.DEFAULT_EVENTS_URL == "http://127.0.0.1:8766/events" def test_source_is_https_and_never_guessed(): assert C.source_from_env({"EMERGENCY_VIDEO_SOURCE": "https://video.example/channel.m3u8"}).endswith( "channel.m3u8" ) with pytest.raises(C.EmergencyStreamError, match="set EMERGENCY_VIDEO_SOURCE"): C.source_from_env({}) with pytest.raises(C.EmergencyStreamError, match="https"): C.source_from_env({"EMERGENCY_VIDEO_SOURCE": "http://video.example/channel.m3u8"}) def test_priority_slot_follows_the_init_header_and_selects_newest_https_video(): doc = { "meta": {"instanceId": "race-7", "cameraSlots": ["boat-2", "boat-9"]}, "media": {"boat-2": [ {"tSec": 10, "file": "https://media.example/old.mp4", "video": True}, {"tSec": 12, "file": "http://unsafe.example/no.mp4", "video": True}, {"tSec": 20, "file": "https://media.example/new.mp4", "video": True}, ]}, } assert C.priority_source_from_init(doc, "A") == "https://media.example/new.mp4" def test_idle_or_unassigned_priority_slot_waits_without_guessing(): with pytest.raises(C.EmergencyStreamError, match="waiting for a race"): C.priority_source_from_init({"meta": {"idle": True}}, "A") with pytest.raises(C.EmergencyStreamError, match="not assigned"): C.priority_source_from_init({"meta": {"cameraSlots": []}}, "A") def test_priority_source_reads_only_the_sse_init_and_never_logs_the_signed_url(): payload = ( b"event: init\n" b"data: {\"meta\":{\"cameraSlots\":[\"boat-a\"]},\"media\":{\"boat-a\":[" b"{\"tSec\":1,\"file\":\"https://media.example/clip.mp4?sig=SECRET\",\"video\":true}]}}\n\n" ) class Response(BytesIO): def __enter__(self): return self def __exit__(self, *_args): self.close() opener = lambda _url, timeout: Response(payload) # noqa: E731 assert C.priority_source_from_events("https://broadcast.example/events", "A", opener=opener).endswith( "sig=SECRET" ) def test_capture_burns_an_earlier_picture_label_and_uses_silent_audio(tmp_path: Path): cmd = C.capture_cmd( source="https://video.example/channel.m3u8", output=tmp_path / "candidate.mp4", label="Priority camera A", captured=datetime(2026, 8, 13, 18, 30, tzinfo=UTC), font=tmp_path / "font.ttf", duration_s=10, ) joined = " ".join(cmd) assert "EARLIER PICTURES | Priority camera A" in joined assert "2026-08-13 18\\:30 UTC" in joined assert "anullsrc=r=48000:cl=stereo" in joined assert f"scale={C.WIDTH}:{C.HEIGHT}" in joined and f"fps={C.FPS}" in joined assert "-map 0:v:0 -map 1:a:0" in joined def test_a_bad_label_cannot_escape_the_filter(tmp_path: Path): cmd = C.capture_cmd( source="https://video.example/channel.m3u8", output=tmp_path / "x.mp4", label=" camera:'bad' name ", captured=datetime.now(UTC), font=tmp_path / "font.ttf", ) joined = " ".join(cmd) assert "camera bad name" in joined assert "camera:'bad'" not in joined def test_capture_replaces_only_after_validation(monkeypatch, tmp_path: Path): output = tmp_path / "last_good.mp4" output.write_bytes(b"old-known-good") def fake_run(cmd, **_kw): Path(cmd[-1]).write_bytes(b"new-complete") monkeypatch.setattr(C.subprocess, "run", fake_run) monkeypatch.setattr(C, "default_font", lambda: tmp_path / "font.ttf") monkeypatch.setattr(C, "verify_media", lambda p, **_kw: {"bytes": p.stat().st_size, "video": "ok"}) result = C.capture_once( output=output, label="A", source="https://video.example/a.m3u8", now=datetime(2026, 8, 13, 18, 30, tzinfo=UTC), ) assert output.read_bytes() == b"new-complete" assert result["path"] == str(output.resolve()) def test_failed_validation_preserves_the_previous_file(monkeypatch, tmp_path: Path): output = tmp_path / "last_good.mp4" output.write_bytes(b"old-known-good") def fake_run(cmd, **_kw): Path(cmd[-1]).write_bytes(b"partial") monkeypatch.setattr(C.subprocess, "run", fake_run) monkeypatch.setattr(C, "default_font", lambda: tmp_path / "font.ttf") monkeypatch.setattr( C, "verify_media", lambda _p, **_kw: (_ for _ in ()).throw(C.EmergencyStreamError("broken decoder")) ) with pytest.raises(C.EmergencyStreamError, match="broken decoder"): C.capture_once(output=output, label="A", source="https://video.example/a.m3u8") assert output.read_bytes() == b"old-known-good" assert not (tmp_path / ".last_good.candidate.mp4").exists() def test_waiting_health_is_redacted_and_preserves_last_good(tmp_path: Path): output = tmp_path / "last_good.mp4" output.write_bytes(b"previous") health = tmp_path / "health.json" code = C.run( output=output, label="Priority camera A", once=True, health_path=health, source_resolver=lambda: (_ for _ in ()).throw(C.EmergencyStreamError("no active race")), ) doc = __import__("json").loads(health.read_text(encoding="utf-8")) assert code == 2 and doc["status"] == "waiting" and doc["lastGoodPresent"] is True assert output.read_bytes() == b"previous" def test_windows_startup_launcher_has_no_user_or_tool_version_hardcoding(): deploy = Path(C.__file__).resolve().parents[1] / "deploy" runner = (deploy / "run_priority_video_cache.ps1").read_text(encoding="utf-8") command = (deploy / "run_priority_video_cache.cmd").read_text(encoding="utf-8") hidden = (deploy / "run_priority_video_cache_hidden.vbs").read_text(encoding="utf-8") assert "$env:ProgramData" in runner and "$env:LOCALAPPDATA" in runner assert "Get-Command python.exe" in runner and "Get-Command ffmpeg.exe" in runner assert "if ($ffmpeg.Source)" in runner and "if ($ffprobe.Source)" in runner assert "*>> $logPath" in runner assert "C:\\Users\\" not in runner and "ffmpeg-8." not in runner assert "%ProgramData%" in command and "run_priority_video_cache.ps1" in command assert "C:\\Users\\" not in command and "ffmpeg-8." not in command assert "run_priority_video_cache.cmd" in hidden