327 lines
11 KiB
Python
327 lines
11 KiB
Python
"""Tests for HLS preview manager functionality."""
|
|
|
|
import pytest
|
|
import asyncio
|
|
from pathlib import Path
|
|
from unittest.mock import Mock, AsyncMock, MagicMock, patch, call, PropertyMock
|
|
from backend.app.config import Settings
|
|
from backend.app.utils.hls import HLSPreviewManager
|
|
|
|
|
|
@pytest.fixture
|
|
def settings():
|
|
"""Create test settings."""
|
|
return Settings(
|
|
ffmpeg_path="/usr/bin/ffmpeg",
|
|
recording_dir="/recordings",
|
|
deltacast_port_count=4,
|
|
srt_enabled=True,
|
|
srt_latency=5000,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def hls_manager(settings, tmp_path):
|
|
"""Create HLS preview manager with temp directory."""
|
|
return HLSPreviewManager(settings, hls_dir=str(tmp_path / "hls"))
|
|
|
|
|
|
class TestHLSManagerInitialization:
|
|
"""Test HLS manager initialization."""
|
|
|
|
def test_hls_manager_initial_state(self, hls_manager):
|
|
"""No previews running initially."""
|
|
assert hls_manager._processes == {}
|
|
assert len(hls_manager._processes) == 0
|
|
|
|
|
|
class TestHLSCommandBuilder:
|
|
"""Test HLS FFmpeg command building."""
|
|
|
|
def test_build_hls_command_includes_deltacast_input(self, hls_manager):
|
|
"""Command includes deltacast:// input."""
|
|
command = hls_manager._build_hls_command(0)
|
|
assert "deltacast://0" in command
|
|
|
|
def test_build_hls_command_includes_hls_output(self, hls_manager):
|
|
"""Command includes -f hls output."""
|
|
command = hls_manager._build_hls_command(0)
|
|
assert "-f" in command
|
|
assert "hls" in command
|
|
|
|
def test_build_hls_command_has_correct_video_params(self, hls_manager):
|
|
"""Command includes libx264, 5M bitrate, 1280x720."""
|
|
command = hls_manager._build_hls_command(0)
|
|
assert "libx264" in command
|
|
assert "-b:v" in command
|
|
assert "5M" in command
|
|
assert "-s" in command
|
|
assert "1280x720" in command
|
|
|
|
def test_build_hls_command_has_hls_time_5(self, hls_manager):
|
|
"""Command includes -hls_time 5."""
|
|
command = hls_manager._build_hls_command(0)
|
|
assert "-hls_time" in command
|
|
assert "5" in command
|
|
|
|
def test_build_hls_command_has_zerolatency_tune(self, hls_manager):
|
|
"""Command includes -tune zerolatency."""
|
|
command = hls_manager._build_hls_command(0)
|
|
assert "-tune" in command
|
|
assert "zerolatency" in command
|
|
|
|
def test_build_hls_command_has_hls_list_size_3(self, hls_manager):
|
|
"""Command includes -hls_list_size 3."""
|
|
command = hls_manager._build_hls_command(0)
|
|
assert "-hls_list_size" in command
|
|
assert "3" in command
|
|
|
|
def test_build_hls_command_has_delete_segments_flag(self, hls_manager):
|
|
"""Command includes -hls_flags delete_segments."""
|
|
command = hls_manager._build_hls_command(0)
|
|
assert "-hls_flags" in command
|
|
assert "delete_segments" in command
|
|
|
|
|
|
class TestPlaylistPath:
|
|
"""Test playlist path generation."""
|
|
|
|
def test_get_playlist_path(self, hls_manager):
|
|
"""Returns correct .m3u8 file path."""
|
|
path = hls_manager.get_playlist_path(0)
|
|
assert isinstance(path, Path)
|
|
assert path.name == "port_0.m3u8"
|
|
assert "hls" in str(path)
|
|
|
|
def test_get_playlist_path_different_ports(self, hls_manager):
|
|
"""Returns different paths for different ports."""
|
|
path0 = hls_manager.get_playlist_path(0)
|
|
path1 = hls_manager.get_playlist_path(1)
|
|
path4 = hls_manager.get_playlist_path(4)
|
|
|
|
assert path0.name == "port_0.m3u8"
|
|
assert path1.name == "port_1.m3u8"
|
|
assert path4.name == "port_4.m3u8"
|
|
|
|
|
|
class TestPreviewState:
|
|
"""Test preview state tracking."""
|
|
|
|
def test_is_previewing_false_initially(self, hls_manager):
|
|
"""is_previewing returns False when not started."""
|
|
assert not hls_manager.is_previewing(0)
|
|
assert not hls_manager.is_previewing(1)
|
|
|
|
def test_is_previewing_false_when_not_in_dict(self, hls_manager):
|
|
"""is_previewing returns False for unknown port."""
|
|
assert not hls_manager.is_previewing(999)
|
|
|
|
|
|
def create_mock_process():
|
|
"""Create a properly mocked subprocess.Process."""
|
|
mock_process = MagicMock()
|
|
mock_process.pid = 12345
|
|
mock_process.done.return_value = False
|
|
mock_stdin = MagicMock()
|
|
mock_stdin.is_closing.return_value = False
|
|
mock_process.stdin = mock_stdin
|
|
# Make wait an async function
|
|
mock_process.wait = AsyncMock(return_value=None)
|
|
# Make stdin methods async
|
|
mock_stdin.write = MagicMock(return_value=None)
|
|
mock_stdin.drain = AsyncMock()
|
|
return mock_process
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_start_preview_sets_state(hls_manager, tmp_path):
|
|
"""is_previewing returns True after start_preview."""
|
|
with patch('asyncio.create_subprocess_exec') as mock_create:
|
|
mock_process = create_mock_process()
|
|
mock_create.return_value = mock_process
|
|
|
|
await hls_manager.start_preview(0)
|
|
|
|
assert hls_manager.is_previewing(0)
|
|
assert hls_manager._processes[0] is not None
|
|
assert hls_manager._processes[0].pid == 12345
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_start_preview_noop_when_already_running(hls_manager):
|
|
"""start_preview is no-op if already running."""
|
|
with patch('asyncio.create_subprocess_exec') as mock_create:
|
|
mock_process = create_mock_process()
|
|
mock_create.return_value = mock_process
|
|
|
|
# Start first time
|
|
await hls_manager.start_preview(0)
|
|
assert mock_create.call_count == 1
|
|
|
|
# Try to start again
|
|
await hls_manager.start_preview(0)
|
|
# Should not create new process
|
|
assert mock_create.call_count == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_start_preview_creates_hls_dir(hls_manager, tmp_path):
|
|
"""start_preview creates HLS directory if it doesn't exist."""
|
|
hls_dir = tmp_path / "new_hls"
|
|
manager = HLSPreviewManager(hls_manager.settings, hls_dir=str(hls_dir))
|
|
|
|
assert not hls_dir.exists()
|
|
|
|
with patch('asyncio.create_subprocess_exec') as mock_create:
|
|
mock_process = create_mock_process()
|
|
mock_create.return_value = mock_process
|
|
|
|
await manager.start_preview(0)
|
|
|
|
assert hls_dir.exists()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stop_preview_noop_when_not_running(hls_manager):
|
|
"""stop_preview does nothing if not running."""
|
|
# Should not raise, just log and return
|
|
await hls_manager.stop_preview(0)
|
|
assert 0 not in hls_manager._processes or hls_manager._processes[0] is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stop_preview_graceful_shutdown(hls_manager):
|
|
"""stop_preview sends quit command first."""
|
|
with patch('asyncio.create_subprocess_exec') as mock_create:
|
|
mock_process = create_mock_process()
|
|
mock_create.return_value = mock_process
|
|
|
|
await hls_manager.start_preview(0)
|
|
await hls_manager.stop_preview(0)
|
|
|
|
# Verify stdin write was called
|
|
mock_process.stdin.write.assert_called_with(b"q\n")
|
|
mock_process.stdin.drain.assert_called()
|
|
# Verify process is cleaned up
|
|
assert hls_manager._processes[0] is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stop_preview_sigterm_fallback(hls_manager):
|
|
"""stop_preview sends SIGTERM if graceful quit times out."""
|
|
with patch('asyncio.create_subprocess_exec') as mock_create:
|
|
mock_process = create_mock_process()
|
|
|
|
# First wait (graceful): timeout
|
|
# Second wait (SIGTERM): success
|
|
mock_process.wait = AsyncMock(side_effect=[
|
|
asyncio.TimeoutError(),
|
|
None # Process exits after SIGTERM
|
|
])
|
|
|
|
mock_create.return_value = mock_process
|
|
|
|
await hls_manager.start_preview(0)
|
|
await hls_manager.stop_preview(0)
|
|
|
|
# Verify terminate was called
|
|
mock_process.terminate.assert_called_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stop_preview_sigkill_fallback(hls_manager):
|
|
"""stop_preview sends SIGKILL if SIGTERM times out."""
|
|
with patch('asyncio.create_subprocess_exec') as mock_create:
|
|
mock_process = create_mock_process()
|
|
|
|
# Both waits timeout: graceful then SIGTERM
|
|
# Then SIGKILL and final wait
|
|
mock_process.wait = AsyncMock(side_effect=[
|
|
asyncio.TimeoutError(), # Graceful timeout
|
|
asyncio.TimeoutError(), # SIGTERM timeout
|
|
None # Process exits after SIGKILL
|
|
])
|
|
|
|
mock_create.return_value = mock_process
|
|
|
|
await hls_manager.start_preview(0)
|
|
await hls_manager.stop_preview(0)
|
|
|
|
# Verify kill was called
|
|
mock_process.kill.assert_called_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stop_preview_handles_process_lookup_error(hls_manager):
|
|
"""stop_preview handles ProcessLookupError gracefully."""
|
|
with patch('asyncio.create_subprocess_exec') as mock_create:
|
|
mock_process = create_mock_process()
|
|
|
|
# Graceful quit timeout, then ProcessLookupError on terminate
|
|
mock_process.wait.return_value = None
|
|
mock_process.terminate.side_effect = ProcessLookupError()
|
|
|
|
mock_create.return_value = mock_process
|
|
|
|
await hls_manager.start_preview(0)
|
|
await hls_manager.stop_preview(0)
|
|
|
|
# Should clean up gracefully
|
|
assert hls_manager._processes[0] is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stop_all_stops_multiple_ports(hls_manager):
|
|
"""stop_all stops all running previews."""
|
|
with patch('asyncio.create_subprocess_exec') as mock_create:
|
|
mock_process = create_mock_process()
|
|
mock_create.return_value = mock_process
|
|
|
|
# Start previews on ports 0, 1, 2
|
|
await hls_manager.start_preview(0)
|
|
await hls_manager.start_preview(1)
|
|
await hls_manager.start_preview(2)
|
|
|
|
assert hls_manager.is_previewing(0)
|
|
assert hls_manager.is_previewing(1)
|
|
assert hls_manager.is_previewing(2)
|
|
|
|
# Stop all
|
|
await hls_manager.stop_all()
|
|
|
|
# All should be stopped
|
|
assert not hls_manager.is_previewing(0)
|
|
assert not hls_manager.is_previewing(1)
|
|
assert not hls_manager.is_previewing(2)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_start_preview_failure_is_raised(hls_manager):
|
|
"""start_preview raises exception on subprocess creation failure."""
|
|
with patch('asyncio.create_subprocess_exec') as mock_create:
|
|
mock_create.side_effect = OSError("FFmpeg not found")
|
|
|
|
with pytest.raises(OSError):
|
|
await hls_manager.start_preview(0)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_start_preview_builds_correct_command(hls_manager):
|
|
"""start_preview builds and executes correct FFmpeg command."""
|
|
with patch('asyncio.create_subprocess_exec') as mock_create:
|
|
mock_process = create_mock_process()
|
|
mock_create.return_value = mock_process
|
|
|
|
await hls_manager.start_preview(3)
|
|
|
|
# Get the command that was passed to create_subprocess_exec
|
|
call_args = mock_create.call_args
|
|
command = call_args[0] # Positional args are unpacked
|
|
|
|
# Verify command structure
|
|
assert command[0] == hls_manager.settings.ffmpeg_path
|
|
assert "deltacast://3" in command
|
|
assert "-f" in command
|
|
assert "hls" in command
|
|
assert "-hls_time" in command
|
|
assert "5" in command
|