NeoAnd/netbox
1
0
Fork 0
mirror of https://github.com/netbox-community/netbox.git synced 2025-12-19 03:42:25 -06:00
Code Issues Actions 12 Packages Projects Releases Wiki Activity

Merge branch 'develop' into feature

Browse Source
This commit is contained in:
Jeremy Stretch
2023-12-27 16:34:38 -05:00
parent d3cba5dbfc 20f91a4546
commit 284e68ed9d
38 changed files with 520 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
netbox/core/management/commands/syncdatasource.py
View File

@@ -1,5 +1,6 @@
from django.core.management.base import BaseCommand, CommandError
from core.choices import DataSourceStatusChoices
from core.models import DataSource
@@ -33,9 +34,13 @@ class Command(BaseCommand):
for i, datasource in enumerate(datasources, start=1):
self.stdout.write(f"[{i}] Syncing {datasource}... ", ending='')
self.stdout.flush()
datasource.sync()
self.stdout.write(datasource.get_status_display())
self.stdout.flush()
try:
datasource.sync()
self.stdout.write(datasource.get_status_display())
self.stdout.flush()
except Exception as e:
DataSource.objects.filter(pk=datasource.pk).update(status=DataSourceStatusChoices.FAILED)
raise e
if len(options['name']) > 1:
self.stdout.write(f"Finished.")

15
netbox/core/views.py
View File

@@ -1,4 +1,5 @@
from django.contrib import messages
from django.core.cache import cache
from django.http import HttpResponseForbidden
from django.shortcuts import get_object_or_404, redirect, render
from django.views.generic import View
@@ -159,12 +160,14 @@ class ConfigView(generic.ObjectView):
queryset = ConfigRevision.objects.all()
def get_object(self, **kwargs):
if config := self.queryset.first():
return config
# Instantiate a dummy default config if none has been created yet
return ConfigRevision(
data=get_config().defaults
)
revision_id = cache.get('config_version')
try:
return ConfigRevision.objects.get(pk=revision_id)
except ConfigRevision.DoesNotExist:
# Fall back to using the active config data if no record is found
return ConfigRevision(
data=get_config()
)
class ConfigRevisionListView(generic.ObjectListView):

1
netbox/dcim/filtersets.py
View File

