""" Unit tests for streams app services. """ import tempfile import shutil import os from unittest.mock import patch, MagicMock, call from django.test import TestCase from django.contrib.auth import get_user_model from django.utils import timezone from apps.streams.models import Channel, StreamSession, VideoConfiguration, AudioConfiguration from apps.streams.services import StreamCaptureService User = get_user_model() class StreamCaptureServiceTest(TestCase): """Test cases for StreamCaptureService.""" def setUp(self): self.user = User.objects.create_user( username='testuser', email='test@example.com', password='testpass123' ) self.temp_dir = tempfile.mkdtemp() self.channel = Channel.objects.create( name='Test Channel', slug='test-channel', hls_url='https://example.com/stream.m3u8', output_directory=self.temp_dir, created_by=self.user ) self.video_config = VideoConfiguration.objects.create( name='Test Video', codec='h264', bitrate=1000000, width=1280, height=720, framerate=25.0, created_by=self.user ) self.audio_config = AudioConfiguration.objects.create( name='Test Audio', codec='aac', bitrate=128000, sample_rate=44100, channels=2, created_by=self.user ) self.service = StreamCaptureService() def tearDown(self): shutil.rmtree(self.temp_dir, ignore_errors=True) def test_build_ffmpeg_command_basic(self): """Test basic FFmpeg command building.""" cmd = self.service.build_ffmpeg_command(self.channel) self.assertIn('ffmpeg', cmd[0]) self.assertIn('-i', cmd) self.assertIn(self.channel.hls_url, cmd) self.assertIn('-f', cmd) self.assertIn('hls', cmd) self.assertIn('-hls_time', cmd) self.assertIn(str(self.channel.segment_duration), cmd) def test_build_ffmpeg_command_with_configs(self): """Test FFmpeg command building with video/audio configs.""" cmd = self.service.build_ffmpeg_command( self.channel, self.video_config, self.audio_config ) # Check video codec self.assertIn('-c:v', cmd) self.assertIn('libx264', cmd) self.assertIn('-b:v', cmd) self.assertIn('1000000', cmd) # Check audio codec self.assertIn('-c:a', cmd) self.assertIn('aac', cmd) self.assertIn('-b:a', cmd) self.assertIn('128000', cmd) # Check resolution self.assertIn('-s', cmd) self.assertIn('1280x720', cmd) # Check framerate self.assertIn('-r', cmd) self.assertIn('25', cmd) def test_ensure_output_directory(self): """Test output directory creation.""" test_dir = os.path.join(self.temp_dir, 'new_directory') self.assertFalse(os.path.exists(test_dir)) self.service.ensure_output_directory(test_dir) self.assertTrue(os.path.exists(test_dir)) self.assertTrue(os.path.isdir(test_dir)) @patch('subprocess.Popen') def test_start_capture_success(self, mock_popen): """Test successful stream capture start.""" # Mock successful process mock_process = MagicMock() mock_process.pid = 12345 mock_process.poll.return_value = None # Process is running mock_popen.return_value = mock_process session = self.service.start_capture(self.channel) self.assertIsNotNone(session) self.assertEqual(session.channel, self.channel) self.assertEqual(session.status, 'active') self.assertEqual(session.process_id, 12345) # Verify FFmpeg was called mock_popen.assert_called_once() @patch('subprocess.Popen') def test_start_capture_failure(self, mock_popen): """Test stream capture start failure.""" # Mock failed process mock_popen.side_effect = Exception("FFmpeg not found") session = self.service.start_capture(self.channel) self.assertIsNone(session) @patch('subprocess.Popen') @patch('os.kill') def test_stop_capture_success(self, mock_kill, mock_popen): """Test successful stream capture stop.""" # Create active session session = StreamSession.objects.create( channel=self.channel, status='active', process_id=12345, started_at=timezone.now() ) result = self.service.stop_capture(session) self.assertTrue(result) session.refresh_from_db() self.assertEqual(session.status, 'completed') self.assertIsNotNone(session.ended_at) # Verify process was killed mock_kill.assert_called_once_with(12345, 15) # SIGTERM def test_stop_capture_no_process(self): """Test stopping capture with no process ID.""" session = StreamSession.objects.create( channel=self.channel, status='active', started_at=timezone.now() ) result = self.service.stop_capture(session) self.assertTrue(result) session.refresh_from_db() self.assertEqual(session.status, 'completed') @patch('subprocess.Popen') def test_capture_with_retry_success(self, mock_popen): """Test capture with retry logic - success case.""" mock_process = MagicMock() mock_process.pid = 12345 mock_process.poll.return_value = None mock_popen.return_value = mock_process session = self.service.capture_with_retry(self.channel) self.assertIsNotNone(session) self.assertEqual(mock_popen.call_count, 1) @patch('subprocess.Popen') @patch('time.sleep') def test_capture_with_retry_failure_then_success(self, mock_sleep, mock_popen): """Test capture with retry logic - failure then success.""" # First call fails, second succeeds mock_process = MagicMock() mock_process.pid = 12345 mock_process.poll.return_value = None mock_popen.side_effect = [Exception("Network error"), mock_process] session = self.service.capture_with_retry(self.channel) self.assertIsNotNone(session) self.assertEqual(mock_popen.call_count, 2) mock_sleep.assert_called_once_with(10) # retry_interval @patch('subprocess.Popen') @patch('time.sleep') def test_capture_with_retry_all_failures(self, mock_sleep, mock_popen): """Test capture with retry logic - all attempts fail.""" mock_popen.side_effect = Exception("Persistent error") session = self.service.capture_with_retry(self.channel) self.assertIsNone(session) self.assertEqual(mock_popen.call_count, self.channel.retry_attempts) self.assertEqual(mock_sleep.call_count, self.channel.retry_attempts - 1) @patch('os.listdir') @patch('os.path.getmtime') @patch('os.remove') def test_cleanup_old_segments(self, mock_remove, mock_getmtime, mock_listdir): """Test cleanup of old segments.""" # Mock directory listing mock_listdir.return_value = ['segment_001.ts', 'segment_002.ts', 'segment_003.ts'] # Mock file modification times (old files) old_time = timezone.now().timestamp() - 3600 # 1 hour ago mock_getmtime.return_value = old_time output_path = self.channel.get_output_path() cleaned_count = self.service.cleanup_old_segments(output_path, max_age_hours=0.5) self.assertEqual(cleaned_count, 3) self.assertEqual(mock_remove.call_count, 3) def test_get_capture_command_args(self): """Test capture command arguments generation.""" args = self.service.get_capture_command_args( self.channel, self.video_config, self.audio_config ) self.assertIsInstance(args, dict) self.assertIn('input_url', args) self.assertIn('output_path', args) self.assertIn('video_codec', args) self.assertIn('audio_codec', args) self.assertEqual(args['input_url'], self.channel.hls_url) self.assertEqual(args['video_codec'], 'libx264') self.assertEqual(args['audio_codec'], 'aac')