RomanNum3ral
/
netbox
mirror of https://github.com/netbox-community/netbox.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'main' into 23134-circuit-changelog

This commit is contained in:
Arthur 2026-09-08 10:49:08 -07:00
parent 8ec91108f2 7b56158d47
commit 8e409d0fff
17 changed files with 535 additions and 104 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -301,6 +301,9 @@ All custom script variables support the following default options:
* `required` - Indicates whether the field is mandatory (all fields are required by default)
* `widget` - The class of form widget to use (see the [Django documentation](https://docs.djangoproject.com/en/stable/ref/forms/widgets/))
!!! warning "Reserved variable names"
The names `_commit`, `_schedule_at`, `_interval`, and `_notifications` are reserved for the execution parameters which NetBox renders alongside a script's own fields. A variable declared with one of these names shadows its execution parameter, and its value is not passed to `run()`. Choose a different name.
### StringVar
Stores a string of characters (i.e. text). Options include:
@ -546,6 +549,9 @@ http://netbox/api/extras/scripts/example.MyReport/ \
Optionally `schedule_at` can be passed in the form data with a datetime string to schedule a script at the specified date and time.
!!! note
Script input submitted through the REST API is validated against the variables declared by the script. Missing required variables or invalid values result in an HTTP 400 response, and undeclared keys are discarded rather than passed to `run()`. Existing API clients that relied on the previous pass-through behavior may need to update their requests. Scripts declaring a `FileVar` must be run via a `multipart/form-data` request, passing `data` as a JSON string alongside the uploaded file.
### Via the CLI
Scripts can be run on the CLI by invoking the management command:

2
netbox/core/api/serializers_/data.py
View File

@ -26,7 +26,7 @@ class DataSourceSerializer(PrimaryModelSerializer):
model = DataSource
fields = [
'id', 'url', 'display_url', 'display', 'name', 'type', 'source_url', 'enabled', 'status', 'description',
'sync_interval', 'parameters', 'ignore_rules', 'owner', 'comments', 'custom_fields', 'created',
'sync_interval', 'parameters', 'ignore_rules', 'owner', 'comments', 'tags', 'custom_fields', 'created',
'last_updated', 'last_synced', 'file_count',
]
brief_fields = ('id', 'url', 'display', 'name', 'description')

2
netbox/core/tests/query_counts.json
View File

@ -1,7 +1,7 @@
{
"datafile:api_list_objects": 10,
"datafile:list_objects_with_permission": 17,
"datasource:api_list_objects": 11,
"datasource:api_list_objects": 12,
"datasource:list_objects_with_permission": 17,
"job:api_list_objects": 12,
"job:list_objects_with_permission": 19

53
netbox/core/tests/test_api.py
View File

@ -12,7 +12,7 @@ from rq.registry import FailedJobRegistry, StartedJobRegistry
from users.constants import TOKEN_PREFIX
from users.models import Token
from utilities.testing import APITestCase, APIViewTestCases, GraphQLQueryTest, TestCase
from utilities.testing import APITestCase, APIViewTestCases, GraphQLQueryTest, TestCase, create_tags
from utilities.testing.mixins import RQQueueTestMixin
from utilities.testing.utils import disable_logging
@ -100,6 +100,57 @@ class DataSourceTestCase(APIViewTestCases.APIViewTestCase):
},
]
def test_tags_in_representation(self):
"""Assigned tags are rendered in the detail representation."""
data_source = DataSource.objects.first()
data_source.tags.set(create_tags('Alpha'))
self.add_permissions('core.view_datasource')
response = self.client.get(self._get_detail_url(data_source), **self.header)
self.assertHttpStatus(response, status.HTTP_200_OK)
self.assertIn('tags', response.data)
self.assertEqual([tag['slug'] for tag in response.data['tags']], ['alpha'])
def test_create_with_tags(self):
"""Tags supplied on creation are assigned to the new data source."""
create_tags('Alpha')
self.add_permissions('core.add_datasource', 'extras.view_tag')
data = {
'name': 'Data Source 7',
'type': 'git',
'source_url': 'https://example.com/git/source7',
'tags': [{'slug': 'alpha'}],
}
response = self.client.post(self._get_list_url(), data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_201_CREATED)
data_source = DataSource.objects.get(pk=response.data['id'])
self.assertEqual(list(data_source.tags.values_list('slug', flat=True)), ['alpha'])
def test_update_tags(self):
"""Tags supplied on update replace the existing assignment."""
data_source = DataSource.objects.first()
tags = create_tags('Alpha', 'Bravo')
data_source.tags.set([tags[0]])
self.add_permissions('core.change_datasource', 'extras.view_tag')
data = {'tags': [{'slug': 'bravo'}]}
response = self.client.patch(self._get_detail_url(data_source), data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_200_OK)
self.assertEqual(list(data_source.tags.values_list('slug', flat=True)), ['bravo'])
def test_clear_tags(self):
"""An empty tag list clears the existing assignment."""
data_source = DataSource.objects.first()
data_source.tags.set(create_tags('Alpha'))
self.add_permissions('core.change_datasource')
data = {'tags': []}
response = self.client.patch(self._get_detail_url(data_source), data, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_200_OK)
self.assertEqual(list(data_source.tags.values_list('slug', flat=True)), [])
def assert_only_source_1(self, data):
"""The JSON lookup returns exactly the source carrying the matching value."""
ids = sorted(result['id'] for result in data['data_source_list'])

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

@ -188,6 +188,16 @@ class ScriptInputSerializer(serializers.Serializer):
if script and script.python_class:
self.fields['notifications'].default = script.python_class.notifications_default
def validate_data(self, value):
"""
Validates that the script input is an object mapping variable names to values.
"""
if not isinstance(value, dict):
raise serializers.ValidationError(
_('Invalid data payload; expected an object mapping variable names to values.')
)
return value
def validate_schedule_at(self, value):
"""
Validates the specified schedule time for a script execution.

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

@ -1,9 +1,9 @@
from django.core.exceptions import NON_FIELD_ERRORS
from django.core.exceptions import ValidationError as DjangoValidationError
from django.http import Http404
from django.shortcuts import get_object_or_404
from django.utils.translation import gettext_lazy as _
from drf_spectacular.utils import OpenApiResponse, OpenApiTypes, extend_schema
from rest_framework import status
from rest_framework.decorators import action
from rest_framework.exceptions import PermissionDenied, ValidationError
from rest_framework.generics import RetrieveUpdateDestroyAPIView
@ -16,6 +16,7 @@ from core.choices import ManagedFileRootPathChoices
from extras import filtersets
from extras.jobs import ScriptJob
from extras.models import *
from extras.scripts import EXEC_PARAM_FIELDS, prepare_script_form
from netbox.api.authentication import IsAuthenticatedOrLoginNotRequired, TokenWritePermission
from netbox.api.features import SyncedDataMixin
from netbox.api.metadata import ContentTypeMetadata
@ -404,29 +405,56 @@ class ScriptViewSet(ListModelMixin, RetrieveModelMixin, BaseViewSet):
if not any_workers_for_queue('default'):
raise RQWorkerNotRunningException()
if input_serializer.is_valid():
try:
ScriptJob.enqueue(
instance=script,
user=request.user,
data=input_serializer.data['data'],
request=copy_safe_request(request),
commit=input_serializer.data['commit'],
job_timeout=script.python_class.job_timeout,
schedule_at=input_serializer.validated_data.get('schedule_at'),
interval=input_serializer.validated_data.get('interval'),
notifications=input_serializer.validated_data.get('notifications'),
)
except DjangoValidationError as e:
# The script's execution configuration is invalid (see #22872). Surface it as a 400 rather than
# allowing the exception to bubble up as an HTTP 500. These are script-level config errors, not
# request-field errors, so report them under the non-field "detail" key.
raise ValidationError({'detail': e.messages}) from e
serializer = serializers.ScriptDetailSerializer(script, context={'request': request})
input_serializer.is_valid(raise_exception=True)
return Response(serializer.data)
validated = input_serializer.validated_data
return Response(input_serializer.errors, status=status.HTTP_400_BAD_REQUEST)
payload = validated['data']
# Guaranteed non-None by the is_executable check above
script_class = script.python_class
script_instance = script_class()
form = prepare_script_form(script_instance, payload, files=request.FILES)
if not form.is_valid():
# Exec params are validated separately via ScriptInputSerializer. Excluded by name
# rather than by '_' prefix, which would also strip Django's NON_FIELD_ERRORS
# key ('__all__').
errors = {k: v for k, v in form.errors.items() if k not in EXEC_PARAM_FIELDS}
if not errors:
# Every error was on an exec-param field, which a client can bind by naming one
# in 'data' (e.g. {"_interval": "abc"}). NON_FIELD_ERRORS is never among them --
# the filter above retains '__all__' -- so there is nothing to re-surface here;
# report a generic message rather than an empty body.
errors = {NON_FIELD_ERRORS: [_('Invalid script input.')]}
# Nest under 'data' so script-variable errors can't collide with the
# serializer's own top-level fields (commit, schedule_at, interval, ...).
raise ValidationError({'data': errors})
data = form.cleaned_data.copy()
for k in EXEC_PARAM_FIELDS:
data.pop(k, None)
try:
ScriptJob.enqueue(
instance=script,
user=request.user,
data=data,
request=copy_safe_request(request),
commit=validated.get('commit'),
job_timeout=script_class.job_timeout,
schedule_at=validated.get('schedule_at'),
interval=validated.get('interval'),
notifications=validated.get('notifications'),
)
except DjangoValidationError as e:
# The script's execution configuration is invalid (see #22872). Surface it as a 400 rather than
# allowing the exception to bubble up as an HTTP 500. These are script-level config errors, not
# request-field errors, so report them under the non-field "detail" key.
raise ValidationError({'detail': e.messages}) from e
serializer = serializers.ScriptDetailSerializer(script, context={'request': request})
return Response(serializer.data)
#

14
netbox/extras/management/commands/runscript.py
View File

@ -7,7 +7,7 @@ from django.core.exceptions import ValidationError
from django.core.management.base import BaseCommand, CommandError
from extras.jobs import ScriptJob
from extras.scripts import get_module_and_script
from extras.scripts import EXEC_PARAM_FIELDS, get_module_and_script
from users.models import User
from utilities.request import NetBoxFakeRequest
@ -82,11 +82,11 @@ class Command(BaseCommand):
logger.error(f'\t{field}: {error.get("message")}')
raise CommandError()
# Remove extra fields from ScriptForm before passing data to script
form.cleaned_data.pop('_schedule_at')
form.cleaned_data.pop('_interval')
form.cleaned_data.pop('_commit')
notifications = form.cleaned_data.pop('_notifications')
# Remove exec-parameter fields from ScriptForm before passing data to the script
cleaned_data = form.cleaned_data.copy()
notifications = cleaned_data.pop('_notifications')
for key in EXEC_PARAM_FIELDS:
cleaned_data.pop(key, None)
# Execute the script.
try:
@ -94,7 +94,7 @@ class Command(BaseCommand):
instance=script_obj,
user=user,
immediate=True,
data=form.cleaned_data,
data=cleaned_data,
notifications=notifications,
request=NetBoxFakeRequest({
'META': {},

1
netbox/extras/models/customfields.py
View File

@ -1091,6 +1091,7 @@ class CustomField(CloningMixin, ExportTemplatesMixin, OwnerMixin, ChangeLoggedMo
# Multiselect
elif self.type == CustomFieldTypeChoices.TYPE_MULTISELECT:
# Do not pin lookup_expr: FILTER_ARRAY_BASED_LOOKUP_MAP preserves the class default under negation
filter_class = filters.MultiValueArrayFilter
# Object

29
netbox/extras/scripts.py
View File

@ -46,6 +46,11 @@ __all__ = (
'get_module_and_script',
)
# Internal ScriptForm fields used to carry execution parameters (see ScriptForm in
# extras/forms/scripts.py). These are validated/sourced separately from the script's own
# declared variables and must never be treated as script data or surfaced as script errors.
EXEC_PARAM_FIELDS = ('_commit', '_schedule_at', '_interval', '_notifications')
# Sentinel distinguishing "argument not supplied" from an explicit None in validate_meta().
_UNSET = object()
@ -706,3 +711,27 @@ def get_module_and_script(module_name, script_name):
module = ScriptModule.objects.get(file_path=f'{module_name}.py')
script = module.scripts.get(name=script_name)
return module, script
def prepare_script_form(script_instance, data, files=None):
"""
Return a bound ScriptForm for the given Script instance, back-filling the declared
`default` of any variable omitted from `data`.
`data` is copied rather than coerced to a plain dict, so a QueryDict retains the
multi-value semantics a MultiObjectVar's multi-select field depends on.
"""
data = data.copy() if data is not None else {}
for name, var in script_instance._get_vars().items():
if name in data:
continue
if (initial := var.field_attrs.get('initial')) is None:
continue
if isinstance(initial, (list, tuple)) and hasattr(data, 'setlist'):
# Assigning a list to a QueryDict stores it as a single nested value, which a
# multi-select widget reads back as one bogus choice. Set the values individually
# so a MultiChoiceVar/MultiObjectVar default binds as it does for a plain dict.
data.setlist(name, list(initial))
else:
data[name] = initial
return script_instance.as_form(data=data, files=files)

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

@ -13,14 +13,14 @@ from django.urls import reverse
from django.utils.timezone import make_aware, now
from rest_framework import status
from core.choices import ManagedFileRootPathChoices
from core.choices import JobNotificationChoices, ManagedFileRootPathChoices
from core.events import *
from core.models import DataFile, DataSource, Job, ObjectType
from dcim.models import Device, DeviceRole, DeviceType, Location, Manufacturer, Rack, RackRole, Site
from extras.api.serializers import EventRuleSerializer
from extras.choices import *
from extras.models import *
from extras.scripts import BooleanVar, IntegerVar, StringVar
from extras.scripts import BooleanVar, IntegerVar, MultiObjectVar, ObjectVar, StringVar
from extras.scripts import Script as PythonClass
from netbox.event_rules import EventRuleAction, register_event_rule_action
from netbox.registry import registry
@ -1890,6 +1890,208 @@ class ScriptTestCase(APITestCase):
self.assertHttpStatus(response, status.HTTP_404_NOT_FOUND)
class ScriptRunExecutionTestCase(APITestCase):
"""
Exercises ScriptViewSet.run() end-to-end: request -> serializer -> form -> enqueue.
"""
class TestScriptClass(PythonClass):
class Meta:
name = 'Test run script'
site = ObjectVar(model=Site)
sites = MultiObjectVar(model=Site, required=False)
label = StringVar(default='hello')
flag = BooleanVar(default=True)
def run(self, data, commit=True):
return 'ok'
@classmethod
def setUpTestData(cls):
cls.sites = [
Site.objects.create(name=f'Test Site {i}', slug=f'test-site-{i}') for i in range(1, 3)
]
with patch.object(ScriptModule, 'sync_classes'):
module = ScriptModule.objects.create(
file_root=ManagedFileRootPathChoices.SCRIPTS,
file_path='run_script.py',
)
script = Script.objects.create(
module=module,
name='Test run script',
is_executable=True,
)
cls.url = reverse('extras-api:script-detail', kwargs={'pk': script.pk})
def setUp(self):
super().setUp()
self.add_permissions('extras.run_script')
# Monkey-patch the Script model to return our TestScriptClass above, restoring
# the real property afterwards so later tests aren't left with our stub.
python_class_patch = patch.object(Script, 'python_class', new=self.TestScriptClass)
python_class_patch.start()
self.addCleanup(python_class_patch.stop)
# The script-run endpoint gates on a live RQ worker. Tests run without one, so
# bypass the check to exercise validation and the enqueue path.
worker_patch = patch('extras.api.views.any_workers_for_queue', return_value=True)
worker_patch.start()
self.addCleanup(worker_patch.stop)
@patch('extras.jobs.ScriptJob.enqueue')
def test_run_forwards_commit_value(self, mock_enqueue):
for commit_value in (True, False):
with self.subTest(commit=commit_value):
mock_enqueue.reset_mock()
payload = {'data': {'site': self.sites[0].pk}, 'commit': commit_value}
response = self.client.post(self.url, payload, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_200_OK)
mock_enqueue.assert_called_once()
self.assertIs(mock_enqueue.call_args.kwargs['commit'], commit_value)
@patch('extras.jobs.ScriptJob.enqueue')
def test_run_forwards_notifications_value(self, mock_enqueue):
# Regression: ScriptForm.clean() overwrites an empty '_notifications' with the
# field's own initial, so a client-supplied value never reached ScriptJob.enqueue.
payload = {
'data': {'site': self.sites[0].pk},
'commit': True,
'notifications': JobNotificationChoices.NOTIFICATION_NEVER,
}
response = self.client.post(self.url, payload, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_200_OK)
self.assertEqual(
mock_enqueue.call_args.kwargs['notifications'],
JobNotificationChoices.NOTIFICATION_NEVER,
)
@patch('extras.jobs.ScriptJob.enqueue')
def test_run_forwards_schedule_at_and_interval(self, mock_enqueue):
# Regression: schedule_at/interval were likewise read from the form (always
# absent there) instead of the validated request
schedule_at = now() + datetime.timedelta(hours=1)
payload = {
'data': {'site': self.sites[0].pk},
'commit': True,
'schedule_at': schedule_at,
'interval': 60,
}
response = self.client.post(self.url, payload, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_200_OK)
kwargs = mock_enqueue.call_args.kwargs
self.assertEqual(kwargs['schedule_at'], schedule_at)
self.assertEqual(kwargs['interval'], 60)
@patch('extras.jobs.ScriptJob.enqueue')
def test_run_converts_objectvar_and_multiobjectvar_ids(self, mock_enqueue):
payload = {
'data': {
'site': self.sites[0].pk,
'sites': [site.pk for site in self.sites],
},
'commit': True,
}
response = self.client.post(self.url, payload, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_200_OK)
data = mock_enqueue.call_args.kwargs['data']
self.assertEqual(data['site'], self.sites[0])
self.assertEqual(
set(data['sites'].values_list('pk', flat=True)),
{site.pk for site in self.sites},
)
@patch('extras.jobs.ScriptJob.enqueue')
def test_run_backfills_default_for_omitted_required_var(self, mock_enqueue):
# Regression: required vars declaring `default=` were not back-filled before
# binding the form on the API path (unlike the UI path), so they 400'd even
# though the client legitimately omitted them.
payload = {'data': {'site': self.sites[0].pk}, 'commit': True} # 'label' omitted
response = self.client.post(self.url, payload, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_200_OK)
self.assertEqual(mock_enqueue.call_args.kwargs['data']['label'], 'hello')
def test_run_rejects_non_dict_payload(self):
payload = {'data': 'not-a-dict', 'commit': True}
response = self.client.post(self.url, payload, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
@patch('extras.jobs.ScriptJob.enqueue')
def test_run_rejects_nonexistent_object_id(self, mock_enqueue):
# This is the primary new failure mode introduced by converting raw IDs to model
# instances: a PK that doesn't resolve must 400 cleanly, not enqueue a broken job
# or raise an unhandled DoesNotExist.
nonexistent_pk = Site.objects.order_by('-pk').first().pk + 1000
payload = {'data': {'site': nonexistent_pk}, 'commit': True}
response = self.client.post(self.url, payload, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
mock_enqueue.assert_not_called()
@patch('extras.jobs.ScriptJob.enqueue')
def test_run_ignores_undeclared_keys(self, mock_enqueue):
# Binding 'data' to the script's form means keys which don't correspond to a declared
# variable are dropped rather than forwarded to run(). This is the contract documented
# under "Running Custom Scripts > Via the API"; pin it so it can't regress silently.
payload = {
'data': {'site': self.sites[0].pk, 'bogus': 'ignored', 'id': 99},
'commit': True,
}
response = self.client.post(self.url, payload, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_200_OK)
data = mock_enqueue.call_args.kwargs['data']
self.assertEqual(data['site'], self.sites[0])
self.assertNotIn('bogus', data)
self.assertNotIn('id', data)
@patch('extras.jobs.ScriptJob.enqueue')
def test_run_rejects_omitted_required_var(self, mock_enqueue):
# 'site' is required and declares no default, so unlike 'label' it cannot be
# back-filled: omitting it must 400 rather than enqueue a job that fails at runtime.
payload = {'data': {'label': 'hi'}, 'commit': True}
response = self.client.post(self.url, payload, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
self.assertIn('site', response.data['data'])
mock_enqueue.assert_not_called()
@patch('extras.jobs.ScriptJob.enqueue')
def test_run_resolves_booleanvar_default_and_explicit_values(self, mock_enqueue):
# BooleanVar renders as a checkbox, and CheckboxInput reads a missing key as False.
# An omitted BooleanVar must therefore pick up its declared default, while an
# explicitly supplied False must not be overwritten by that default.
for case, supplied, expected in (
('omitted', {}, True),
('explicit False', {'flag': False}, False),
('explicit True', {'flag': True}, True),
):
with self.subTest(case=case):
mock_enqueue.reset_mock()
payload = {'data': {'site': self.sites[0].pk, **supplied}, 'commit': True}
response = self.client.post(self.url, payload, format='json', **self.header)
self.assertHttpStatus(response, status.HTTP_200_OK)
self.assertIs(mock_enqueue.call_args.kwargs['data']['flag'], expected)
class CreatedUpdatedFilterTestCase(APITestCase):
@classmethod

26
netbox/extras/tests/test_customfields.py
View File

@ -2752,6 +2752,7 @@ class CustomFieldModelFilterTestCase(TestCase):
'cf4': None,
'cf6': None,
'cf7': None,
'cf10': None,
})
for filter_name, value in (
@ -2766,6 +2767,7 @@ class CustomFieldModelFilterTestCase(TestCase):
('cf_cf7__nic', 'a'),
('cf_cf7__nisw', 'http://'),
('cf_cf7__niew', '.com'),
('cf_cf10__n', 'A'),
):
with self.subTest(filter_name):
pks = set(
@ -2836,6 +2838,15 @@ class CustomFieldModelFilterTestCase(TestCase):
def test_filter_multiselect(self):
self.assertEqual(self.filterset({'cf_cf10': ['A']}, self.queryset).qs.count(), 1)
self.assertEqual(self.filterset({'cf_cf10': ['A', 'C']}, self.queryset).qs.count(), 2)
# Negation excludes the objects whose array holds the value, not those whose array equals it
self.assertEqual(
set(self.filterset({'cf_cf10__n': ['A']}, self.queryset).qs.values_list('slug', flat=True)),
{'site-2', 'site-3', 'site-4'}
)
self.assertEqual(
set(self.filterset({'cf_cf10__n': ['A', 'C']}, self.queryset).qs.values_list('slug', flat=True)),
{'site-3', 'site-4'}
)
# Matches both the object holding a literal null and the one carrying no key, as `empty` does
self.assertEqual(self.filterset({'cf_cf10': ['null']}, self.queryset).qs.count(), 2)
self.assertEqual(self.filterset({'cf_cf10__empty': True}, self.queryset).qs.count(), 2)
@ -2876,9 +2887,18 @@ def hold_data_lock(custom_field):
cursor.execute('SELECT pg_try_advisory_lock(%s, %s)', lock_key)
if not cursor.fetchone()[0]:
raise RuntimeError(f"Failed to acquire the data lock for {custom_field}")
yield
released = False
try:
yield
finally:
# Closing the connection releases the lock asynchronously, so the next deletion can race it
with connection.cursor() as cursor:
cursor.execute('SELECT pg_advisory_unlock(%s, %s)', lock_key)
released = cursor.fetchone()[0]
# Outside the finally, so a failing body is reported as itself
if not released:
raise RuntimeError(f"Failed to release the data lock for {custom_field}")
finally:
# Closing the session releases any advisory lock held on it
connection.close()
@ -3739,7 +3759,7 @@ class DeferredCustomFieldDataTestCase(TestCase):
# delete() has returned and its own atomic block has exited, but the enclosing transaction
# has yet to commit, so the lock must still be held
with self.assertRaises(RuntimeError):
with self.assertRaisesMessage(RuntimeError, "Failed to acquire the data lock"):
with hold_data_lock(cf):
pass

58
netbox/extras/tests/test_views.py
View File

@ -14,7 +14,7 @@ from core.models import Job, ObjectType
from dcim.models import DeviceType, Manufacturer, Site
from extras.choices import *
from extras.models import *
from extras.scripts import BooleanVar, IntegerVar
from extras.scripts import BooleanVar, IntegerVar, MultiChoiceVar, StringVar
from extras.scripts import Script as PythonClass
from users.models import Group, ObjectPermission, User
from utilities.testing import TestCase, ViewTestCases
@ -1279,6 +1279,62 @@ class ScriptModuleCreateViewTestCase(TestCase):
self.assertEqual(response.context['return_url'], reverse('extras:script_list'))
class ScriptDefaultBackfillTestCase(TestCase):
"""
The UI and the REST API now share prepare_script_form(), so the UI's back-filling of
declared defaults (previously inline in ScriptView.post()) must keep working after
that logic moved into the helper.
"""
user_permissions = ['extras.view_script', 'extras.run_script']
class TestScriptClass(PythonClass):
class Meta:
name = 'Backfill test'
commit_default = False
label = StringVar(default='hello')
flag = BooleanVar(default=True)
picks = MultiChoiceVar(choices=(('a', 'A'), ('b', 'B'), ('c', 'C')), default=['a', 'b'])
def run(self, data, commit):
return 'Complete'
@classmethod
def setUpTestData(cls):
with patch.object(ScriptModule, 'sync_classes'):
module = ScriptModule.objects.create(
file_root=ManagedFileRootPathChoices.SCRIPTS,
file_path='backfill_script.py',
)
cls.script = Script.objects.create(module=module, name='Backfill test', is_executable=True)
def setUp(self):
super().setUp()
python_class_patch = patch.object(Script, 'python_class', new=self.TestScriptClass)
python_class_patch.start()
self.addCleanup(python_class_patch.stop)
@tag('regression')
def test_ui_backfills_declared_defaults(self):
url = reverse('extras:script', kwargs={'pk': self.script.pk})
with (
patch('extras.views.any_workers_for_queue', return_value=True),
patch('extras.jobs.ScriptJob.enqueue') as mock_enqueue,
):
mock_enqueue.return_value.pk = 1
response = self.client.post(url, {'_commit': 'true'})
self.assertEqual(response.status_code, 302)
data = mock_enqueue.call_args.kwargs['data']
self.assertEqual(data['label'], 'hello')
self.assertIs(data['flag'], True)
# A multi-value default must be set on the QueryDict with setlist(): a plain
# assignment stores the list as one nested value, which the multi-select widget
# then rejects as a single invalid choice.
self.assertEqual(data['picks'], ['a', 'b'])
class ScriptValidationErrorTestCase(TestCase):
user_permissions = ['extras.view_script', 'extras.run_script']

9
netbox/extras/views.py
View File

@ -24,6 +24,7 @@ from dcim.models import Device, DeviceRole, Platform
from extras.choices import LogLevelChoices
from extras.dashboard.forms import DashboardWidgetAddForm, DashboardWidgetForm
from extras.dashboard.utils import get_widget_class
from extras.scripts import prepare_script_form
from extras.utils import SharedObjectViewMixin
from netbox.object_actions import *
from netbox.ui import layout
@ -1739,13 +1740,7 @@ class ScriptView(BaseScriptView):
'script': script,
})
# Populate missing variables with their default values, if defined
post_data = request.POST.copy()
for name, var in script_class._get_vars().items():
if name not in post_data and (initial := var.field_attrs.get('initial')) is not None:
post_data[name] = initial
form = script_class.as_form(post_data, request.FILES)
form = prepare_script_form(script_class, request.POST, request.FILES)
# Allow execution only if RQ worker process is running
if not any_workers_for_queue('default'):

7
netbox/netbox/filtersets.py
View File

@ -18,6 +18,7 @@ from extras.models import CustomField, SavedFilter
from users.filterset_mixins import OwnerFilterMixin
from utilities import filters
from utilities.constants import (
FILTER_ARRAY_BASED_LOOKUP_MAP,
FILTER_CHAR_BASED_LOOKUP_MAP,
FILTER_NEGATION_LOOKUP_MAP,
FILTER_NUMERIC_BASED_LOOKUP_MAP,
@ -170,6 +171,12 @@ class BaseFilterSet(django_filters.FilterSet):
# These filter types support only negation
return FILTER_NEGATION_LOOKUP_MAP
if isinstance(existing_filter, (
filters.MultiValueArrayFilter,
)):
# Must precede the char-based branch below, which would otherwise shadow this subclass
return FILTER_ARRAY_BASED_LOOKUP_MAP
if isinstance(existing_filter, (
django_filters.filters.CharFilter,
django_filters.ChoiceFilter,

127
netbox/translations/en/LC_MESSAGES/django.po
View File

@ -8,7 +8,7 @@ msgid ""
msgstr ""
"Project-Id-Version: PACKAGE VERSION\n"
"Report-Msgid-Bugs-To: \n"
"POT-Creation-Date: 2026-09-02 16:46+0000\n"
"POT-Creation-Date: 2026-09-08 05:02+0000\n"
"PO-Revision-Date: YEAR-MO-DA HO:MI+ZONE\n"
"Last-Translator: FULL NAME <EMAIL@ADDRESS>\n"
"Language-Team: LANGUAGE <LL@li.org>\n"
@ -2309,7 +2309,7 @@ msgid "Sync interval"
msgstr ""
#: netbox/core/forms/bulk_edit.py:34 netbox/extras/forms/model_forms.py:429
#: netbox/extras/views.py:399 netbox/vpn/forms/filtersets.py:107
#: netbox/extras/views.py:400 netbox/vpn/forms/filtersets.py:107
#: netbox/vpn/forms/filtersets.py:138 netbox/vpn/forms/filtersets.py:163
#: netbox/vpn/forms/filtersets.py:183 netbox/vpn/forms/model_forms.py:335
#: netbox/vpn/forms/model_forms.py:361 netbox/vpn/forms/model_forms.py:387
@ -2715,7 +2715,7 @@ msgstr ""
msgid "last updated"
msgstr ""
#: netbox/core/models/data.py:306 netbox/dcim/models/cables.py:797
#: netbox/core/models/data.py:306 netbox/dcim/models/cables.py:812
msgid "path"
msgstr ""
@ -6907,110 +6907,110 @@ msgstr ""
msgid "cables"
msgstr ""
#: netbox/dcim/models/cables.py:282
#: netbox/dcim/models/cables.py:296
msgid "Must specify a unit when setting a cable length"
msgstr ""
#: netbox/dcim/models/cables.py:285
#: netbox/dcim/models/cables.py:299
msgid "Must define A and B terminations when creating a new cable."
msgstr ""
#: netbox/dcim/models/cables.py:296
#: netbox/dcim/models/cables.py:310
msgid "Cannot connect different termination types to same end of cable."
msgstr ""
#: netbox/dcim/models/cables.py:304
#: netbox/dcim/models/cables.py:318
#, python-brace-format
msgid "Incompatible termination types: {type_a} and {type_b}"
msgstr ""
#: netbox/dcim/models/cables.py:314
#: netbox/dcim/models/cables.py:328
msgid "A and B terminations cannot connect to the same object."
msgstr ""
#: netbox/dcim/models/cables.py:574 netbox/ipam/models/asns.py:38
#: netbox/dcim/models/cables.py:589 netbox/ipam/models/asns.py:38
msgid "end"
msgstr ""
#: netbox/dcim/models/cables.py:645
#: netbox/dcim/models/cables.py:660
msgid "cable termination"
msgstr ""
#: netbox/dcim/models/cables.py:646
#: netbox/dcim/models/cables.py:661
msgid "cable terminations"
msgstr ""
#: netbox/dcim/models/cables.py:659
#: netbox/dcim/models/cables.py:674
#, python-brace-format
msgid ""
"Cannot connect a cable to {obj_parent} > {obj} because it is marked as "
"connected."
msgstr ""
#: netbox/dcim/models/cables.py:676
#: netbox/dcim/models/cables.py:691
#, python-brace-format
msgid ""
"Duplicate termination found for {app_label}.{model} {termination_id}: cable "
"{cable_pk}"
msgstr ""
#: netbox/dcim/models/cables.py:688
#: netbox/dcim/models/cables.py:703
msgid ""
"Cables cannot be terminated directly to a channel subinterface; cable the "
"parent interface instead."
msgstr ""
#: netbox/dcim/models/cables.py:694
#: netbox/dcim/models/cables.py:709
#, python-brace-format
msgid "Cables cannot be terminated to {type_display} interfaces"
msgstr ""
#: netbox/dcim/models/cables.py:701
#: netbox/dcim/models/cables.py:716
msgid "Circuit terminations attached to a provider network may not be cabled."
msgstr ""
#: netbox/dcim/models/cables.py:733
#: netbox/dcim/models/cables.py:748
msgid ""
"Invalid cable termination: the assigned termination object does not exist."
msgstr ""
#: netbox/dcim/models/cables.py:801 netbox/extras/models/configs.py:105
#: netbox/dcim/models/cables.py:816 netbox/extras/models/configs.py:105
msgid "is active"
msgstr ""
#: netbox/dcim/models/cables.py:805
#: netbox/dcim/models/cables.py:820
msgid "is complete"
msgstr ""
#: netbox/dcim/models/cables.py:809
#: netbox/dcim/models/cables.py:824
msgid "is split"
msgstr ""
#: netbox/dcim/models/cables.py:822
#: netbox/dcim/models/cables.py:837
msgid "cable path"
msgstr ""
#: netbox/dcim/models/cables.py:823
#: netbox/dcim/models/cables.py:838
msgid "cable paths"
msgstr ""
#: netbox/dcim/models/cables.py:914
#: netbox/dcim/models/cables.py:929
msgid "All originating terminations must be attached to the same link"
msgstr ""
#: netbox/dcim/models/cables.py:932
#: netbox/dcim/models/cables.py:947
msgid "All mid-span terminations must have the same termination type"
msgstr ""
#: netbox/dcim/models/cables.py:940
#: netbox/dcim/models/cables.py:955
msgid "All mid-span terminations must have the same parent object"
msgstr ""
#: netbox/dcim/models/cables.py:970
#: netbox/dcim/models/cables.py:985
msgid "All links must be cable or wireless"
msgstr ""
#: netbox/dcim/models/cables.py:972
#: netbox/dcim/models/cables.py:987
msgid "All links must match first link type"
msgstr ""
@ -9732,7 +9732,7 @@ msgstr ""
msgid "Unknown related object(s): {name}"
msgstr ""
#: netbox/extras/api/mixins.py:92 netbox/extras/api/views.py:257
#: netbox/extras/api/mixins.py:92 netbox/extras/api/views.py:258
msgid ""
"The rendered config template. When the client requests `text/plain`, the raw "
"rendered content is returned in place of the JSON object."
@ -9742,11 +9742,11 @@ msgstr ""
msgid "No config template could be resolved for this object."
msgstr ""
#: netbox/extras/api/mixins.py:102 netbox/extras/api/views.py:263
#: netbox/extras/api/mixins.py:102 netbox/extras/api/views.py:264
msgid "An error occurred while rendering the config template."
msgstr ""
#: netbox/extras/api/mixins.py:125 netbox/extras/views.py:1316
#: netbox/extras/api/mixins.py:125 netbox/extras/views.py:1317
#, python-brace-format
msgid "Config template with ID {id} not found."
msgstr ""
@ -9797,31 +9797,40 @@ msgid ""
msgstr ""
#: netbox/extras/api/serializers_/scripts.py:197
msgid ""
"Invalid data payload; expected an object mapping variable names to values."
msgstr ""
#: netbox/extras/api/serializers_/scripts.py:207
#: netbox/extras/api/serializers_/scripts.py:217
msgid "Scheduling is not enabled for this script."
msgstr ""
#: netbox/extras/api/serializers_/scripts.py:199
#: netbox/extras/api/serializers_/scripts.py:209
#: netbox/extras/forms/reports.py:45 netbox/extras/forms/scripts.py:62
msgid "Scheduled time must be in the future."
msgstr ""
#: netbox/extras/api/views.py:371
#: netbox/extras/api/views.py:372
msgid "The script has been enqueued for execution."
msgstr ""
#: netbox/extras/api/views.py:384
#: netbox/extras/api/views.py:385
msgid "This token does not permit write operations (running a script)."
msgstr ""
#: netbox/extras/api/views.py:389
#: netbox/extras/api/views.py:390
msgid "This user does not have permission to run this script."
msgstr ""
#: netbox/extras/api/views.py:399
#: netbox/extras/api/views.py:400
msgid "This script is not currently executable."
msgstr ""
#: netbox/extras/api/views.py:429
msgid "Invalid script input."
msgstr ""
#: netbox/extras/choices.py:30 netbox/extras/forms/misc.py:14
msgid "Text"
msgstr ""
@ -10489,7 +10498,7 @@ msgstr ""
msgid "Event types"
msgstr ""
#: netbox/extras/forms/bulk_edit.py:315 netbox/extras/views.py:847
#: netbox/extras/forms/bulk_edit.py:315 netbox/extras/views.py:848
msgid "Conditions"
msgstr ""
@ -12010,28 +12019,28 @@ msgstr ""
msgid "tagged items"
msgstr ""
#: netbox/extras/scripts.py:437
#: netbox/extras/scripts.py:442
#, python-brace-format
msgid ""
"Invalid job_timeout value '{value}': must be an integer (seconds) or a "
"duration string such as '1h' or '30m'."
msgstr ""
#: netbox/extras/scripts.py:442
#: netbox/extras/scripts.py:447
#, python-brace-format
msgid "Invalid job_timeout value '{value}': must be a positive duration."
msgstr ""
#: netbox/extras/scripts.py:451
#: netbox/extras/scripts.py:456
#, python-brace-format
msgid "Invalid notifications value '{value}': must be one of {valid}."
msgstr ""
#: netbox/extras/scripts.py:556
#: netbox/extras/scripts.py:561
msgid "Script Data"
msgstr ""
#: netbox/extras/scripts.py:563
#: netbox/extras/scripts.py:568
msgid "Script Execution Parameters"
msgstr ""
@ -12249,8 +12258,8 @@ msgstr ""
msgid "Attachment"
msgstr ""
#: netbox/extras/ui/panels.py:264 netbox/extras/views.py:252
#: netbox/extras/views.py:324
#: netbox/extras/ui/panels.py:264 netbox/extras/views.py:253
#: netbox/extras/views.py:325
msgid "Assigned Models"
msgstr ""
@ -12333,67 +12342,67 @@ msgstr ""
msgid "Invalid attribute \"{name}\" for {model}"
msgstr ""
#: netbox/extras/views.py:255
#: netbox/extras/views.py:256
msgid "Link Text"
msgstr ""
#: netbox/extras/views.py:256
#: netbox/extras/views.py:257
msgid "Link URL"
msgstr ""
#: netbox/extras/views.py:325 netbox/extras/views.py:1219
#: netbox/extras/views.py:326 netbox/extras/views.py:1220
msgid "Environment Parameters"
msgstr ""
#: netbox/extras/views.py:328 netbox/extras/views.py:1222
#: netbox/extras/views.py:329 netbox/extras/views.py:1223
msgid "Template"
msgstr ""
#: netbox/extras/views.py:492
#: netbox/extras/views.py:493
msgid "Table configurations must be created from an object list view."
msgstr ""
#: netbox/extras/views.py:777
#: netbox/extras/views.py:778
msgid "Additional Headers"
msgstr ""
#: netbox/extras/views.py:778
#: netbox/extras/views.py:779
msgid "Body Template"
msgstr ""
#: netbox/extras/views.py:921
#: netbox/extras/views.py:922
msgid "Tagged Objects"
msgstr ""
#: netbox/extras/views.py:1014
#: netbox/extras/views.py:1015
msgid "JSON Schema"
msgstr ""
#: netbox/extras/views.py:1511
#: netbox/extras/views.py:1512
msgid "Your dashboard has been reset."
msgstr ""
#: netbox/extras/views.py:1557
#: netbox/extras/views.py:1558
msgid "Added widget: "
msgstr ""
#: netbox/extras/views.py:1598
#: netbox/extras/views.py:1599
msgid "Updated widget: "
msgstr ""
#: netbox/extras/views.py:1634
#: netbox/extras/views.py:1635
msgid "Deleted widget: "
msgstr ""
#: netbox/extras/views.py:1636
#: netbox/extras/views.py:1637
msgid "Error deleting widget: "
msgstr ""
#: netbox/extras/views.py:1752
#: netbox/extras/views.py:1747
msgid "Unable to run script: RQ worker process not running."
msgstr ""
#: netbox/extras/views.py:1771
#: netbox/extras/views.py:1766
#, python-brace-format
msgid "Unable to run script: {error}"
msgstr ""

6
netbox/utilities/constants.py
View File

@ -17,6 +17,12 @@ FILTER_CHAR_BASED_LOOKUP_MAP = dict(
iregex='iregex',
)
# A member is a scalar inside a stored array, so negation cannot fall back to equality
FILTER_ARRAY_BASED_LOOKUP_MAP = {
**FILTER_CHAR_BASED_LOOKUP_MAP,
'n': 'contains',
}
FILTER_NUMERIC_BASED_LOOKUP_MAP = dict(
n='exact',
lte='lte',

11
netbox/utilities/tests/test_filters.py
View File

@ -28,6 +28,7 @@ from ipam.filtersets import ASNFilterSet
from ipam.models import ASN, RIR
from netbox.filtersets import BaseFilterSet
from utilities.filters import (
MultiValueArrayFilter,
MultiValueCharFilter,
MultiValueDateFilter,
MultiValueDateTimeFilter,
@ -209,6 +210,9 @@ class BaseFilterSetTestCase(TestCase):
multiplechoicefield = django_filters.MultipleChoiceFilter(
field_name='choicefield'
)
multivaluearrayfield = MultiValueArrayFilter(
field_name='charfield' # We're pretending this is an array field
)
multivaluecharfield = MultiValueCharFilter(
field_name='charfield'
)
@ -326,6 +330,13 @@ class BaseFilterSetTestCase(TestCase):
self.assertEqual(self.filters['modelmultiplechoicefield__n'].lookup_expr, 'exact')
self.assertEqual(self.filters['modelmultiplechoicefield__n'].exclude, True)
def test_multi_value_array_filter(self):
self.assertIsInstance(self.filters['multivaluearrayfield'], MultiValueArrayFilter)
self.assertEqual(self.filters['multivaluearrayfield'].lookup_expr, 'contains')
self.assertEqual(self.filters['multivaluearrayfield'].exclude, False)
self.assertEqual(self.filters['multivaluearrayfield__n'].lookup_expr, 'contains')
self.assertEqual(self.filters['multivaluearrayfield__n'].exclude, True)
def test_multi_value_char_filter(self):
self.assertIsInstance(self.filters['multivaluecharfield'], MultiValueCharFilter)
self.assertEqual(self.filters['multivaluecharfield'].lookup_expr, 'exact')