95 lines
3.1 KiB
Python
95 lines
3.1 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_000.mp4"
|
|
|
|
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 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 TestFrameIndex:
|
|
def test_next_frame_number_empty(self, manager):
|
|
manager.setup_dirs()
|
|
assert manager._next_frame_number() == 1
|
|
|
|
def test_next_frame_number_with_existing(self, manager):
|
|
manager.setup_dirs()
|
|
index = [{"id": "F0001"}, {"id": "F0002"}]
|
|
(manager.frames_dir / "index.json").write_text(json.dumps(index))
|
|
assert manager._next_frame_number() == 3
|
|
|
|
def test_append_frame_index(self, manager):
|
|
manager.setup_dirs()
|
|
entry = {"id": "F0001", "timestamp": 5.0, "path": "/tmp/F0001.jpg", "sent_to_agent": False}
|
|
manager._append_frame_index(entry)
|
|
index = json.loads((manager.frames_dir / "index.json").read_text())
|
|
assert len(index) == 1
|
|
assert index[0]["id"] == "F0001"
|
|
|
|
def test_append_frame_index_accumulates(self, manager):
|
|
manager.setup_dirs()
|
|
for i in range(3):
|
|
entry = {"id": f"F{i+1:04d}", "timestamp": float(i), "path": f"/tmp/F{i+1:04d}.jpg", "sent_to_agent": False}
|
|
manager._append_frame_index(entry)
|
|
index = json.loads((manager.frames_dir / "index.json").read_text())
|
|
assert len(index) == 3
|
|
|
|
|
|
class TestSceneDetector:
|
|
def test_start_scene_detector_stores_callback(self, manager):
|
|
cb = MagicMock()
|
|
manager.start_scene_detector(on_new_frames=cb)
|
|
assert manager._on_new_frames is cb
|
|
|
|
def test_update_scene_threshold(self, manager):
|
|
manager.setup_dirs()
|
|
# Mock restart_recorder to avoid launching ffmpeg
|
|
manager.restart_recorder = MagicMock()
|
|
manager.update_scene_threshold(0.25)
|
|
assert manager.scene_threshold == 0.25
|
|
manager.restart_recorder.assert_called_once()
|