mariano/mitus
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

working the player

Browse Source
This commit is contained in:
buenosairesam
2026-04-01 19:23:17 -03:00
parent 68802db15c
commit 0f7e4424bc
13 changed files with 1013 additions and 571 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
tests/test_ffmpeg.py
View File

@@ -41,11 +41,11 @@ class TestReceiveAndRecord:
)
assert "-hide_banner" in node.compile()
def test_mpegts_format(self, tmp_path):
def test_matroska_format(self, tmp_path):
node = ff.receive_and_record(
"tcp://0.0.0.0:4444?listen", tmp_path / "rec.ts"
"tcp://0.0.0.0:4444?listen", tmp_path / "rec.mkv"
)
assert "mpegts" in node.compile()
assert "matroska" in node.compile()
class TestExtractSceneFrames:

143
tests/test_manager.py
View File

@@ -1,4 +1,4 @@
"""Tests for cht.stream.manager — StreamManager orchestration."""
"""Tests for cht.stream.manager — StreamManager."""
import json
import time
@@ -18,14 +18,12 @@ def manager(tmp_path):
class TestInit:
def test_session_id_default(self, tmp_path):
with patch("cht.stream.manager.SESSIONS_DIR", tmp_path):
mgr = StreamManager()
assert mgr.session_id
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()
@@ -38,52 +36,25 @@ class TestSetupDirs:
assert manager.transcript_dir.is_dir()
assert manager.agent_dir.is_dir()
def test_idempotent(self, manager):
manager.setup_dirs()
manager.setup_dirs()
assert manager.stream_dir.is_dir()
class TestStreamUrl:
def test_default_url(self, manager):
assert "0.0.0.0" in manager.stream_url
assert "4444" in manager.stream_url
assert "listen" in manager.stream_url
class TestRecordingPath:
def test_is_in_stream_dir(self, manager):
assert manager.recording_path.parent == manager.stream_dir
assert manager.recording_path.name == "recording.ts"
class TestStartRecorder:
@patch("cht.stream.manager.ff.run_async")
@patch("cht.stream.manager.ff.receive_and_record")
def test_calls_ffmpeg_module(self, mock_record, mock_async, manager):
def test_starts_ffmpeg(self, mock_record, mock_async, manager):
manager.setup_dirs()
mock_node = MagicMock()
mock_record.return_value = mock_node
mock_record.return_value = MagicMock()
manager.start_recorder()
mock_record.assert_called_once_with(
manager.stream_url, manager.recording_path,
)
mock_async.assert_called_once_with(mock_node, pipe_stderr=True)
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):
proc1, proc2 = MagicMock(), MagicMock()
manager._procs = {"a": proc1, "b": proc2}
proc = MagicMock()
manager._procs = {"recorder": proc}
manager.stop_all()
mock_stop.assert_any_call(proc1)
mock_stop.assert_any_call(proc2)
mock_stop.assert_called_with(proc)
assert len(manager._procs) == 0
def test_sets_stop_flag(self, manager):
@@ -91,89 +62,41 @@ class TestStopAll:
assert "stop" in manager._stop_flags
class TestExtractNewFrames:
class TestDetectScenes:
@patch("cht.stream.manager.ff.extract_scene_frames")
def test_calls_ffmpeg_with_start_time(self, mock_extract, manager):
def test_returns_new_frames(self, mock_extract, manager):
manager.setup_dirs()
rec = manager.recording_path
rec.touch()
mock_extract.return_value = ("", "")
manager._extract_new_frames(rec, start_time=10.0, start_number=5)
mock_extract.assert_called_once_with(
rec,
manager.frames_dir,
scene_threshold=0.3,
max_interval=30,
start_number=5,
start_time=10.0,
)
@patch("cht.stream.manager.ff.extract_scene_frames")
def test_indexes_new_frames(self, mock_extract, manager):
manager.setup_dirs()
rec = manager.recording_path
rec.touch()
# Simulate ffmpeg creating a frame file during extraction
def create_frame_and_return(*args, **kwargs):
def create_frame(*args, **kwargs):
(manager.frames_dir / "F0001.jpg").touch()
return ("", "[Parsed_showinfo_1 @ 0x1] n:0 pts:1000 pts_time:10.5 stuff\n")
return ("", "[Parsed_showinfo_1 @ 0x1] n:0 pts:100 pts_time:10.5 stuff\n")
mock_extract.side_effect = create_frame_and_return
mock_extract.side_effect = create_frame
count, max_ts = manager._extract_new_frames(rec, start_number=1)
assert count == 1
assert max_ts == 10.5
index_path = manager.frames_dir / "index.json"
with open(index_path) as f:
index = json.load(f)
assert len(index) == 1
assert index[0]["id"] == "F0001"
assert index[0]["timestamp"] == 10.5
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_handles_ffmpeg_failure(self, mock_extract, manager):
manager.setup_dirs()
rec = manager.recording_path
rec.touch()
mock_extract.side_effect = RuntimeError("ffmpeg died")
count, max_ts = manager._extract_new_frames(rec)
assert count == 0
@patch("cht.stream.manager.ff.extract_scene_frames")
def test_skips_preexisting_frames(self, mock_extract, manager):
manager.setup_dirs()
rec = manager.recording_path
rec.touch()
# Pre-existing frame
(manager.frames_dir / "F0001.jpg").touch()
# ffmpeg "creates" no new files, just returns showinfo for existing
stderr = "[Parsed_showinfo_1 @ 0x1] n:0 pts:100 pts_time:5.0 stuff\n"
mock_extract.return_value = ("", stderr)
count, _ = manager._extract_new_frames(rec, start_number=1)
# F0001 already existed before extraction, should not be counted
assert count == 0
class TestSceneDetector:
@patch("cht.stream.manager.ff.extract_scene_frames")
def test_detects_growing_file(self, mock_extract, manager):
def test_passes_duration(self, mock_extract, manager):
manager.setup_dirs()
manager.recording_path.touch()
mock_extract.return_value = ("", "")
# Create recording with some data
rec = manager.recording_path
rec.write_bytes(b"\x00" * 200_000)
manager._detect_scenes(start_time=10, end_time=25, start_number=1)
manager.start_scene_detector()
time.sleep(12) # wait for one cycle
manager.stop_all()
call_kwargs = mock_extract.call_args
assert call_kwargs.kwargs["start_time"] == 10
assert call_kwargs.kwargs["duration"] == 15
mock_extract.assert_called()
@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 == []

