test: add more tests for video_app and move file validation to serializer
This commit is contained in:
Chneemann
parent
3fa00c0aa9
commit
429a686376
4 changed files with 170 additions and 36 deletions
|
|
@ -1,7 +1,25 @@
|
|||
from rest_framework import serializers
|
||||
from video_app.class_assets import VIDEO_GENRES
|
||||
from .models import Video
|
||||
|
||||
class VideoSerializer(serializers.ModelSerializer):
|
||||
class Meta:
|
||||
model = Video
|
||||
fields = "__all__"
|
||||
fields = "__all__"
|
||||
|
||||
class VideoUploadSerializer(serializers.ModelSerializer):
|
||||
class Meta:
|
||||
model = Video
|
||||
fields = ['title', 'description', 'genre', 'file_path']
|
||||
|
||||
file_path = serializers.FileField(write_only=True)
|
||||
genre = serializers.ChoiceField(choices=VIDEO_GENRES, required=True)
|
||||
title = serializers.CharField(required=True)
|
||||
description = serializers.CharField(required=True)
|
||||
|
||||
def validate_file_path(self, value):
|
||||
if not value.name.endswith('.mp4'):
|
||||
raise serializers.ValidationError("Invalid file type. Only .mp4 files are allowed.")
|
||||
if value.size > 20 * 1024 * 1024:
|
||||
raise serializers.ValidationError("File size exceeds the allowed limit of 20MB.")
|
||||
return value
|
||||
|
|
@ -3,18 +3,6 @@ from django.core.files.storage import FileSystemStorage
|
|||
from django.conf import settings
|
||||
from .models import Video
|
||||
|
||||
def validate_video_file(uploaded_file, max_size_mb=20):
|
||||
"""
|
||||
Validate a video file
|
||||
"""
|
||||
if not uploaded_file:
|
||||
return 'No file uploaded'
|
||||
if not uploaded_file.name.endswith('.mp4'):
|
||||
return 'Invalid file type. Only .mp4 files are allowed.'
|
||||
if uploaded_file.size > max_size_mb * 1024 * 1024:
|
||||
return f'File size exceeds the allowed limit of {max_size_mb}MB'
|
||||
return None
|
||||
|
||||
def save_video_file(uploaded_file):
|
||||
"""
|
||||
Save an uploaded video file and return the short name and the file path.
|
||||
|
|
|
|||
|
|
@ -1,13 +1,76 @@
|
|||
from django.test import TestCase
|
||||
import os, shutil, tempfile
|
||||
from django.conf import settings
|
||||
from django.test import TestCase, override_settings
|
||||
from django.urls import reverse
|
||||
from rest_framework.test import APIClient
|
||||
from unittest.mock import patch, MagicMock
|
||||
from django.core.files.uploadedfile import SimpleUploadedFile
|
||||
from .tasks import convert_video_to_hls, delete_original_video
|
||||
from .models import Video
|
||||
from .class_assets import VIDEO_GENRES
|
||||
from user_app.models import CustomUser
|
||||
|
||||
@override_settings(MEDIA_ROOT=tempfile.gettempdir())
|
||||
class CheckVideoResolutionsTest(TestCase):
|
||||
def setUp(self):
|
||||
self.user = CustomUser.objects.create_user(
|
||||
username='testuser', email='testuser@example.com', password='testpass'
|
||||
)
|
||||
self.client = APIClient()
|
||||
self.client.force_authenticate(user=self.user)
|
||||
self.video = Video.objects.create(title='Test Video', file_name='testvideo', is_available=True, creator=self.user)
|
||||
self.video_id = self.video.id
|
||||
|
||||
self.video_dir = os.path.join(settings.MEDIA_ROOT, 'videos', str(self.video_id))
|
||||
os.makedirs(self.video_dir, exist_ok=True)
|
||||
|
||||
with open(os.path.join(self.video_dir, 'testvideo_720p.m3u8'), 'w') as f:
|
||||
f.write('#EX1TM3U')
|
||||
|
||||
def tearDown(self):
|
||||
shutil.rmtree(self.video_dir, ignore_errors=True)
|
||||
|
||||
def test_check_video_resolutions(self):
|
||||
response = self.client.get(reverse('check_video_resolutions', args=[self.video_id]))
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertTrue(response.json()['720p'])
|
||||
self.assertFalse(response.json()['360p'])
|
||||
self.assertFalse(response.json()['1080p'])
|
||||
|
||||
def test_check_video_resolutions_invalid_id(self):
|
||||
response = self.client.get(reverse('check_video_resolutions', args=[9999]))
|
||||
self.assertEqual(response.status_code, 404)
|
||||
self.assertIn('error', response.json())
|
||||
|
||||
class VideoListViewTest(TestCase):
|
||||
def setUp(self):
|
||||
self.user = CustomUser.objects.create_user(
|
||||
username='testuser', email='testuser@example.com', password='testpass'
|
||||
)
|
||||
self.client = APIClient()
|
||||
self.client.force_authenticate(user=self.user)
|
||||
|
||||
def test_video_list_with_no_available_videos(self):
|
||||
response = self.client.get(reverse('video_list'))
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertEqual(response.data, [])
|
||||
|
||||
class GenreListViewTest(TestCase):
|
||||
def setUp(self):
|
||||
self.user = CustomUser.objects.create_user(
|
||||
username='testuser', email='testuser@example.com', password='testpass'
|
||||
)
|
||||
self.client = APIClient()
|
||||
self.client.force_authenticate(user=self.user)
|
||||
|
||||
def test_genre_list_requires_authentication(self):
|
||||
client = APIClient()
|
||||
response = client.get(reverse('genre_list'))
|
||||
self.assertEqual(response.status_code, 401)
|
||||
|
||||
class VideoTasksTest(TestCase):
|
||||
@patch('subprocess.run')
|
||||
def test_convert_video_to_hls(self, mock_subprocess_run):
|
||||
"""
|
||||
Tests whether the HLS conversion script is called correctly and the correct parameters are passed.
|
||||
"""
|
||||
mock_subprocess_run.return_value = MagicMock()
|
||||
convert_video_to_hls('test/source.mp4', '1920x1080', 1)
|
||||
mock_subprocess_run.assert_called_once()
|
||||
|
|
@ -16,8 +79,77 @@ class VideoTasksTest(TestCase):
|
|||
|
||||
@patch('os.remove')
|
||||
def test_delete_original_video(self, mock_os_remove):
|
||||
"""
|
||||
Checks whether the original video file is successfully deleted after conversion.
|
||||
"""
|
||||
delete_original_video('test/source.mp4')
|
||||
mock_os_remove.assert_called_once_with('test/source.mp4')
|
||||
mock_os_remove.assert_called_once_with('test/source.mp4')
|
||||
|
||||
@patch('os.remove', side_effect=FileNotFoundError)
|
||||
def test_delete_original_video_handles_missing_file(self, mock_os_remove):
|
||||
try:
|
||||
delete_original_video('nonexistent.mp4')
|
||||
except FileNotFoundError:
|
||||
self.fail("delete_original_video should handle FileNotFoundError gracefully")
|
||||
|
||||
class VideoUploadValidationTest(TestCase):
|
||||
def setUp(self):
|
||||
self.user = CustomUser.objects.create_user(
|
||||
username='uploader', email='uploader@example.com', password='testpass'
|
||||
)
|
||||
self.client = APIClient()
|
||||
self.client.force_authenticate(user=self.user)
|
||||
|
||||
def test_video_upload_missing_required_fields(self):
|
||||
required_fields = ['title', 'description', 'genre', 'file_path']
|
||||
for field in required_fields:
|
||||
with self.subTest(missing=field):
|
||||
uploaded_file = SimpleUploadedFile("video.mp4", b"dummy content", content_type="video/mp4")
|
||||
data = {
|
||||
'title': 'Test title',
|
||||
'description': 'Test description',
|
||||
'genre': VIDEO_GENRES[0][0],
|
||||
'file_path': uploaded_file,
|
||||
}
|
||||
del data[field]
|
||||
|
||||
response = self.client.post(reverse('video_upload'), data, format='multipart')
|
||||
self.assertEqual(response.status_code, 400)
|
||||
self.assertIn(field, response.data)
|
||||
|
||||
def test_video_upload_invalid_filetype(self):
|
||||
txt_file = SimpleUploadedFile("document.txt", b"dummy content", content_type="text/plain")
|
||||
data = {
|
||||
'title': 'Invalid File',
|
||||
'description': 'Test description',
|
||||
'genre': VIDEO_GENRES[0][0],
|
||||
'file_path': txt_file,
|
||||
}
|
||||
|
||||
response = self.client.post(reverse('video_upload'), data, format='multipart')
|
||||
self.assertEqual(response.status_code, 400)
|
||||
self.assertIn('file_path', response.data)
|
||||
|
||||
def test_video_upload_file_too_large(self):
|
||||
big_file = SimpleUploadedFile("video.mp4", b"a" * (21 * 1024 * 1024), content_type="video/mp4")
|
||||
data = {
|
||||
'title': 'Too big',
|
||||
'description': 'Test description',
|
||||
'genre': VIDEO_GENRES[0][0],
|
||||
'file_path': big_file,
|
||||
}
|
||||
|
||||
response = self.client.post(reverse('video_upload'), data, format='multipart')
|
||||
self.assertEqual(response.status_code, 400)
|
||||
self.assertIn('file_path', response.data)
|
||||
self.assertIn('File size exceeds', str(response.data['file_path']))
|
||||
|
||||
def test_video_upload_success(self):
|
||||
small_file = SimpleUploadedFile("video.mp4", b"dummy video content", content_type="video/mp4")
|
||||
data = {
|
||||
'title': 'Valid Upload',
|
||||
'description': 'A valid video upload test',
|
||||
'genre': VIDEO_GENRES[0][0],
|
||||
'file_path': small_file,
|
||||
}
|
||||
|
||||
response = self.client.post(reverse('video_upload'), data, format='multipart')
|
||||
self.assertEqual(response.status_code, 201)
|
||||
self.assertIn('title', response.data)
|
||||
|
|
@ -1,10 +1,9 @@
|
|||
from django.forms import ValidationError
|
||||
from rest_framework.decorators import api_view, permission_classes
|
||||
from rest_framework.permissions import IsAuthenticated
|
||||
from rest_framework.response import Response
|
||||
from .serializer import VideoSerializer
|
||||
from .serializer import VideoSerializer, VideoUploadSerializer
|
||||
from .models import Video
|
||||
from .services import validate_video_file, save_video_file, create_video_record
|
||||
from .services import save_video_file, create_video_record
|
||||
from .class_assets import VIDEO_GENRES
|
||||
from django.conf import settings
|
||||
from django.http import JsonResponse
|
||||
|
|
@ -51,7 +50,7 @@ def check_video_resolutions(request, id):
|
|||
video_file_path = os.path.join(video_dir, video_file_name)
|
||||
result[res] = os.path.exists(video_file_path)
|
||||
|
||||
return JsonResponse(result)
|
||||
return Response(result)
|
||||
|
||||
@api_view(['POST'])
|
||||
@permission_classes([IsAuthenticated])
|
||||
|
|
@ -59,19 +58,16 @@ def video_upload(request):
|
|||
"""
|
||||
Handles the upload of a video file:
|
||||
"""
|
||||
uploaded_file = request.FILES.get('file_path')
|
||||
|
||||
error = validate_video_file(uploaded_file)
|
||||
if error:
|
||||
return Response({'error': error}, status=400)
|
||||
serializer = VideoUploadSerializer(data=request.data)
|
||||
if not serializer.is_valid():
|
||||
return Response(serializer.errors, status=400)
|
||||
|
||||
uploaded_file = serializer.validated_data['file_path']
|
||||
short_name, file_path = save_video_file(uploaded_file)
|
||||
|
||||
try:
|
||||
short_name, file_path = save_video_file(uploaded_file)
|
||||
with transaction.atomic():
|
||||
video = create_video_record(request.data, request.user, short_name, file_path)
|
||||
video = create_video_record(serializer.validated_data, request.user, short_name, file_path)
|
||||
return Response(VideoSerializer(video).data, status=201)
|
||||
|
||||
except ValidationError as ve:
|
||||
return Response({'error': f'Validation Error: {str(ve)}'}, status=400)
|
||||
except Exception as e:
|
||||
return Response({'error': f'Error while saving video file: {str(e)}'}, status=500)
|
||||
return Response({'error': f'Error while saving video: {str(e)}'}, status=500)
|
||||
Loading…
Reference in a new issue