@@ -1020,6 +1020,7 @@ class DeviceFilterSet(
Q(serial__icontains=value.strip()) |
Q(inventoryitems__serial__icontains=value.strip()) |
Q(asset_tag__icontains=value.strip()) |
Q(description_icontains=value.strip()) |
Q(comments__icontains=value) |
Q(primary_ip4__address__startswith=value) |
Q(primary_ip6__address__startswith=value)

22
netbox/dcim/migrations/0182_zero_length_cable_fix.py Normal file
View File

@@ -0,0 +1,22 @@
from django.db import migrations
def update_cable_lengths(apps, schema_editor):
Cable = apps.get_model('dcim', 'Cable')
# Set the absolute length for any zero-length Cables
Cable.objects.filter(length=0).update(_abs_length=0)
class Migration(migrations.Migration):
dependencies = [
('dcim', '0181_rename_device_role_device_role'),
]
operations = [
migrations.RunPython(
code=update_cable_lengths,
reverse_code=migrations.RunPython.noop
),
]

2
netbox/dcim/migrations/0182_devicetype_exclude_from_utilization.py → netbox/dcim/migrations/0183_devicetype_exclude_from_utilization.py
View File

@@ -5,7 +5,7 @@ from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('dcim', '0181_rename_device_role_device_role'),
('dcim', '0182_zero_length_cable_fix'),
]
operations = [

2
netbox/dcim/migrations/0183_protect_child_interfaces.py → netbox/dcim/migrations/0184_protect_child_interfaces.py
View File

@@ -7,7 +7,7 @@ import django.db.models.deletion
class Migration(migrations.Migration):
dependencies = [
('dcim', '0182_devicetype_exclude_from_utilization'),
('dcim', '0183_devicetype_exclude_from_utilization'),
]
operations = [

2
netbox/dcim/migrations/0184_gfk_indexes.py → netbox/dcim/migrations/0185_gfk_indexes.py
View File

@@ -6,7 +6,7 @@ from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('dcim', '0183_protect_child_interfaces'),
('dcim', '0184_protect_child_interfaces'),
]
operations = [

2
netbox/dcim/models/cables.py
View File

@@ -200,7 +200,7 @@ class Cable(PrimaryModel):
_created = self.pk is None
# Store the given length (if any) in meters for use in database ordering
if self.length and self.length_unit:
if self.length is not None and self.length_unit:
self._abs_length = to_meters(self.length, self.length_unit)
else:
self._abs_length = None

4
netbox/dcim/svg/cables.py
View File

@@ -277,7 +277,7 @@ class CableTraceSVG:
if cable.type:
# Include the cable type in the tooltip
description.append(cable.get_type_display())
if cable.length and cable.length_unit:
if cable.length is not None and cable.length_unit:
# Include the cable length in the tooltip
description.append(f'{cable.length} {cable.get_length_unit_display()}')
else:
@@ -288,7 +288,7 @@ class CableTraceSVG:
description = []
if cable.type:
labels.append(cable.get_type_display())
if cable.length and cable.length_unit:
if cable.length is not None and cable.length_unit:
# Include the cable length in the tooltip
labels.append(f'{cable.length} {cable.get_length_unit_display()}')

2
netbox/dcim/tables/devices.py
View File

@@ -1085,7 +1085,7 @@ class VirtualDeviceContextTable(TenancyColumnsMixin, NetBoxTable):
comments = columns.MarkdownColumn()
tags = columns.TagColumn(
url_name='dcim:vdc_list'
url_name='dcim:virtualdevicecontext_list'
)
class Meta(NetBoxTable.Meta):

3
netbox/dcim/views.py
View File

@@ -692,8 +692,7 @@ class RackRackReservationsView(generic.ObjectChildrenView):
label=_('Reservations'),
badge=lambda obj: obj.reservations.count(),
permission='dcim.view_rackreservation',
weight=510,
hide_if_empty=True
weight=510
)
def get_children(self, request, parent):

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

@@ -114,7 +114,7 @@ class Command(BaseCommand):
# Create the job
job = Job.objects.create(
object=module,
name=script.name,
name=script.class_name,
user=User.objects.filter(is_superuser=True).order_by('pk')[0],
job_id=uuid.uuid4()
)

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

@@ -579,8 +579,7 @@ class CustomField(CloningMixin, ExportTemplatesMixin, ChangeLoggedModel):
# Multiselect
elif self.type == CustomFieldTypeChoices.TYPE_MULTISELECT:
filter_class = filters.MultiValueCharFilter
kwargs['lookup_expr'] = 'has_key'
filter_class = filters.MultiValueArrayFilter
# Object
elif self.type == CustomFieldTypeChoices.TYPE_OBJECT:

2
netbox/extras/models/models.py
View File

@@ -398,7 +398,7 @@ class CustomLink(CloningMixin, ExportTemplatesMixin, ChangeLoggedModel):
text = clean_html(text, allowed_schemes)
# Sanitize link
link = urllib.parse.quote(link, safe='/:?&=%+[]@#,;')
link = urllib.parse.quote(link, safe='/:?&=%+[]@#,;!')
# Verify link scheme is allowed
result = urllib.parse.urlparse(link)

29
netbox/extras/signals.py
View File

@@ -69,21 +69,20 @@ def handle_changed_object(sender, instance, **kwargs):
return
# Record an ObjectChange if applicable
if hasattr(instance, 'to_objectchange'):
if m2m_changed:
ObjectChange.objects.filter(
changed_object_type=ContentType.objects.get_for_model(instance),
changed_object_id=instance.pk,
request_id=request.id
).update(
postchange_data=instance.to_objectchange(action).postchange_data
)
else:
objectchange = instance.to_objectchange(action)
if objectchange and objectchange.has_changes:
objectchange.user = request.user
objectchange.request_id = request.id
objectchange.save()
if m2m_changed:
ObjectChange.objects.filter(
changed_object_type=ContentType.objects.get_for_model(instance),
changed_object_id=instance.pk,
request_id=request.id
).update(
postchange_data=instance.to_objectchange(action).postchange_data
)
else:
objectchange = instance.to_objectchange(action)
if objectchange and objectchange.has_changes:
objectchange.user = request.user
objectchange.request_id = request.id
objectchange.save()
# If this is an M2M change, update the previously queued webhook (from post_save)
queue = events_queue.get()

265
netbox/extras/tests/test_custom_validation.py Normal file
View File

@@ -0,0 +1,265 @@
from django.test import TestCase
from django.test import override_settings
from circuits.api.serializers import ProviderSerializer
from circuits.forms import ProviderForm
from circuits.models import Provider
from ipam.models import ASN, RIR
from utilities.choices import CSVDelimiterChoices, ImportFormatChoices
from utilities.testing import APITestCase, ModelViewTestCase, create_tags, post_data
class ModelFormCustomValidationTest(TestCase):
@override_settings(CUSTOM_VALIDATORS={
'circuits.provider': [
{'tags': {'required': True}}
]
})
def test_tags_validation(self):
"""
Check that custom validation rules work for tag assignment.
"""
data = {
'name': 'Provider 1',
'slug': 'provider-1',
}
form = ProviderForm(data)
self.assertFalse(form.is_valid())
tags = create_tags('Tag1', 'Tag2', 'Tag3')
data['tags'] = [tag.pk for tag in tags]
form = ProviderForm(data)
self.assertTrue(form.is_valid())
@override_settings(CUSTOM_VALIDATORS={
'circuits.provider': [
{'asns': {'required': True}}
]
})
def test_m2m_validation(self):
"""
Check that custom validation rules work for many-to-many fields.
"""
data = {
'name': 'Provider 1',
'slug': 'provider-1',
}
form = ProviderForm(data)
self.assertFalse(form.is_valid())
rir = RIR.objects.create(name='RIR 1', slug='rir-1')
asns = ASN.objects.bulk_create((
ASN(rir=rir, asn=65001),
ASN(rir=rir, asn=65002),
ASN(rir=rir, asn=65003),
))
data['asns'] = [asn.pk for asn in asns]
form = ProviderForm(data)
self.assertTrue(form.is_valid())
class BulkEditCustomValidationTest(ModelViewTestCase):
model = Provider
@classmethod
def setUpTestData(cls):
rir = RIR.objects.create(name='RIR 1', slug='rir-1')
asns = ASN.objects.bulk_create((
ASN(rir=rir, asn=65001),
ASN(rir=rir, asn=65002),
ASN(rir=rir, asn=65003),
))
providers = (
Provider(name='Provider 1', slug='provider-1'),
Provider(name='Provider 2', slug='provider-2'),
Provider(name='Provider 3', slug='provider-3'),
)
Provider.objects.bulk_create(providers)
for provider in providers:
provider.asns.set(asns)
@override_settings(CUSTOM_VALIDATORS={
'circuits.provider': [
{'asns': {'required': True}}
]
})
def test_bulk_edit_without_m2m(self):
"""
Check that custom validation rules do not interfere with bulk editing.
"""
data = {
'pk': list(Provider.objects.values_list('pk', flat=True)),
'_apply': '',
'description': 'New description',
}
self.add_permissions(
'circuits.view_provider',
'circuits.change_provider',
)
# Bulk edit the description without changing ASN assignments
request = {
'path': self._get_url('bulk_edit'),
'data': post_data(data),
}
response = self.client.post(**request)
self.assertHttpStatus(response, 302)
self.assertEqual(
Provider.objects.filter(description=data['description']).count(),
len(data['pk'])
)
@override_settings(CUSTOM_VALIDATORS={
'circuits.provider': [
{'asns': {'required': True}}
]
})
def test_bulk_edit_m2m(self):
"""
Test that custom validation rules are enforced during bulk editing.
"""
data = {
'pk': list(Provider.objects.values_list('pk', flat=True)),
'_apply': '',
'description': 'New description',
}
self.add_permissions(
'circuits.view_provider',
'circuits.change_provider',
'ipam.view_asn',
)
# Change the ASN assignments
asn = ASN.objects.first()
data['asns'] = [asn.pk]
request = {
'path': self._get_url('bulk_edit'),
'data': post_data(data),
}
response = self.client.post(**request)
self.assertHttpStatus(response, 302)
for provider in Provider.objects.all():
self.assertEqual(len(provider.asns.all()), 1)
# Attempt to remove the ASN assignments
data.pop('asns')
data['_nullify'] = 'asns'
request = {
'path': self._get_url('bulk_edit'),
'data': post_data(data),
}
response = self.client.post(**request)
self.assertHttpStatus(response, 200)
for provider in Provider.objects.all():
self.assertTrue(provider.asns.exists())
class BulkImportCustomValidationTest(ModelViewTestCase):
model = Provider
@classmethod
def setUpTestData(cls):
create_tags('Tag1', 'Tag2', 'Tag3')
@override_settings(CUSTOM_VALIDATORS={
'circuits.provider': [
{'tags': {'required': True}}
]
})
def test_bulk_import_invalid(self):
"""
Test that custom validation rules are enforced during bulk import.
"""
csv_data = (
"name,slug",
"Provider 1,provider-1",
"Provider 2,provider-2",
"Provider 3,provider-3",
)
data = {
'data': '\n'.join(csv_data),
'format': ImportFormatChoices.CSV,
'csv_delimiter': CSVDelimiterChoices.COMMA,
}
self.add_permissions(
'circuits.view_provider',
'circuits.add_provider',
'extras.view_tag',
)
# Attempt to import providers without tags
request = {
'path': self._get_url('import'),
'data': post_data(data),
}
response = self.client.post(**request)
self.assertHttpStatus(response, 200)
self.assertFalse(Provider.objects.exists())
# Import providers successfully with tag assignments
csv_data = (
"name,slug,tags",
"Provider 1,provider-1,tag1",
"Provider 2,provider-2,tag2",
"Provider 3,provider-3,tag3",
)
data['data'] = '\n'.join(csv_data)
request = {
'path': self._get_url('import'),
'data': post_data(data),
}
response = self.client.post(**request)
self.assertHttpStatus(response, 302)
self.assertTrue(Provider.objects.exists())
class APISerializerCustomValidationTest(APITestCase):
@override_settings(CUSTOM_VALIDATORS={
'circuits.provider': [
{'tags': {'required': True}}
]
})
def test_tags_validation(self):
"""
Check that custom validation rules work for tag assignment.
"""
data = {
'name': 'Provider 1',
'slug': 'provider-1',
}
serializer = ProviderSerializer(data=data)
self.assertFalse(serializer.is_valid())
tags = create_tags('Tag1', 'Tag2', 'Tag3')
data['tags'] = [tag.pk for tag in tags]
serializer = ProviderSerializer(data=data)
self.assertTrue(serializer.is_valid())
@override_settings(CUSTOM_VALIDATORS={
'circuits.provider': [
{'asns': {'required': True}}
]
})
def test_m2m_validation(self):
"""
Check that custom validation rules work for many-to-many fields.
"""
data = {
'name': 'Provider 1',
'slug': 'provider-1',
}
serializer = ProviderSerializer(data=data)
self.assertFalse(serializer.is_valid())
rir = RIR.objects.create(name='RIR 1', slug='rir-1')
asns = ASN.objects.bulk_create((
ASN(rir=rir, asn=65001),
ASN(rir=rir, asn=65002),
ASN(rir=rir, asn=65003),
))
data['asns'] = [asn.pk for asn in asns]
serializer = ProviderSerializer(data=data)
self.assertTrue(serializer.is_valid())

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

@@ -1329,7 +1329,7 @@ class CustomFieldModelFilterTest(TestCase):
choice_set = CustomFieldChoiceSet.objects.create(
name='Custom Field Choice Set 1',
extra_choices=(('a', 'A'), ('b', 'B'), ('c', 'C'), ('x', 'X'))
extra_choices=(('a', 'A'), ('b', 'B'), ('c', 'C'))
)
# Integer filtering
@@ -1435,7 +1435,7 @@ class CustomFieldModelFilterTest(TestCase):
'cf7': 'http://a.example.com',
'cf8': 'http://a.example.com',
'cf9': 'A',
'cf10': ['A', 'X'],
'cf10': ['A', 'B'],
'cf11': manufacturers[0].pk,
'cf12': [manufacturers[0].pk, manufacturers[3].pk],
}),
@@ -1449,7 +1449,7 @@ class CustomFieldModelFilterTest(TestCase):
'cf7': 'http://b.example.com',
'cf8': 'http://b.example.com',
'cf9': 'B',
'cf10': ['B', 'X'],
'cf10': ['B', 'C'],
'cf11': manufacturers[1].pk,
'cf12': [manufacturers[1].pk, manufacturers[3].pk],
}),
@@ -1463,7 +1463,7 @@ class CustomFieldModelFilterTest(TestCase):
'cf7': 'http://c.example.com',
'cf8': 'http://c.example.com',
'cf9': 'C',
'cf10': ['C', 'X'],
'cf10': None,
'cf11': manufacturers[2].pk,
'cf12': [manufacturers[2].pk, manufacturers[3].pk],
}),
@@ -1531,8 +1531,9 @@ class CustomFieldModelFilterTest(TestCase):
self.assertEqual(self.filterset({'cf_cf9': ['A', 'B']}, self.queryset).qs.count(), 2)
def test_filter_multiselect(self):
self.assertEqual(self.filterset({'cf_cf10': ['A', 'B']}, self.queryset).qs.count(), 2)
self.assertEqual(self.filterset({'cf_cf10': ['X']}, self.queryset).qs.count(), 3)
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)
self.assertEqual(self.filterset({'cf_cf10': ['null']}, self.queryset).qs.count(), 1)
def test_filter_object(self):
manufacturer_ids = Manufacturer.objects.values_list('id', flat=True)

