Ensure that file_format and preset match during file_upload requests.

Author: rtibbles

contentcuration/contentcuration/tests/viewsets/test_file.py

@@ -438,6 +438,7 @@ def test_duration_invalid(self):
     def test_duration_missing(self):
         del self.file["duration"]
         self.file["file_format"] = file_formats.EPUB
+        self.file["preset"] = format_presets.EPUB
 
         self.client.force_authenticate(user=self.user)
         response = self.client.post(
@@ -461,9 +462,23 @@ def test_duration_missing_but_required(self):
 
         self.assertEqual(response.status_code, 400)
 
+    def test_duration_present_but_not_allowed(self):
+        self.file["file_format"] = file_formats.EPUB
+        self.file["preset"] = format_presets.DOCUMENT
+
+        self.client.force_authenticate(user=self.user)
+        response = self.client.post(
+            reverse("file-upload-url"),
+            self.file,
+            format="json",
+        )
+
+        self.assertEqual(response.status_code, 400)
+
     def test_duration_null(self):
         self.file["duration"] = None
         self.file["file_format"] = file_formats.EPUB
+        self.file["preset"] = format_presets.EPUB
 
         self.client.force_authenticate(user=self.user)
         response = self.client.post(
@@ -517,6 +532,18 @@ def test_invalid_preset_upload(self):
         response = self.client.post(reverse("file-upload-url"), file, format="json")
         self.assertEqual(response.status_code, 400)
 
+    def test_mismatched_preset_upload(self):
+        self.file["file_format"] = file_formats.EPUB
+
+        self.client.force_authenticate(user=self.user)
+        response = self.client.post(
+            reverse("file-upload-url"),
+            self.file,
+            format="json",
+        )
+
+        self.assertEqual(response.status_code, 400)
+
     def test_insufficient_storage(self):
         self.file["size"] = 100000000000000
 

contentcuration/contentcuration/viewsets/file.py

@@ -32,6 +32,9 @@
 from contentcuration.viewsets.sync.utils import generate_update_event
 
 
+PRESET_LOOKUP = {p.id: p for p in format_presets.PRESETLIST}
+
+
 class StrictFloatField(serializers.FloatField):
     def to_internal_value(self, data):
         # If data is a string, reject it even if it represents a number.
@@ -81,6 +84,11 @@ def validate(self, attrs):
                 raise serializers.ValidationError(
                     "Duration is required for audio/video files"
                 )
+        preset_obj = PRESET_LOOKUP[attrs["preset"]]
+        if attrs["file_format"] not in preset_obj.allowed_formats:
+            raise serializers.ValidationError(
+                f"File format {attrs['file_format']} is not an allowed format for this preset {attrs['preset']}"
+            )
         return attrs