Add backend/tests/test_hls.py
This commit is contained in:
parent
67fdd91285
commit
9dd83b2f9a
1 changed files with 327 additions and 0 deletions
327
backend/tests/test_hls.py
Normal file
327
backend/tests/test_hls.py
Normal file
|
|
@ -0,0 +1,327 @@
|
|||
"""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
|
||||
Loading…
Reference in a new issue