23
netbox/extras/validators.py
View File

@@ -91,8 +91,7 @@ class CustomValidator:
def __call__(self, instance):
# Validate instance attributes per validation rules
for attr_name, rules in self.validation_rules.items():
assert hasattr(instance, attr_name), f"Invalid attribute '{attr_name}' for {instance.__class__.__name__}"
attr = getattr(instance, attr_name)
attr = self._getattr(instance, attr_name)
for descriptor, value in rules.items():
validator = self.get_validator(descriptor, value)
try:
@@ -104,6 +103,26 @@ class CustomValidator:
# Execute custom validation logic (if any)
self.validate(instance)
@staticmethod
def _getattr(instance, name):
# Attempt to resolve many-to-many fields to their stored values
m2m_fields = [f.name for f in instance._meta.local_many_to_many]
if name in m2m_fields:
if name in getattr(instance, '_m2m_values', []):
return instance._m2m_values[name]
if instance.pk:
return list(getattr(instance, name).all())
return []
# Raise a ValidationError for unknown attributes
if not hasattr(instance, name):
raise ValidationError(_('Invalid attribute "{name}" for {model}').format(
name=name,
model=instance.__class__.__name__
))
return getattr(instance, name)
def get_validator(self, descriptor, value):
"""
Instantiate and return the appropriate validator based on the descriptor given. For

8
netbox/ipam/filtersets.py
View File

@@ -949,6 +949,10 @@ class VLANFilterSet(NetBoxModelFilterSet, TenancyFilterSet):
choices=VLANStatusChoices,
null_value=None
)
available_at_site = django_filters.ModelChoiceFilter(
queryset=Site.objects.all(),
method='get_for_site'
)
available_on_device = django_filters.ModelChoiceFilter(
queryset=Device.objects.all(),
method='get_for_device'
@@ -983,6 +987,10 @@ class VLANFilterSet(NetBoxModelFilterSet, TenancyFilterSet):
pass
return queryset.filter(qs_filter)
@extend_schema_field(OpenApiTypes.STR)
def get_for_site(self, queryset, name, value):
return queryset.get_for_site(value)
@extend_schema_field(OpenApiTypes.STR)
def get_for_device(self, queryset, name, value):
return queryset.get_for_device(value)

6
netbox/ipam/models/vlans.py
View File

@@ -225,11 +225,11 @@ class VLAN(PrimaryModel):
# Validate VLAN group (if assigned)
if self.group and self.site and self.group.scope != self.site:
raise ValidationError({
'group': _(
raise ValidationError(
_(
"VLAN is assigned to group {group} (scope: {scope}); cannot also assign to site {site}."
).format(group=self.group, scope=self.group.scope, site=self.site)
})
)
# Validate group min/max VIDs
if self.group and not self.group.min_vid <= self.vid <= self.group.max_vid:

29
netbox/ipam/querysets.py
View File

@@ -69,6 +69,35 @@ class VLANGroupQuerySet(RestrictedQuerySet):
class VLANQuerySet(RestrictedQuerySet):
def get_for_site(self, site):
"""
Return all VLANs in the specified site
"""
from .models import VLANGroup
q = Q()
q |= Q(
scope_type=ContentType.objects.get_by_natural_key('dcim', 'site'),
scope_id=site.pk
)
if site.region:
q |= Q(
scope_type=ContentType.objects.get_by_natural_key('dcim', 'region'),
scope_id__in=site.region.get_ancestors(include_self=True)
)
if site.group:
q |= Q(
scope_type=ContentType.objects.get_by_natural_key('dcim', 'sitegroup'),
scope_id__in=site.group.get_ancestors(include_self=True)
)
return self.filter(
Q(group__in=VLANGroup.objects.filter(q)) |
Q(site=site) |
Q(group__scope_id__isnull=True, site__isnull=True) | # Global group VLANs
Q(group__isnull=True, site__isnull=True) # Global VLANs
)
def get_for_device(self, device):
"""
Return all VLANs available to the specified Device.

