From 9dd83b2f9a260efdb165b8351cddcba1af7ed048 Mon Sep 17 00:00:00 2001
From: Zac Gaetano <zgaetano@wilddragon.net>
Date: Tue, 14 Apr 2026 09:21:12 -0400
Subject: [PATCH] Add backend/tests/test_hls.py

---
 backend/tests/test_hls.py | 327 ++++++++++++++++++++++++++++++++++++++
 1 file changed, 327 insertions(+)
 create mode 100644 backend/tests/test_hls.py

diff --git a/backend/tests/test_hls.py b/backend/tests/test_hls.py
new file mode 100644
index 0000000..42c9813
--- /dev/null
+++ b/backend/tests/test_hls.py
@@ -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