from django.test import TestCase, Client from django.urls import reverse from django.contrib.auth import get_user_model from django.utils import timezone from django.core.exceptions import ValidationError from django.core.files.uploadedfile import SimpleUploadedFile from datetime import datetime, timedelta import json from unittest.mock import patch, MagicMock from .models import ( Codec, Zone, Channel, ChannelZone, Jingle, Show, EPGEntry ) from .services import ChannelService, EPGService, JingleService, CodecService from .forms import ChannelForm, EPGEntryForm, EPGBulkImportForm User = get_user_model() class CodecModelTest(TestCase): """ Test cases for Codec model. Tests model creation, validation, and methods. """ def setUp(self): """Set up test data.""" self.codec_data = { 'name': 'H.264', 'codec_type': 'video', 'description': 'Advanced Video Coding', 'mime_type': 'video/mp4', 'bitrate_min': 1000, 'bitrate_max': 10000 } def test_codec_creation(self): """Test codec creation with valid data.""" codec = Codec.objects.create(**self.codec_data) self.assertEqual(codec.name, 'H.264') self.assertEqual(codec.codec_type, 'video') self.assertTrue(codec.is_active) self.assertIsNotNone(codec.created_at) def test_codec_str_representation(self): """Test codec string representation.""" codec = Codec.objects.create(**self.codec_data) self.assertEqual(str(codec), 'H.264 (video)') def test_bitrate_validation(self): """Test bitrate validation.""" # Test invalid bitrate range invalid_data = self.codec_data.copy() invalid_data['bitrate_max'] = 500 # Less than min codec = Codec(**invalid_data) with self.assertRaises(ValidationError): codec.full_clean() def test_bitrate_range_property(self): """Test bitrate range property.""" codec = Codec.objects.create(**self.codec_data) expected_range = "1000 - 10000 kbps" self.assertEqual(codec.bitrate_range, expected_range) # Test with only min bitrate codec.bitrate_max = None codec.save() self.assertEqual(codec.bitrate_range, "≥ 1000 kbps") class ZoneModelTest(TestCase): """ Test cases for Zone model. Tests model creation, validation, and relationships. """ def setUp(self): """Set up test data.""" self.zone_data = { 'name': 'North America', 'code': 'NA', 'description': 'North American distribution zone', 'timezone': 'America/New_York' } def test_zone_creation(self): """Test zone creation with valid data.""" zone = Zone.objects.create(**self.zone_data) self.assertEqual(zone.name, 'North America') self.assertEqual(zone.code, 'NA') self.assertTrue(zone.is_active) def test_zone_str_representation(self): """Test zone string representation.""" zone = Zone.objects.create(**self.zone_data) self.assertEqual(str(zone), 'North America (NA)') def test_unique_code_constraint(self): """Test unique code constraint.""" Zone.objects.create(**self.zone_data) # Try to create another zone with same code duplicate_data = self.zone_data.copy() duplicate_data['name'] = 'Different Name' with self.assertRaises(Exception): # IntegrityError Zone.objects.create(**duplicate_data) class ChannelModelTest(TestCase): """ Test cases for Channel model. Tests model creation, validation, properties, and relationships. """ def setUp(self): """Set up test data.""" self.zone = Zone.objects.create( name='Test Zone', code='TZ', description='Test zone' ) self.channel_data = { 'name': 'Test Channel', 'call_sign': 'TEST', 'channel_number': '1.1', 'channel_type': 'terrestrial', 'description': 'Test channel description', 'content_rating': 'general', 'language': 'en', 'country': 'US' } def test_channel_creation(self): """Test channel creation with valid data.""" channel = Channel.objects.create(**self.channel_data) self.assertEqual(channel.name, 'Test Channel') self.assertEqual(channel.call_sign, 'TEST') self.assertTrue(channel.is_active) self.assertIsNotNone(channel.slug) def test_channel_str_representation(self): """Test channel string representation.""" channel = Channel.objects.create(**self.channel_data) self.assertEqual(str(channel), 'Test Channel (TEST)') def test_slug_generation(self): """Test automatic slug generation.""" channel = Channel.objects.create(**self.channel_data) self.assertEqual(channel.slug, 'test-channel') def test_active_zones_property(self): """Test active zones property.""" channel = Channel.objects.create(**self.channel_data) # Create channel zone relationship ChannelZone.objects.create( channel=channel, zone=self.zone, is_active=True ) active_zones = channel.active_zones self.assertEqual(active_zones.count(), 1) self.assertEqual(active_zones.first(), self.zone) def test_zones_count_property(self): """Test active zones count property.""" channel = Channel.objects.create(**self.channel_data) # Initially no zones self.assertEqual(channel.active_zones_count, 0) # Add a zone ChannelZone.objects.create( channel=channel, zone=self.zone, is_active=True ) # Refresh from database channel.refresh_from_db() self.assertEqual(channel.active_zones_count, 1) class EPGEntryModelTest(TestCase): """ Test cases for EPGEntry model. Tests model creation, validation, properties, and methods. """ def setUp(self): """Set up test data.""" self.channel = Channel.objects.create( name='Test Channel', call_sign='TEST', channel_type='terrestrial' ) self.show = Show.objects.create( title='Test Show', show_type='series', description='Test show description' ) self.now = timezone.now() self.epg_data = { 'channel': self.channel, 'show': self.show, 'start_time': self.now, 'end_time': self.now + timedelta(hours=1), 'episode_title': 'Test Episode', 'episode_number': 1, 'season_number': 1 } def test_epg_entry_creation(self): """Test EPG entry creation with valid data.""" entry = EPGEntry.objects.create(**self.epg_data) self.assertEqual(entry.channel, self.channel) self.assertEqual(entry.show, self.show) self.assertEqual(entry.episode_title, 'Test Episode') def test_epg_entry_str_representation(self): """Test EPG entry string representation.""" entry = EPGEntry.objects.create(**self.epg_data) expected_str = f"{self.channel.name}: {self.show.title} - {self.now.strftime('%Y-%m-%d %H:%M')}" self.assertEqual(str(entry), expected_str) def test_duration_property(self): """Test duration property calculation.""" entry = EPGEntry.objects.create(**self.epg_data) expected_duration = timedelta(hours=1) self.assertEqual(entry.duration, expected_duration) def test_is_currently_airing_property(self): """Test is currently airing property.""" # Create entry that is currently airing current_entry = EPGEntry.objects.create( channel=self.channel, show=self.show, start_time=self.now - timedelta(minutes=30), end_time=self.now + timedelta(minutes=30) ) # Create entry that is not currently airing future_entry = EPGEntry.objects.create( channel=self.channel, show=self.show, start_time=self.now + timedelta(hours=1), end_time=self.now + timedelta(hours=2) ) self.assertTrue(current_entry.is_currently_airing) self.assertFalse(future_entry.is_currently_airing) def test_time_validation(self): """Test start/end time validation.""" # Test invalid time range (end before start) invalid_data = self.epg_data.copy() invalid_data['end_time'] = self.now - timedelta(hours=1) entry = EPGEntry(**invalid_data) with self.assertRaises(ValidationError): entry.full_clean() class ChannelServiceTest(TestCase): """ Test cases for ChannelService. Tests service methods for channel management. """ def setUp(self): """Set up test data.""" self.zone = Zone.objects.create( name='Test Zone', code='TZ', description='Test zone' ) self.video_codec = Codec.objects.create( name='H.264', codec_type='video' ) self.audio_codec = Codec.objects.create( name='AAC', codec_type='audio' ) self.channel_data = { 'name': 'Test Channel', 'call_sign': 'TEST', 'channel_type': 'terrestrial' } self.zone_configs = [{ 'zone_id': self.zone.id, 'video_codec': self.video_codec, 'audio_codec': self.audio_codec, 'bitrate': 5000, 'resolution_width': 1920, 'resolution_height': 1080, 'frame_rate': 30.0 }] def test_create_channel_with_zones(self): """Test creating channel with zone configurations.""" channel = ChannelService.create_channel_with_zones( self.channel_data, self.zone_configs ) self.assertEqual(channel.name, 'Test Channel') self.assertEqual(channel.channel_zones.count(), 1) zone_config = channel.channel_zones.first() self.assertEqual(zone_config.zone, self.zone) self.assertEqual(zone_config.video_codec, self.video_codec) def test_get_channel_statistics(self): """Test getting channel statistics.""" # Create test channels Channel.objects.create( name='Channel 1', call_sign='CH1', channel_type='terrestrial', is_active=True, is_hd=True ) Channel.objects.create( name='Channel 2', call_sign='CH2', channel_type='satellite', is_active=False ) stats = ChannelService.get_channel_statistics() self.assertEqual(stats['total_channels'], 2) self.assertEqual(stats['active_channels'], 1) self.assertEqual(stats['hd_channels'], 1) self.assertIn('channels_by_type', stats) def test_validate_channel_configuration(self): """Test channel configuration validation.""" channel = Channel.objects.create(**self.channel_data) # Test channel without zones warnings = ChannelService.validate_channel_configuration(channel) self.assertIn("Channel has no zone configurations", warnings) # Add zone configuration ChannelZone.objects.create( channel=channel, zone=self.zone, video_codec=self.video_codec, audio_codec=self.audio_codec ) # Test again - should have fewer warnings warnings = ChannelService.validate_channel_configuration(channel) self.assertNotIn("Channel has no zone configurations", warnings) class EPGServiceTest(TestCase): """ Test cases for EPGService. Tests service methods for EPG management. """ def setUp(self): """Set up test data.""" self.channel = Channel.objects.create( name='Test Channel', call_sign='TEST', channel_type='terrestrial' ) self.show = Show.objects.create( title='Test Show', show_type='series' ) def test_import_epg_from_csv(self): """Test EPG import from CSV.""" csv_content = """show_title,start_time,end_time,episode_title,show_type Test Show,2024-01-01 10:00,2024-01-01 11:00,Episode 1,series Another Show,2024-01-01 11:00,2024-01-01 12:00,Episode 1,movie""" results = EPGService.import_epg_from_csv( csv_content=csv_content, channel_id=self.channel.id, overwrite=False ) self.assertEqual(results['success_count'], 2) self.assertEqual(results['error_count'], 0) self.assertEqual(EPGEntry.objects.count(), 2) def test_export_epg_to_csv(self): """Test EPG export to CSV.""" # Create test EPG entry start_time = timezone.make_aware(datetime(2024, 1, 1, 10, 0)) end_time = timezone.make_aware(datetime(2024, 1, 1, 11, 0)) EPGEntry.objects.create( channel=self.channel, show=self.show, start_time=start_time, end_time=end_time, episode_title='Test Episode' ) csv_content = EPGService.export_epg_to_csv(channel_id=self.channel.id) self.assertIn('Test Channel', csv_content) self.assertIn('Test Show', csv_content) self.assertIn('Test Episode', csv_content) def test_get_schedule_conflicts(self): """Test schedule conflict detection.""" base_time = timezone.make_aware(datetime(2024, 1, 1, 10, 0)) # Create overlapping entries EPGEntry.objects.create( channel=self.channel, show=self.show, start_time=base_time, end_time=base_time + timedelta(hours=1, minutes=30) ) EPGEntry.objects.create( channel=self.channel, show=self.show, start_time=base_time + timedelta(hours=1), end_time=base_time + timedelta(hours=2) ) conflicts = EPGService.get_schedule_conflicts(self.channel.id, days=1) self.assertEqual(len(conflicts), 1) self.assertEqual(conflicts[0]['overlap_duration'], 30.0) # 30 minutes class ChannelViewTest(TestCase): """ Test cases for Channel views. Tests view functionality, permissions, and responses. """ def setUp(self): """Set up test data.""" self.client = Client() # Create test user self.user = User.objects.create_user( email='test@example.com', password='testpass123', name='Test User' ) # Create admin user self.admin_user = User.objects.create_user( email='admin@example.com', password='adminpass123', name='Admin User', is_staff=True, is_superuser=True ) self.channel = Channel.objects.create( name='Test Channel', call_sign='TEST', channel_type='terrestrial' ) def test_channel_list_view_authenticated(self): """Test channel list view for authenticated users.""" self.client.login(email='test@example.com', password='testpass123') response = self.client.get(reverse('channels:channel_list')) self.assertEqual(response.status_code, 200) self.assertContains(response, 'Test Channel') def test_channel_list_view_unauthenticated(self): """Test channel list view for unauthenticated users.""" response = self.client.get(reverse('channels:channel_list')) # Should redirect to login self.assertEqual(response.status_code, 302) def test_channel_detail_view(self): """Test channel detail view.""" self.client.login(email='test@example.com', password='testpass123') response = self.client.get( reverse('channels:channel_detail', kwargs={'slug': self.channel.slug}) ) self.assertEqual(response.status_code, 200) self.assertContains(response, 'Test Channel') def test_channel_create_view_admin(self): """Test channel creation view for admin users.""" self.client.login(email='admin@example.com', password='adminpass123') response = self.client.get(reverse('channels:channel_create')) self.assertEqual(response.status_code, 200) self.assertContains(response, 'Create Channel') def test_channel_create_view_regular_user(self): """Test channel creation view for regular users.""" self.client.login(email='test@example.com', password='testpass123') response = self.client.get(reverse('channels:channel_create')) # Should be forbidden self.assertEqual(response.status_code, 403) def test_channel_create_post(self): """Test channel creation via POST.""" self.client.login(email='admin@example.com', password='adminpass123') data = { 'name': 'New Channel', 'call_sign': 'NEW', 'channel_type': 'satellite', 'description': 'New channel description', 'content_rating': 'general', 'language': 'en', 'country': 'US' } response = self.client.post(reverse('channels:channel_create'), data) # Should redirect after successful creation self.assertEqual(response.status_code, 302) # Check if channel was created self.assertTrue(Channel.objects.filter(name='New Channel').exists()) class ChannelFormTest(TestCase): """ Test cases for Channel forms. Tests form validation and functionality. """ def test_channel_form_valid_data(self): """Test channel form with valid data.""" form_data = { 'name': 'Test Channel', 'call_sign': 'TEST', 'channel_type': 'terrestrial', 'description': 'Test description', 'content_rating': 'general', 'language': 'en', 'country': 'US' } form = ChannelForm(data=form_data) self.assertTrue(form.is_valid()) def test_channel_form_invalid_data(self): """Test channel form with invalid data.""" form_data = { 'name': '', # Required field 'call_sign': 'TEST', 'channel_type': 'invalid_type' } form = ChannelForm(data=form_data) self.assertFalse(form.is_valid()) self.assertIn('name', form.errors) self.assertIn('channel_type', form.errors) class EPGEntryFormTest(TestCase): """ Test cases for EPGEntry forms. Tests form validation and functionality. """ def setUp(self): """Set up test data.""" self.channel = Channel.objects.create( name='Test Channel', call_sign='TEST', channel_type='terrestrial' ) self.show = Show.objects.create( title='Test Show', show_type='series' ) def test_epg_entry_form_valid_data(self): """Test EPG entry form with valid data.""" start_time = timezone.now() + timedelta(hours=1) end_time = start_time + timedelta(hours=1) form_data = { 'channel': self.channel.id, 'show': self.show.id, 'start_time': start_time.strftime('%Y-%m-%d %H:%M'), 'end_time': end_time.strftime('%Y-%m-%d %H:%M'), 'episode_title': 'Test Episode' } form = EPGEntryForm(data=form_data) self.assertTrue(form.is_valid()) def test_epg_entry_form_time_validation(self): """Test EPG entry form time validation.""" start_time = timezone.now() + timedelta(hours=1) end_time = start_time - timedelta(hours=1) # Invalid: end before start form_data = { 'channel': self.channel.id, 'show': self.show.id, 'start_time': start_time.strftime('%Y-%m-%d %H:%M'), 'end_time': end_time.strftime('%Y-%m-%d %H:%M') } form = EPGEntryForm(data=form_data) self.assertFalse(form.is_valid()) class APIEndpointTest(TestCase): """ Test cases for API endpoints. Tests API functionality and responses. """ def setUp(self): """Set up test data.""" self.client = Client() self.user = User.objects.create_user( email='test@example.com', password='testpass123', name='Test User' ) self.channel = Channel.objects.create( name='Test Channel', call_sign='TEST', channel_type='terrestrial' ) self.zone = Zone.objects.create( name='Test Zone', code='TZ' ) def test_channel_zones_api(self): """Test channel zones API endpoint.""" self.client.login(email='test@example.com', password='testpass123') # Create channel zone ChannelZone.objects.create( channel=self.channel, zone=self.zone, bitrate=5000 ) response = self.client.get( reverse('channels:channel_zones_api', kwargs={'channel_id': self.channel.id}) ) self.assertEqual(response.status_code, 200) data = json.loads(response.content) self.assertEqual(len(data), 1) self.assertEqual(data[0]['zone_name'], 'Test Zone') def test_epg_calendar_api(self): """Test EPG calendar API endpoint.""" self.client.login(email='test@example.com', password='testpass123') # Create show and EPG entry show = Show.objects.create( title='Test Show', show_type='series' ) start_time = timezone.now() EPGEntry.objects.create( channel=self.channel, show=show, start_time=start_time, end_time=start_time + timedelta(hours=1) ) response = self.client.get(reverse('channels:epg_calendar_api')) self.assertEqual(response.status_code, 200) data = json.loads(response.content) self.assertEqual(len(data), 1) self.assertIn('Test Channel: Test Show', data[0]['title']) class TaskTest(TestCase): """ Test cases for Celery tasks. Tests asynchronous task functionality. """ def setUp(self): """Set up test data.""" self.channel = Channel.objects.create( name='Test Channel', call_sign='TEST', channel_type='terrestrial' ) @patch('apps.channels.tasks.send_mail') def test_epg_import_notification_task(self, mock_send_mail): """Test EPG import notification task.""" from .tasks import send_epg_import_notification results = { 'success_count': 10, 'error_count': 2, 'errors': ['Error 1', 'Error 2'], 'warnings': ['Warning 1'] } send_epg_import_notification( user_email='test@example.com', channel_id=self.channel.id, results=results ) # Check if send_mail was called mock_send_mail.assert_called_once() # Check email content call_args = mock_send_mail.call_args self.assertIn('EPG Import Completed', call_args[1]['subject']) self.assertIn('test@example.com', call_args[1]['recipient_list']) def test_cleanup_expired_epg_entries_task(self): """Test cleanup expired EPG entries task.""" from .tasks import cleanup_expired_epg_entries # Create expired EPG entry show = Show.objects.create( title='Old Show', show_type='series' ) old_time = timezone.now() - timedelta(days=35) EPGEntry.objects.create( channel=self.channel, show=show, start_time=old_time, end_time=old_time + timedelta(hours=1) ) # Create recent EPG entry recent_time = timezone.now() - timedelta(days=5) EPGEntry.objects.create( channel=self.channel, show=show, start_time=recent_time, end_time=recent_time + timedelta(hours=1) ) # Run cleanup task result = cleanup_expired_epg_entries() self.assertTrue(result['success']) self.assertEqual(result['deleted_count'], 1) # Check that only recent entry remains self.assertEqual(EPGEntry.objects.count(), 1) remaining_entry = EPGEntry.objects.first() self.assertEqual(remaining_entry.start_time.date(), recent_time.date()) class IntegrationTest(TestCase): """ Integration test cases. Tests complete workflows and interactions between components. """ def setUp(self): """Set up test data.""" self.client = Client() # Create admin user self.admin_user = User.objects.create_user( email='admin@example.com', password='adminpass123', name='Admin User', is_staff=True, is_superuser=True ) # Create codecs self.video_codec = Codec.objects.create( name='H.264', codec_type='video' ) self.audio_codec = Codec.objects.create( name='AAC', codec_type='audio' ) # Create zone self.zone = Zone.objects.create( name='Test Zone', code='TZ' ) def test_complete_channel_setup_workflow(self): """Test complete channel setup workflow.""" self.client.login(email='admin@example.com', password='adminpass123') # Step 1: Create channel channel_data = { 'name': 'Integration Test Channel', 'call_sign': 'ITC', 'channel_type': 'terrestrial', 'description': 'Integration test channel', 'content_rating': 'general', 'language': 'en', 'country': 'US' } response = self.client.post(reverse('channels:channel_create'), channel_data) self.assertEqual(response.status_code, 302) channel = Channel.objects.get(call_sign='ITC') # Step 2: Add zone configuration zone_config_data = { 'channel': channel.id, 'zone': self.zone.id, 'video_codec': self.video_codec.id, 'audio_codec': self.audio_codec.id, 'bitrate': 5000, 'resolution_width': 1920, 'resolution_height': 1080, 'frame_rate': 30.0 } ChannelZone.objects.create(**zone_config_data) # Step 3: Create show show = Show.objects.create( title='Integration Test Show', show_type='series', description='Test show for integration' ) # Step 4: Add EPG entry start_time = timezone.now() + timedelta(hours=1) end_time = start_time + timedelta(hours=1) epg_data = { 'channel': channel.id, 'show': show.id, 'start_time': start_time.strftime('%Y-%m-%d %H:%M'), 'end_time': end_time.strftime('%Y-%m-%d %H:%M'), 'episode_title': 'Integration Test Episode' } response = self.client.post(reverse('channels:epg_create'), epg_data) self.assertEqual(response.status_code, 302) # Step 5: Verify complete setup channel.refresh_from_db() # Check channel exists and is configured self.assertEqual(channel.name, 'Integration Test Channel') self.assertEqual(channel.channel_zones.count(), 1) # Check EPG entry exists epg_entry = EPGEntry.objects.get(channel=channel, show=show) self.assertEqual(epg_entry.episode_title, 'Integration Test Episode') # Check API endpoints work zones_response = self.client.get( reverse('channels:channel_zones_api', kwargs={'channel_id': channel.id}) ) self.assertEqual(zones_response.status_code, 200) calendar_response = self.client.get(reverse('channels:epg_calendar_api')) self.assertEqual(calendar_response.status_code, 200) # Check statistics stats = ChannelService.get_channel_statistics() self.assertGreaterEqual(stats['total_channels'], 1) self.assertGreaterEqual(stats['active_channels'], 1)