6
netbox/ipam/signals.py
View File

@@ -56,8 +56,12 @@ def clear_primary_ip(instance, **kwargs):
"""
field_name = f'primary_ip{instance.family}'
if device := Device.objects.filter(**{field_name: instance}).first():
device.snapshot()
setattr(device, field_name, None)
device.save()
if virtualmachine := VirtualMachine.objects.filter(**{field_name: instance}).first():
virtualmachine.snapshot()
setattr(virtualmachine, field_name, None)
virtualmachine.save()
@@ -67,4 +71,6 @@ def clear_oob_ip(instance, **kwargs):
When an IPAddress is deleted, trigger save() on any Devices for which it was a OOB IP.
"""
if device := Device.objects.filter(oob_ip=instance).first():
device.snapshot()
device.oob_ip = None
device.save()

13
netbox/ipam/tests/test_filtersets.py
View File

@@ -1359,6 +1359,7 @@ class VLANTestCase(TestCase, ChangeLoggedFilterSetTests):
VLANGroup(name='VLAN Group 1', slug='vlan-group-1'),
VLANGroup(name='VLAN Group 2', slug='vlan-group-2'),
VLANGroup(name='VLAN Group 3', slug='vlan-group-3'),
VLANGroup(name='VLAN Group 4', slug='vlan-group-4'),
)
VLANGroup.objects.bulk_create(groups)
@@ -1415,6 +1416,9 @@ class VLANTestCase(TestCase, ChangeLoggedFilterSetTests):
VLAN(vid=301, name='VLAN 301', site=sites[5], group=groups[23], role=roles[2], tenant=tenants[2], status=VLANStatusChoices.STATUS_RESERVED),
VLAN(vid=302, name='VLAN 302', site=sites[5], group=groups[23], role=roles[2], tenant=tenants[2], status=VLANStatusChoices.STATUS_RESERVED),
# Create one globally available VLAN on a VLAN group
VLAN(vid=500, name='VLAN Group 1', group=groups[24]),
# Create one globally available VLAN
VLAN(vid=1000, name='Global VLAN'),
)
@@ -1488,12 +1492,17 @@ class VLANTestCase(TestCase, ChangeLoggedFilterSetTests):
def test_available_on_device(self):
device_id = Device.objects.first().pk
params = {'available_on_device': device_id}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 6) # 5 scoped + 1 global
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 7) # 5 scoped + 1 global group + 1 global
def test_available_on_virtualmachine(self):
vm_id = VirtualMachine.objects.first().pk
params = {'available_on_virtualmachine': vm_id}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 6) # 5 scoped + 1 global
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 7) # 5 scoped + 1 global group + 1 global
def test_available_at_site(self):
site_id = Site.objects.first().pk
params = {'available_at_site': site_id}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 5) # 4 scoped + 1 global group + 1 global
class ServiceTemplateTestCase(TestCase, ChangeLoggedFilterSetTests):

