starred/netbox
1
0
Fork 0
mirror of https://github.com/netbox-community/netbox.git synced 2026-04-02 15:37:18 +02:00
Code Issues 273 Packages Projects Releases 10 Wiki Activity

change to scripts/upload api

Browse Source
This commit is contained in:
Arthur
2026-04-01 10:24:11 -07:00
parent 280bca81ef
commit 7f8dfa4427
5 changed files with 28 additions and 153 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
docs/customization/custom-scripts.md
View File

@@ -386,25 +386,14 @@ A complete date & time. Returns a `datetime.datetime` object.
## Uploading Scripts via the API
Script modules can be uploaded to NetBox via the REST API by sending a `multipart/form-data` POST request to `/api/extras/script-upload/`. The caller must have the `extras.add_scriptmodule` and `core.add_managedfile` permissions.
Script modules can be uploaded to NetBox via the REST API by sending a `multipart/form-data` POST request to `/api/extras/scripts/upload/`. The caller must have the `extras.add_scriptmodule` and `core.add_managedfile` permissions.
```no-highlight
curl -X POST \
-H "Authorization: Token $TOKEN" \
-H "Accept: application/json; indent=4" \
-F "upload_file=@/path/to/myscript.py" \
http://netbox/api/extras/script-upload/
```
Alternatively, a script module can be linked to an existing data source and data file instead of uploading a file directly:
```no-highlight
curl -X POST \
-H "Authorization: Token $TOKEN" \
-H "Content-Type: application/json" \
-H "Accept: application/json; indent=4" \
http://netbox/api/extras/script-upload/ \
--data '{"data_source": 1, "data_file": 42}'
-F "file=@/path/to/myscript.py" \
http://netbox/api/extras/scripts/upload/
```
## Running Custom Scripts

74
netbox/extras/api/serializers_/scripts.py
View File

