Merge pull request #3662 from ozer550/Validate_file_formats

Validate file format in upload_url endpoint

contentcuration/contentcuration/models.py

@@ -8,7 +8,6 @@
 from datetime import datetime
 
 import pytz
-from celery import states
 from django.conf import settings
 from django.contrib.auth.base_user import AbstractBaseUser
 from django.contrib.auth.base_user import BaseUserManager
@@ -2197,6 +2196,12 @@ def save(self, set_by_file_on_disk=True, *args, **kwargs):
             2. fill the other fields accordingly
         """
         from contentcuration.utils.user import calculate_user_storage
+
+        # check if the file format exists in file_formats.choices
+        if self.file_format_id:
+            if self.file_format_id not in dict(file_formats.choices):
+                raise ValidationError("Invalid file_format")
+
         if set_by_file_on_disk and self.file_on_disk:  # if file_on_disk is supplied, hash out the file
             if self.checksum is None or self.checksum == "":
                 md5 = hashlib.md5()

contentcuration/contentcuration/tests/test_models.py

@@ -4,6 +4,7 @@
 import pytest
 from django.conf import settings
 from django.core.cache import cache
+from django.core.exceptions import ValidationError
 from django.db.utils import IntegrityError
 from django.utils import timezone
 from le_utils.constants import content_kinds
@@ -669,6 +670,15 @@ def test_duration_check_constraint__not_media(self):
                 duration=10,
             )
 
+    def test_invalid_file_format(self):
+        channel = testdata.channel()
+        with self.assertRaises(ValidationError, msg="Invalid file_format"):
+            File.objects.create(
+                contentnode=create_contentnode(channel.main_tree_id),
+                preset_id=format_presets.EPUB,
+                file_format_id='pptx',
+            )
+
 
 class AssessmentItemFilePermissionTestCase(PermissionQuerysetTestCase):
     @property

contentcuration/contentcuration/tests/viewsets/test_file.py

@@ -413,6 +413,22 @@ def test_duration_invalid(self):
 
         self.assertEqual(response.status_code, 400)
 
+    def test_invalid_file_format_upload(self):
+        self.client.force_authenticate(user=self.user)
+        file = {
+            "size": 1000,
+            "checksum": uuid.uuid4().hex,
+            "name": "le_studio",
+            "file_format": "ppx",
+            "preset": format_presets.AUDIO,
+            "duration": 10.123
+        }
+        response = self.client.post(
+            reverse("file-upload-url"), file, format="json",
+        )
+
+        self.assertEqual(response.status_code, 400)
+
     def test_insufficient_storage(self):
         self.file["size"] = 100000000000000
 

contentcuration/contentcuration/viewsets/file.py

@@ -3,6 +3,7 @@
 
 from django.core.exceptions import PermissionDenied
 from django.http import HttpResponseBadRequest
+from le_utils.constants import file_formats
 from rest_framework.decorators import action
 from rest_framework.permissions import IsAuthenticated
 from rest_framework.response import Response
@@ -164,6 +165,9 @@ def upload_url(self, request):
             filepath, checksum_base64, 600, content_length=size
         )
 
+        if file_format not in dict(file_formats.choices):
+            return HttpResponseBadRequest("Invalid file_format!")
+
         file = File(
             file_size=size,
             checksum=checksum,