20
netbox/ipam/views.py
View File

@@ -659,6 +659,26 @@ class IPRangeListView(generic.ObjectListView):
class IPRangeView(generic.ObjectView):
queryset = IPRange.objects.all()
def get_extra_context(self, request, instance):
# Parent prefixes table
parent_prefixes = Prefix.objects.restrict(request.user, 'view').filter(
Q(prefix__net_contains_or_equals=str(instance.start_address.ip)),
Q(prefix__net_contains_or_equals=str(instance.end_address.ip)),
vrf=instance.vrf
).prefetch_related(
'site', 'role', 'tenant', 'vlan', 'role'
)
parent_prefixes_table = tables.PrefixTable(
list(parent_prefixes),
exclude=('vrf', 'utilization'),
orderable=False
)
return {
'parent_prefixes_table': parent_prefixes_table,
}
@register_model_view(IPRange, 'ipaddresses', path='ip-addresses')
class IPRangeIPAddressesView(generic.ObjectChildrenView):

13
netbox/netbox/api/serializers/base.py
View File

@@ -23,16 +23,16 @@ class ValidatedModelSerializer(BaseModelSerializer):
validation. (DRF does not do this by default; see https://github.com/encode/django-rest-framework/issues/3144)
"""
def validate(self, data):
# Remove custom fields data and tags (if any) prior to model validation
attrs = data.copy()
# Remove custom field data (if any) prior to model validation
attrs.pop('custom_fields', None)
attrs.pop('tags', None)
# Skip ManyToManyFields
for field in self.Meta.model._meta.get_fields():
if isinstance(field, ManyToManyField):
attrs.pop(field.name, None)
m2m_values = {}
for field in self.Meta.model._meta.local_many_to_many:
if field.name in attrs:
m2m_values[field.name] = attrs.pop(field.name)
# Run clean() on an instance of the model
if self.instance is None:
@@ -41,6 +41,7 @@ class ValidatedModelSerializer(BaseModelSerializer):
instance = self.instance
for k, v in attrs.items():
setattr(instance, k, v)
instance._m2m_values = m2m_values
instance.full_clean()
return data

11
netbox/netbox/forms/base.py
View File

@@ -57,6 +57,17 @@ class NetBoxModelForm(BootstrapMixin, CheckLastUpdatedMixin, CustomFieldsMixin,
return super().clean()
def _post_clean(self):
"""
Override BaseModelForm's _post_clean() to store many-to-many field values on the model instance.
"""
self.instance._m2m_values = {}
for field in self.instance._meta.local_many_to_many:
if field.name in self.cleaned_data:
self.instance._m2m_values[field.name] = list(self.cleaned_data[field.name])
return super()._post_clean()
class NetBoxModelImportForm(CSVModelForm, NetBoxModelForm):
"""

8
netbox/netbox/views/generic/bulk_views.py
View File

@@ -556,6 +556,14 @@ class BulkEditView(GetReturnURLMixin, BaseMultiObjectView):
elif name in form.changed_data:
obj.custom_field_data[cf_name] = customfield.serialize(form.cleaned_data[name])
# Store M2M values for validation
obj._m2m_values = {}
for field in obj._meta.local_many_to_many:
if value := form.cleaned_data.get(field.name):
obj._m2m_values[field.name] = list(value)
elif field.name in nullified_fields:
obj._m2m_values[field.name] = []
obj.full_clean()
obj.save()
updated_objects.append(obj)

2
netbox/templates/dcim/cable.html
View File

@@ -50,7 +50,7 @@
<tr>
<th scope="row">{% trans "Length" %}</th>
<td>
{% if object.length %}
{% if object.length is not None %}
{{ object.length|floatformat }} {{ object.get_length_unit_display }}
{% else %}
{{ ''|placeholder }}

5
netbox/templates/ipam/iprange.html
View File

@@ -82,6 +82,11 @@
{% plugin_right_page object %}
</div>
</div>
<div class="row">
<div class="col col-md-12">
{% include 'inc/panel_table.html' with table=parent_prefixes_table heading='Parent Prefixes' %}
</div>
</div>
<div class="row">
<div class="col col-md-12">
{% plugin_full_width_page object %}

16
netbox/utilities/filters.py
View File

@@ -9,6 +9,7 @@ from drf_spectacular.types import OpenApiTypes
__all__ = (
'ContentTypeFilter',
'MACAddressFilter',
'MultiValueArrayFilter',
'MultiValueCharFilter',
'MultiValueDateFilter',
'MultiValueDateTimeFilter',
@@ -85,6 +86,21 @@ class MultiValueTimeFilter(django_filters.MultipleChoiceFilter):
field_class = multivalue_field_factory(forms.TimeField)
@extend_schema_field(OpenApiTypes.STR)
class MultiValueArrayFilter(django_filters.MultipleChoiceFilter):
field_class = multivalue_field_factory(forms.CharField)
def __init__(self, *args, lookup_expr='contains', **kwargs):
# Set default lookup_expr to 'contains'
super().__init__(*args, lookup_expr=lookup_expr, **kwargs)
def get_filter_predicate(self, v):
# If filtering for null values, ignore lookup_expr
if v is None:
return {self.field_name: None}
return super().get_filter_predicate(v)
class MACAddressFilter(django_filters.CharFilter):
pass

2
netbox/utilities/forms/fields/dynamic.py
View File

@@ -43,7 +43,7 @@ class DynamicMultipleChoiceField(forms.MultipleChoiceField):
if data is not None:
self.choices = [
choice for choice in self.choices if choice[0] in data
choice for choice in self.choices if choice[0] and choice[0] in data
]
return bound_field