cleanup

netbox/extras/api/serializers_/scripts.py

@@ -39,18 +39,20 @@ class ScriptModuleSerializer(ValidatedModelSerializer):
         return data
 
     def create(self, validated_data):
-        upload_file = validated_data.pop('file')
+        file = validated_data.pop('file')
         storage = storages.create_storage(storages.backends["scripts"])
-        validated_data['file_path'] = storage.save(upload_file.name, upload_file)
+        validated_data['file_path'] = storage.save(file.name, file)
         created = False
         try:
             instance = super().create(validated_data)
             created = True
             return instance
-        except IntegrityError:
-            raise serializers.ValidationError(
-                _("A script module with this file name already exists.")
-            )
+        except IntegrityError as e:
+            if 'file_path' in str(e):
+                raise serializers.ValidationError(
+                    _("A script module with this file name already exists.")
+                )
+            raise
         finally:
             if not created and (file_path := validated_data.get('file_path')):
                 try:

netbox/extras/api/views.py

@@ -1,6 +1,6 @@
 from django.http import Http404
-from drf_spectacular.utils import extend_schema, extend_schema_view
 from django.shortcuts import get_object_or_404
+from drf_spectacular.utils import extend_schema, extend_schema_view
 from django_rq.queues import get_connection
 from rest_framework import status
 from rest_framework.decorators import action

netbox/extras/tests/test_api.py

@@ -1433,6 +1433,27 @@ class ScriptModuleTest(APITestCase):
         mock_storage.save.assert_called_once()
         self.assertTrue(ScriptModule.objects.filter(file_path='test_upload.py').exists())
 
+    def test_upload_script_module_duplicate_fails(self):
+        self.add_permissions('extras.add_scriptmodule', 'core.add_managedfile')
+        script_content = b"from extras.scripts import Script\nclass TestScript(Script):\n    pass\n"
+        mock_storage = MagicMock()
+        mock_storage.save.return_value = 'test_upload.py'
+        with patch('extras.api.serializers_.scripts.storages') as mock_storages:
+            mock_storages.create_storage.return_value = mock_storage
+            mock_storages.backends = {'scripts': {}}
+            # First upload succeeds
+            upload_file = SimpleUploadedFile('test_upload.py', script_content, content_type='text/plain')
+            self.client.post(self.url, {'file': upload_file}, format='multipart', **self.header)
+            # Second upload with same name should fail
+            upload_file = SimpleUploadedFile('test_upload.py', script_content, content_type='text/plain')
+            response = self.client.post(
+                self.url,
+                {'file': upload_file},
+                format='multipart',
+                **self.header,
+            )
+        self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
+
     def test_upload_script_module_without_file_fails(self):
         self.add_permissions('extras.add_scriptmodule', 'core.add_managedfile')
         response = self.client.post(self.url, {}, format='json', **self.header)