103 lines
3.2 KiB
Python
103 lines
3.2 KiB
Python
"""Tests for cht.stream.manager — StreamManager."""
|
|
|
|
import json
|
|
import time
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
import pytest
|
|
|
|
from cht.stream.manager import StreamManager
|
|
|
|
|
|
@pytest.fixture
|
|
def manager(tmp_path):
|
|
with patch("cht.stream.manager.SESSIONS_DIR", tmp_path):
|
|
mgr = StreamManager(session_id="test_session")
|
|
yield mgr
|
|
mgr.stop_all()
|
|
|
|
|
|
class TestInit:
|
|
def test_session_id_custom(self, manager):
|
|
assert manager.session_id == "test_session"
|
|
|
|
def test_recording_path(self, manager):
|
|
assert manager.recording_path.name == "recording.mkv"
|
|
|
|
def test_dirs_not_created_on_init(self, manager):
|
|
assert not manager.stream_dir.exists()
|
|
|
|
|
|
class TestSetupDirs:
|
|
def test_creates_all_subdirs(self, manager):
|
|
manager.setup_dirs()
|
|
assert manager.stream_dir.is_dir()
|
|
assert manager.frames_dir.is_dir()
|
|
assert manager.transcript_dir.is_dir()
|
|
assert manager.agent_dir.is_dir()
|
|
|
|
|
|
class TestStartRecorder:
|
|
@patch("cht.stream.manager.ff.run_async")
|
|
@patch("cht.stream.manager.ff.receive_and_record")
|
|
def test_starts_ffmpeg(self, mock_record, mock_async, manager):
|
|
manager.setup_dirs()
|
|
mock_record.return_value = MagicMock()
|
|
manager.start_recorder()
|
|
mock_record.assert_called_once_with(manager.stream_url, manager.recording_path)
|
|
assert "recorder" in manager._procs
|
|
|
|
|
|
class TestStopAll:
|
|
@patch("cht.stream.manager.ff.stop_proc")
|
|
def test_stops_all_procs(self, mock_stop, manager):
|
|
proc = MagicMock()
|
|
manager._procs = {"recorder": proc}
|
|
manager.stop_all()
|
|
mock_stop.assert_called_with(proc)
|
|
assert len(manager._procs) == 0
|
|
|
|
def test_sets_stop_flag(self, manager):
|
|
manager.stop_all()
|
|
assert "stop" in manager._stop_flags
|
|
|
|
|
|
class TestDetectScenes:
|
|
@patch("cht.stream.manager.ff.extract_scene_frames")
|
|
def test_returns_new_frames(self, mock_extract, manager):
|
|
manager.setup_dirs()
|
|
rec = manager.recording_path
|
|
rec.touch()
|
|
|
|
def create_frame(*args, **kwargs):
|
|
(manager.frames_dir / "F0001.jpg").touch()
|
|
return ("", "[Parsed_showinfo_1 @ 0x1] n:0 pts:100 pts_time:10.5 stuff\n")
|
|
|
|
mock_extract.side_effect = create_frame
|
|
|
|
frames = manager._detect_scenes(start_time=0, end_time=15, start_number=1)
|
|
assert len(frames) == 1
|
|
assert frames[0]["id"] == "F0001"
|
|
assert frames[0]["timestamp"] == 10.5
|
|
|
|
@patch("cht.stream.manager.ff.extract_scene_frames")
|
|
def test_passes_duration(self, mock_extract, manager):
|
|
manager.setup_dirs()
|
|
manager.recording_path.touch()
|
|
mock_extract.return_value = ("", "")
|
|
|
|
manager._detect_scenes(start_time=10, end_time=25, start_number=1)
|
|
|
|
call_kwargs = mock_extract.call_args
|
|
assert call_kwargs.kwargs["start_time"] == 10
|
|
assert call_kwargs.kwargs["duration"] == 15
|
|
|
|
@patch("cht.stream.manager.ff.extract_scene_frames")
|
|
def test_handles_failure(self, mock_extract, manager):
|
|
manager.setup_dirs()
|
|
manager.recording_path.touch()
|
|
mock_extract.side_effect = RuntimeError("boom")
|
|
|
|
frames = manager._detect_scenes(start_time=0, end_time=10, start_number=1)
|
|
assert frames == []
|