@@ -1,13 +1,9 @@
import os
from django.core.files.storage import storages
from django.db import IntegrityError
from django.utils import timezone
from django.utils.translation import gettext_lazy as _
from drf_spectacular.utils import extend_schema_field
from rest_framework import serializers
from core.api.serializers_.data import DataFileSerializer, DataSourceSerializer
from core.api.serializers_.jobs import JobSerializer
from core.choices import ManagedFileRootPathChoices
from extras.models import Script, ScriptModule
@@ -23,92 +19,42 @@ __all__ = (
class ScriptModuleSerializer(ValidatedModelSerializer):
data_source = DataSourceSerializer(nested=True, required=False, allow_null=True)
data_file = DataFileSerializer(nested=True, required=False, allow_null=True)
upload_file = serializers.FileField(write_only=True, required=False, allow_null=True)
file = serializers.FileField(write_only=True)
file_path = serializers.CharField(read_only=True)
class Meta:
model = ScriptModule
fields = [
'id', 'display', 'file_path', 'upload_file',
'data_source', 'data_file', 'auto_sync_enabled',
'created', 'last_updated',
]
fields = ['id', 'display', 'file_path', 'file', 'created', 'last_updated']
brief_fields = ('id', 'display')
def validate(self, data):
upload_file = data.pop('upload_file', None)
# For multipart requests, nested serializer fields (data_source, data_file) are
# silently dropped by DRF's HTML parser, so also check initial_data for raw values.
has_data_file = data.get('data_file') or self.initial_data.get('data_file')
has_data_source = data.get('data_source') or self.initial_data.get('data_source')
if upload_file and (has_data_file or has_data_source):
raise serializers.ValidationError(
_("Cannot upload a file and sync from a data source.")
)
if has_data_source and not has_data_file:
raise serializers.ValidationError(
_("A data file must be specified when syncing from a data source.")
)
if self.instance is None and not upload_file and not has_data_file:
raise serializers.ValidationError(
_("Must upload a file or select a data file to sync.")
)
# ScriptModule.save() sets file_root; inject it here so full_clean() succeeds
# ScriptModule.save() sets file_root; inject it here so full_clean() succeeds.
# Pop 'file' before model instantiation — ScriptModule has no such field.
file = data.pop('file', None)
data['file_root'] = ManagedFileRootPathChoices.SCRIPTS
data = super().validate(data)
data.pop('file_root', None)
if upload_file is not None:
data['upload_file'] = upload_file
if file is not None:
data['file'] = file
return data
def _save_upload(self, upload_file, validated_data):
def create(self, validated_data):
upload_file = validated_data.pop('file')
storage = storages.create_storage(storages.backends["scripts"])
validated_data['file_path'] = storage.save(upload_file.name, upload_file)
def _sync_data_file(self, data_file, validated_data):
"""
Pre-populate file_path/data_path and write the file to disk before create(),
so that save() → sync_classes() fires once with the correct file_path — matching
the UI path where full_clean() sets these fields on the actual instance before save().
"""
file_path = os.path.basename(data_file.path)
validated_data['data_path'] = data_file.path
validated_data['file_path'] = file_path
validated_data['data_synced'] = timezone.now()
storage = storages.create_storage(storages.backends["scripts"])
with storage.open(file_path, 'wb+') as f:
f.write(data_file.data)
def create(self, validated_data):
upload_file = validated_data.pop('upload_file', None)
if upload_file:
self._save_upload(upload_file, validated_data)
elif data_file := validated_data.get('data_file'):
self._sync_data_file(data_file, validated_data)
created = False
try:
instance = super().create(validated_data)
created = True
return instance
except IntegrityError:
# ManagedFile has a single unique constraint: (file_root, file_path), so an
# IntegrityError here always means a duplicate file name regardless of which
# path (upload or data_file sync) set validated_data['file_path'].
raise serializers.ValidationError(
_("A script module with this file name already exists.")
)
finally:
# On any failure, remove the file written to disk so no orphans are left behind.
# Uses best-effort deletion (ignores errors) to avoid masking the original exception.
if not created and (file_path := validated_data.get('file_path')):
try:
storages.create_storage(storages.backends["scripts"]).delete(file_path)
storage.delete(file_path)
except Exception:
pass

2
netbox/extras/api/urls.py
View File

@@ -26,7 +26,7 @@ router.register('journal-entries', views.JournalEntryViewSet)
router.register('config-contexts', views.ConfigContextViewSet)
router.register('config-context-profiles', views.ConfigContextProfileViewSet)
router.register('config-templates', views.ConfigTemplateViewSet)
router.register('script-upload', views.ScriptModuleViewSet)
router.register('scripts/upload', views.ScriptModuleViewSet)
router.register('scripts', views.ScriptViewSet, basename='script')
app_name = 'extras-api'

5
netbox/extras/api/views.py
View File

@@ -1,4 +1,5 @@
from django.http import Http404
from drf_spectacular.utils import extend_schema, extend_schema_view
from django.shortcuts import get_object_or_404
from django_rq.queues import get_connection
from rest_framework import status
@@ -268,6 +269,10 @@ class ScriptModuleViewSet(CreateModelMixin, BaseViewSet):
serializer_class = serializers.ScriptModuleSerializer
@extend_schema_view(
update=extend_schema(request=serializers.ScriptInputSerializer),
partial_update=extend_schema(request=serializers.ScriptInputSerializer),
)
class ScriptViewSet(ModelViewSet):
permission_classes = [IsAuthenticatedOrLoginNotRequired]
queryset = Script.objects.all()

83
netbox/extras/tests/test_api.py
View File

@@ -1390,42 +1390,23 @@ class NotificationTest(APIViewTestCases.APIViewTestCase):
class ScriptModuleTest(APITestCase):
"""
Tests for the /api/extras/script-modules/ endpoint.
Tests for the POST /api/extras/scripts/upload/ endpoint.
ScriptModule is a proxy of core.ManagedFile (a different app) so the standard
APIViewTestCases mixins cannot be used directly:
- GraphQLTestCase: ScriptModule has no GraphQL type.
- CreateObjectViewTestCase: requires multipart file upload, not plain JSON create_data.
- Get/List/Update/DeleteObjectViewTestCase: the mixin's ObjectPermission setup resolves
ScriptModule to ManagedFile's ContentType (core.managedfile), producing a
core.change_managedfile grant. But TokenPermissions checks extras.change_scriptmodule,
causing a 403 despite the ObjectPermission existing.
All tests therefore use add_permissions() with explicit Django model-level permissions.
APIViewTestCases mixins cannot be used directly. All tests use add_permissions()
with explicit Django model-level permissions.
"""
@classmethod
def setUpTestData(cls):
cls.data_source = DataSource.objects.create(name='Test Source', type='local', source_url='/tmp')
script_content = b"from extras.scripts import Script\nclass TestScript(Script):\n pass\n"
cls.data_file = DataFile.objects.create(
source=cls.data_source,
path='test_datasource.py',
last_updated=now(),
size=len(script_content),
hash=hashlib.sha256(script_content).hexdigest(),
data=script_content,
)
def setUp(self):
super().setUp()
self.url_list = reverse('extras-api:scriptmodule-list') # /api/extras/script-upload/
self.url = reverse('extras-api:scriptmodule-list') # /api/extras/scripts/upload/
def test_upload_script_module_without_permission(self):
script_content = b"from extras.scripts import Script\nclass TestScript(Script):\n pass\n"
upload_file = SimpleUploadedFile('test_upload.py', script_content, content_type='text/plain')
response = self.client.post(
self.url_list,
{'upload_file': upload_file},
self.url,
{'file': upload_file},
format='multipart',
**self.header,
)
@@ -1442,8 +1423,8 @@ class ScriptModuleTest(APITestCase):
mock_storages.create_storage.return_value = mock_storage
mock_storages.backends = {'scripts': {}}
response = self.client.post(
self.url_list,
{'upload_file': upload_file},
self.url,
{'file': upload_file},
format='multipart',
**self.header,
)
@@ -1452,53 +1433,7 @@ class ScriptModuleTest(APITestCase):
mock_storage.save.assert_called_once()
self.assertTrue(ScriptModule.objects.filter(file_path='test_upload.py').exists())
def test_upload_with_data_source_fails(self):
"""Supplying both upload_file and data_source must be rejected."""
self.add_permissions('extras.add_scriptmodule', 'core.add_managedfile')
script_content = b"from extras.scripts import Script\nclass TestScript(Script):\n pass\n"
upload_file = SimpleUploadedFile('test_upload.py', script_content, content_type='text/plain')
response = self.client.post(
self.url_list,
{'upload_file': upload_file, 'data_source': self.data_source.pk},
format='multipart',
**self.header,
)
self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
def test_data_source_without_data_file_fails(self):
"""data_source alone (without data_file) must be rejected."""
self.add_permissions('extras.add_scriptmodule', 'core.add_managedfile')
response = self.client.post(
self.url_list,
{'data_source': self.data_source.pk},
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_list, {}, format='json', **self.header)
response = self.client.post(self.url, {}, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
def test_create_script_module_from_data_file(self):
"""POST with data_source + data_file (JSON) creates a ScriptModule with the correct file_path."""
self.add_permissions('extras.add_scriptmodule', 'core.add_managedfile')
mock_storage = MagicMock()
# Patch storages in both the serializer (for _sync_data_file) and the model
# (for ManagedFile.sync_data(), which is called by SyncedDataMixin.clean() during
# ValidatedModelSerializer.validate() → full_clean()).
with patch('extras.api.serializers_.scripts.storages') as mock_ser_storages, \
patch('core.models.files.storages') as mock_model_storages:
for m in (mock_ser_storages, mock_model_storages):
m.create_storage.return_value = mock_storage
m.backends = {'scripts': {}}
response = self.client.post(
self.url_list,
{'data_source': self.data_source.pk, 'data_file': self.data_file.pk},
format='json',
**self.header,
)
self.assertHttpStatus(response, status.HTTP_201_CREATED)
self.assertEqual(response.data['file_path'], 'test_datasource.py')
self.assertTrue(ScriptModule.objects.filter(file_path='test_datasource.py').exists())