47
tests/test_timeline.py Normal file
View File

@@ -0,0 +1,47 @@
"""Tests for cht.ui.timeline — Timeline state machine."""
import pytest
from unittest.mock import MagicMock
# Skip GTK import for headless testing
import sys
from unittest.mock import MagicMock as _MM
# Mock gi modules for headless testing
gi_mock = _MM()
gi_mock.require_version = _MM()
gtk_mock = _MM()
gobject_mock = _MM()
# GObject.Object needs __gsignals__ support
class FakeGObject:
def __init__(self):
self._signals = {}
def emit(self, signal, *args):
for cb in self._signals.get(signal, []):
cb(self, *args)
def connect(self, signal, cb):
self._signals.setdefault(signal, []).append(cb)
# We test the logic, not the GTK widgets
# Import after mocking would be complex, so test the state logic directly
try:
from cht.ui.timeline import State
HAS_GI = True
except ImportError:
HAS_GI = False
@pytest.mark.skipif(not HAS_GI, reason="GTK/gi not available")
class TestTimelineState:
def test_initial_state(self):
assert State.WAITING.name == "WAITING"
assert State.LIVE.name == "LIVE"
assert State.PLAYING.name == "PLAYING"
assert State.PAUSED.name == "PAUSED"
def test_state_values_are_unique(self):
values = [s.value for s in State]
assert len(values) == len(set(values))

38
tests/test_tracker.py Normal file
View File

@@ -0,0 +1,38 @@
"""Tests for cht.stream.tracker — RecordingTracker."""
import time
from pathlib import Path
from unittest.mock import patch, MagicMock
import pytest
from cht.stream.tracker import RecordingTracker
class TestRecordingTracker:
def test_initial_duration_is_zero(self, tmp_path):
tracker = RecordingTracker(tmp_path / "rec.ts")
assert tracker.duration == 0.0
def test_callback_called_on_update(self, tmp_path):
rec = tmp_path / "rec.ts"
rec.write_bytes(b"\x00" * 100_000)
cb = MagicMock()
tracker = RecordingTracker(rec, on_duration_update=cb)
with patch.object(tracker, "_probe_duration", return_value=10.0):
tracker.start()
time.sleep(3)
tracker.stop()
cb.assert_called()
assert cb.call_args[0][0] > 0
def test_no_callback_if_file_missing(self, tmp_path):
cb = MagicMock()
tracker = RecordingTracker(tmp_path / "nonexistent.ts", on_duration_update=cb)
tracker.start()
time.sleep(3)
tracker.stop()
cb.assert_not_called()