parent
f12fe46486
commit
9ebc55c3a4
|
|
@ -364,6 +364,7 @@ class InterfaceSerializer(
|
|||
mac_address = _UNSET
|
||||
if isinstance(data, dict):
|
||||
mac_address = data.pop('mac_address', _UNSET)
|
||||
self._validate_no_mac_conflict(data, mac_address)
|
||||
|
||||
if not self.nested and isinstance(data, dict):
|
||||
if mac_address not in (_UNSET, None):
|
||||
|
|
|
|||
|
|
@ -1,9 +1,10 @@
|
|||
from django.core.exceptions import ValidationError as DjangoValidationError
|
||||
from django.db import transaction
|
||||
from django.utils.translation import gettext as _
|
||||
from netaddr import EUI, AddrFormatError
|
||||
from rest_framework import serializers
|
||||
from rest_framework.exceptions import PermissionDenied
|
||||
|
||||
from dcim.models import MACAddress
|
||||
|
||||
_UNSET = object()
|
||||
|
||||
__all__ = (
|
||||
|
|
@ -14,56 +15,72 @@ __all__ = (
|
|||
|
||||
class MACAddressShortcutMixin:
|
||||
"""
|
||||
Mixin for Interface and VMInterface serializers that adds a write-only `mac_address` shortcut
|
||||
field for creating/updating the primary MACAddress in a single request.
|
||||
Mixin for Interface and VMInterface serializers that adds a `mac_address` shortcut field for
|
||||
creating/updating the primary MACAddress in a single request. The validated write is centralized
|
||||
on the interface model (BaseInterface.set_primary_mac_address); this mixin is a thin adapter that
|
||||
owns the permission check, the shortcut-vs-primary_mac_address conflict guard, and translating the
|
||||
model's validation errors into API errors.
|
||||
"""
|
||||
|
||||
@staticmethod
|
||||
def _validate_no_mac_conflict(data, mac_address):
|
||||
# The mac_address shortcut and the primary_mac_address field both set the primary MAC. Reject
|
||||
# only when both are supplied AND they disagree, so a read-modify-write round-trip (which echoes
|
||||
# both readable fields with matching values) is accepted while a genuine conflict is rejected.
|
||||
if mac_address is _UNSET or not isinstance(data, dict) or 'primary_mac_address' not in data:
|
||||
return
|
||||
|
||||
primary = data['primary_mac_address']
|
||||
primary_value = primary.mac_address if primary is not None else None
|
||||
try:
|
||||
shortcut_value = EUI(mac_address, version=48) if mac_address is not None else None
|
||||
except (AddrFormatError, ValueError, TypeError):
|
||||
# Leave an invalid shortcut value to the format check in each serializer's validate().
|
||||
return
|
||||
|
||||
if shortcut_value != primary_value:
|
||||
raise serializers.ValidationError(
|
||||
_("The provided 'mac_address' and 'primary_mac_address' values conflict.")
|
||||
)
|
||||
|
||||
def _check_add_mac_permission(self, instance, mac_address):
|
||||
# A submitted value that doesn't already exist on this interface will be created, which requires
|
||||
# add_macaddress. An existing value is only reassigned, so it doesn't.
|
||||
if instance is not None and instance.mac_addresses.filter(mac_address=mac_address).exists():
|
||||
return
|
||||
request = self.context.get('request')
|
||||
if request and not request.user.has_perm('dcim.add_macaddress'):
|
||||
raise PermissionDenied(_('You do not have permission to create MAC addresses.'))
|
||||
|
||||
def create(self, validated_data):
|
||||
mac_address = validated_data.pop('mac_address', None)
|
||||
if mac_address is not None:
|
||||
request = self.context.get('request')
|
||||
if request and not request.user.has_perm('dcim.add_macaddress'):
|
||||
raise PermissionDenied(_('You do not have permission to create MAC addresses.'))
|
||||
self._check_add_mac_permission(None, mac_address)
|
||||
with transaction.atomic():
|
||||
instance = super().create(validated_data)
|
||||
if mac_address is not None:
|
||||
mac = MACAddress.objects.create(mac_address=mac_address, assigned_object=instance)
|
||||
instance.primary_mac_address = mac
|
||||
instance.save()
|
||||
instance.__dict__.pop('mac_address', None)
|
||||
self._set_primary_mac(instance, mac_address)
|
||||
return instance
|
||||
|
||||
def update(self, instance, validated_data):
|
||||
mac_address = validated_data.pop('mac_address', _UNSET)
|
||||
|
||||
# Check permission and locate any existing MAC before any writes.
|
||||
if mac_address not in (_UNSET, None):
|
||||
existing_mac = instance.mac_addresses.filter(mac_address=mac_address).first()
|
||||
if existing_mac is None:
|
||||
request = self.context.get('request')
|
||||
if request and not request.user.has_perm('dcim.add_macaddress'):
|
||||
raise PermissionDenied(_('You do not have permission to create MAC addresses.'))
|
||||
else:
|
||||
existing_mac = None
|
||||
|
||||
self._check_add_mac_permission(instance, mac_address)
|
||||
with transaction.atomic():
|
||||
instance = super().update(instance, validated_data)
|
||||
if mac_address is _UNSET:
|
||||
pass
|
||||
elif mac_address is None:
|
||||
if instance.primary_mac_address_id is not None:
|
||||
instance.snapshot()
|
||||
instance.primary_mac_address = None
|
||||
instance.save()
|
||||
else:
|
||||
# Find-or-create: prefer existing MAC on this interface; create only if absent.
|
||||
mac = existing_mac
|
||||
if mac is None:
|
||||
mac = MACAddress.objects.create(mac_address=mac_address, assigned_object=instance)
|
||||
if instance.primary_mac_address_id != mac.pk:
|
||||
instance.snapshot()
|
||||
instance.primary_mac_address = mac
|
||||
instance.save()
|
||||
|
||||
instance.__dict__.pop('mac_address', None)
|
||||
if mac_address is not _UNSET:
|
||||
self._set_primary_mac(instance, mac_address)
|
||||
return instance
|
||||
|
||||
def _set_primary_mac(self, instance, mac_address):
|
||||
# Surface model/custom validation raised by the centralized operation as a DRF 400 rather than
|
||||
# a 500 (it runs after DRF's own validation phase). The viewset's discard_events_on_rollback()
|
||||
# clears any event queued for the rolled-back interface save.
|
||||
try:
|
||||
instance.set_primary_mac_address_from_value(mac_address)
|
||||
except DjangoValidationError as e:
|
||||
# Field-scoped errors (e.g. an interface-level check like qinq_svlan) keep their field key;
|
||||
# non-field errors are attributed to the mac_address shortcut that triggered the operation.
|
||||
if hasattr(e, 'error_dict'):
|
||||
raise serializers.ValidationError(e.message_dict)
|
||||
raise serializers.ValidationError({'mac_address': e.messages})
|
||||
|
|
|
|||
|
|
@ -1,13 +1,13 @@
|
|||
from django import forms
|
||||
from django.db import transaction
|
||||
from django.core.exceptions import ValidationError
|
||||
from django.utils.translation import gettext_lazy as _
|
||||
from netaddr import EUI, AddrFormatError
|
||||
|
||||
from dcim.choices import *
|
||||
from dcim.constants import *
|
||||
from dcim.models import MACAddress
|
||||
from dcim.utils import get_module_bay_positions, resolve_module_placeholder
|
||||
from netbox.context import current_request
|
||||
from utilities.exceptions import AbortRequest
|
||||
from utilities.forms import get_field_value
|
||||
|
||||
__all__ = (
|
||||
|
|
@ -61,12 +61,14 @@ class InterfaceCommonForm(forms.Form):
|
|||
raise forms.ValidationError({
|
||||
'mac_address': _('Enter a valid MAC address (e.g. 00:11:22:33:44:55).')
|
||||
})
|
||||
# Require add_macaddress permission when a MAC value is provided (it may need to be created).
|
||||
request = current_request.get()
|
||||
if request is not None and not request.user.has_perm('dcim.add_macaddress'):
|
||||
raise forms.ValidationError({
|
||||
'mac_address': _('You do not have permission to create MAC addresses.')
|
||||
})
|
||||
# Require add_macaddress only when the field is actually being changed (a MAC may need to be
|
||||
# created). A pre-populated primary MAC left untouched must not gate unrelated edits.
|
||||
if 'mac_address' in self.changed_data:
|
||||
request = current_request.get()
|
||||
if request is not None and not request.user.has_perm('dcim.add_macaddress'):
|
||||
raise forms.ValidationError({
|
||||
'mac_address': _('You do not have permission to create MAC addresses.')
|
||||
})
|
||||
parent_field = 'device' if 'device' in self.cleaned_data else 'virtual_machine'
|
||||
if 'tagged_vlans' in self.fields.keys():
|
||||
tagged_vlans = self.cleaned_data.get('tagged_vlans') if self.is_bound else \
|
||||
|
|
@ -96,30 +98,14 @@ class InterfaceCommonForm(forms.Form):
|
|||
def save(self, commit=True):
|
||||
instance = super().save(commit=commit)
|
||||
|
||||
if not commit or 'mac_address' not in self.changed_data:
|
||||
return instance
|
||||
if commit and 'mac_address' in self.changed_data:
|
||||
try:
|
||||
instance.set_primary_mac_address_from_value(self.cleaned_data.get('mac_address'))
|
||||
except ValidationError as e:
|
||||
# Surface a model/custom validation failure (e.g. a MACAddress CustomValidator) as a
|
||||
# clean request abort rather than letting it escape as a 500.
|
||||
raise AbortRequest('; '.join(e.messages))
|
||||
|
||||
mac_address = self.cleaned_data.get('mac_address')
|
||||
|
||||
with transaction.atomic():
|
||||
if mac_address:
|
||||
# Find an existing MACAddress on this interface with the target value, or create one.
|
||||
# Using find-or-create avoids duplicating a MAC that already exists on this interface.
|
||||
mac = instance.mac_addresses.filter(mac_address=mac_address).first()
|
||||
if mac is None:
|
||||
mac = MACAddress(mac_address=mac_address, assigned_object=instance)
|
||||
mac.save()
|
||||
if instance.primary_mac_address_id != mac.pk:
|
||||
instance.snapshot()
|
||||
instance.primary_mac_address = mac
|
||||
instance.save()
|
||||
else:
|
||||
if instance.primary_mac_address_id is not None:
|
||||
instance.snapshot()
|
||||
instance.primary_mac_address = None
|
||||
instance.save()
|
||||
|
||||
instance.__dict__.pop('mac_address', None)
|
||||
return instance
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -5,7 +5,7 @@ from django.contrib.postgres.fields import ArrayField
|
|||
from django.contrib.postgres.indexes import GistIndex
|
||||
from django.core.exceptions import ObjectDoesNotExist, ValidationError
|
||||
from django.core.validators import MaxValueValidator, MinValueValidator
|
||||
from django.db import models
|
||||
from django.db import models, router, transaction
|
||||
from django.utils.translation import gettext_lazy as _
|
||||
|
||||
from dcim.choices import *
|
||||
|
|
@ -882,20 +882,24 @@ class BaseInterface(models.Model):
|
|||
'qinq_svlan': _("Only Q-in-Q interfaces may specify a service VLAN.")
|
||||
})
|
||||
|
||||
# Check that the primary MAC address (if any) is assigned to this interface
|
||||
if (
|
||||
self.primary_mac_address and
|
||||
self.primary_mac_address.assigned_object is not None and
|
||||
self.primary_mac_address.assigned_object != self
|
||||
):
|
||||
raise ValidationError({
|
||||
'primary_mac_address': _(
|
||||
"MAC address {mac_address} is assigned to a different interface ({interface})."
|
||||
).format(
|
||||
mac_address=self.primary_mac_address,
|
||||
interface=self.primary_mac_address.assigned_object,
|
||||
# A primary MAC address must belong to this interface. On create the MAC is assigned by a
|
||||
# post_save signal after this runs, so an as-yet-unassigned MAC is only rejected on update
|
||||
# (self._state.adding is False), where no such signal fires. These are raised as non-field
|
||||
# errors: primary_mac_address is not an InterfaceForm field (it's edited via the mac_address
|
||||
# shortcut), so a field-keyed error would raise in the form's add_error() rather than render.
|
||||
if self.primary_mac_address:
|
||||
if self.primary_mac_address.assigned_object is None:
|
||||
if not self._state.adding:
|
||||
raise ValidationError(
|
||||
_("Only a MAC address assigned to this interface can be its primary MAC address.")
|
||||
)
|
||||
elif self.primary_mac_address.assigned_object != self:
|
||||
raise ValidationError(
|
||||
_("MAC address {mac_address} is assigned to a different interface ({interface}).").format(
|
||||
mac_address=self.primary_mac_address,
|
||||
interface=self.primary_mac_address.assigned_object,
|
||||
)
|
||||
)
|
||||
})
|
||||
|
||||
def save(self, *args, **kwargs):
|
||||
|
||||
|
|
@ -909,6 +913,67 @@ class BaseInterface(models.Model):
|
|||
|
||||
return super().save(*args, **kwargs)
|
||||
|
||||
def set_primary_mac_address(self, mac):
|
||||
"""
|
||||
Set (or clear) this interface's primary MAC address as a single atomic, validated operation.
|
||||
Pass a MACAddress instance to designate it primary, or None to clear the primary MAC. The
|
||||
callers own permission checks; this method owns the validated write. To set from a submitted
|
||||
address string (find-or-create on this interface) use set_primary_mac_address_from_value().
|
||||
"""
|
||||
self._set_primary_mac_address(mac=mac)
|
||||
set_primary_mac_address.alters_data = True
|
||||
|
||||
def set_primary_mac_address_from_value(self, mac_address):
|
||||
"""
|
||||
Set this interface's primary MAC address from a submitted address string, finding an existing
|
||||
MAC on the interface or creating one, all within the operation's locked transaction. An empty
|
||||
value clears the primary MAC. For the form and API adapters, which receive a string.
|
||||
"""
|
||||
self._set_primary_mac_address(mac_value=mac_address or None)
|
||||
set_primary_mac_address_from_value.alters_data = True
|
||||
|
||||
def _set_primary_mac_address(self, mac=None, mac_value=None):
|
||||
"""
|
||||
Shared implementation of the two public setters. Locks this interface's row, resolves a
|
||||
submitted string to a MACAddress (find-or-create, inside the lock so concurrent requests can't
|
||||
both create the same one), validates, and saves. Callers pass either a resolved MACAddress
|
||||
(`mac`) or an address string (`mac_value`), never both.
|
||||
"""
|
||||
with transaction.atomic(using=router.db_for_write(type(self))):
|
||||
# Lock and re-fetch this interface so concurrent set-primary/find-or-create requests
|
||||
# serialize, and mutate the freshly-loaded row rather than the caller's in-memory instance.
|
||||
# The re-fetch resets change-tracking state (e.g. _original_device) to the persisted values,
|
||||
# so full_clean() validates the persisted object plus this one change, not unrelated edits the
|
||||
# adapter already validated and saved.
|
||||
locked = type(self).objects.select_for_update().get(pk=self.pk)
|
||||
|
||||
# Resolve a submitted string to a MAC inside the lock, so two concurrent requests setting the
|
||||
# same new value can't both miss the lookup and both create a duplicate.
|
||||
if mac_value is not None:
|
||||
mac = locked.mac_addresses.filter(mac_address=mac_value).first()
|
||||
if mac is None:
|
||||
mac = locked.mac_addresses.model(mac_address=mac_value, assigned_object=locked)
|
||||
mac.full_clean()
|
||||
mac.save()
|
||||
|
||||
target_id = mac.pk if mac is not None else None
|
||||
if locked.primary_mac_address_id == target_id:
|
||||
self.primary_mac_address = mac
|
||||
self.__dict__.pop('mac_address', None)
|
||||
return
|
||||
|
||||
# Snapshot the locked row (refetched after any adapter save this request) so the changelog
|
||||
# records the correct pre-change state for this MAC change, not an earlier field edit.
|
||||
locked.snapshot()
|
||||
locked.primary_mac_address = mac
|
||||
locked.full_clean(validate_unique=False)
|
||||
locked.save()
|
||||
|
||||
# Reflect the change on the caller's instance (for success messages and API responses) and
|
||||
# invalidate the cached read-side mac_address property.
|
||||
self.primary_mac_address = mac
|
||||
self.__dict__.pop('mac_address', None)
|
||||
|
||||
@property
|
||||
def tunnel_termination(self):
|
||||
return self.tunnel_terminations.first()
|
||||
|
|
|
|||
|
|
@ -1,3 +1,5 @@
|
|||
from urllib.parse import quote
|
||||
|
||||
import django_tables2 as tables
|
||||
from django.middleware.csrf import get_token
|
||||
from django.urls import reverse
|
||||
|
|
@ -1257,17 +1259,36 @@ class MACAddressActionsColumn(columns.ActionsColumn):
|
|||
request = getattr(table, 'context', {}).get('request')
|
||||
if request:
|
||||
url = reverse('dcim:macaddress_set_primary', kwargs={'pk': record.pk})
|
||||
form_li = format_html(
|
||||
'<li><form method="post" action="{}">'
|
||||
'<input type="hidden" name="csrfmiddlewaretoken" value="{}">'
|
||||
'<button type="submit" class="dropdown-item">'
|
||||
'<i class="mdi mdi-star-outline"></i> {}'
|
||||
'</button></form></li>',
|
||||
url, get_token(request), _('Set as primary'),
|
||||
)
|
||||
# Return the user where they came from, the same way the parent's GET actions
|
||||
# (edit/delete/changelog) do. In an embedded panel ObjectsTablePanel injects the parent
|
||||
# object's URL as ?return_url=, so this lands on the interface; on the list view it falls
|
||||
# back to the list path.
|
||||
return_url = request.GET.get('return_url', request.get_full_path())
|
||||
url = f'{url}?return_url={quote(return_url)}'
|
||||
# Embedded tables need their own form; list tables reuse the surrounding bulk form.
|
||||
if getattr(table, 'embedded', False):
|
||||
# No surrounding form: a self-contained POST form is valid and carries its own CSRF token.
|
||||
action_li = format_html(
|
||||
'<li><form method="post" action="{}">'
|
||||
'<input type="hidden" name="csrfmiddlewaretoken" value="{}">'
|
||||
'<button type="submit" class="dropdown-item">'
|
||||
'<i class="mdi mdi-star-outline"></i> {}'
|
||||
'</button></form></li>',
|
||||
url, get_token(request), _('Set as primary'),
|
||||
)
|
||||
else:
|
||||
# Inside the bulk-edit <form>: a nested <form> is invalid HTML and gets dropped by the
|
||||
# parser, so ride the surrounding form via formaction/formmethod instead (matching the
|
||||
# DataSource sync button in core/tables/template_code.py).
|
||||
action_li = format_html(
|
||||
'<li><button type="submit" formaction="{}" formmethod="post" class="dropdown-item">'
|
||||
'<i class="mdi mdi-star-outline"></i> {}'
|
||||
'</button></li>',
|
||||
url, _('Set as primary'),
|
||||
)
|
||||
html_str = str(html)
|
||||
if '</ul>' in html_str:
|
||||
html = mark_safe(html_str.replace('</ul>', str(form_li) + '</ul>', 1))
|
||||
html = mark_safe(html_str.replace('</ul>', str(action_li) + '</ul>', 1))
|
||||
|
||||
return html
|
||||
|
||||
|
|
|
|||
|
|
@ -1,12 +1,14 @@
|
|||
import json
|
||||
|
||||
from django.conf import settings
|
||||
from django.contrib.contenttypes.models import ContentType
|
||||
from django.test import override_settings, tag
|
||||
from django.urls import reverse
|
||||
from django.utils.translation import gettext as _
|
||||
from rest_framework import status
|
||||
|
||||
from core.models import ObjectType
|
||||
from core.choices import ObjectChangeActionChoices
|
||||
from core.models import ObjectChange, ObjectType
|
||||
from dcim.choices import *
|
||||
from dcim.constants import *
|
||||
from dcim.models import *
|
||||
|
|
@ -3838,6 +3840,185 @@ class InterfaceTestCase(Mixins.ComponentTraceMixin, APIViewTestCases.APIViewTest
|
|||
)
|
||||
self.assertEqual(str(child.primary_mac_address.mac_address).upper(), 'AA:BB:CC:DD:EE:02')
|
||||
|
||||
def test_mac_address_conflicts_with_primary_mac_address(self):
|
||||
"""
|
||||
Supplying both the mac_address shortcut and primary_mac_address in one request is rejected
|
||||
rather than letting one silently win.
|
||||
"""
|
||||
self.add_permissions('dcim.change_interface', 'dcim.add_macaddress', 'dcim.change_macaddress')
|
||||
iface = Interface.objects.first()
|
||||
mac = MACAddress.objects.create(mac_address='DD:EE:FF:00:11:22', assigned_object=iface)
|
||||
url = self._get_detail_url(iface)
|
||||
|
||||
response = self.client.patch(
|
||||
url,
|
||||
{'mac_address': 'DD:EE:FF:00:11:33', 'primary_mac_address': {'mac_address': str(mac.mac_address)}},
|
||||
format='json',
|
||||
**self.header
|
||||
)
|
||||
self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
|
||||
|
||||
def test_mac_address_conflicts_with_explicit_null_primary(self):
|
||||
"""
|
||||
The mac_address shortcut alongside an explicit primary_mac_address=null is a conflict (set vs
|
||||
clear) and is rejected, not silently resolved in the shortcut's favor.
|
||||
"""
|
||||
self.add_permissions('dcim.change_interface', 'dcim.add_macaddress')
|
||||
iface = Interface.objects.first()
|
||||
url = self._get_detail_url(iface)
|
||||
|
||||
response = self.client.patch(
|
||||
url,
|
||||
{'mac_address': 'DD:EE:FF:00:11:44', 'primary_mac_address': None},
|
||||
format='json',
|
||||
**self.header
|
||||
)
|
||||
self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
|
||||
|
||||
def test_mac_address_agreeing_with_primary_mac_address_is_accepted(self):
|
||||
"""
|
||||
A read-modify-write round-trip echoes both readable fields with matching values. When the
|
||||
shortcut and primary_mac_address designate the same MAC, the request is accepted.
|
||||
"""
|
||||
self.add_permissions(
|
||||
'dcim.change_interface', 'dcim.add_macaddress', 'dcim.change_macaddress', 'dcim.view_macaddress'
|
||||
)
|
||||
iface = Interface.objects.first()
|
||||
mac = MACAddress.objects.create(mac_address='DD:EE:FF:00:11:66', assigned_object=iface)
|
||||
iface.primary_mac_address = mac
|
||||
iface.save()
|
||||
url = self._get_detail_url(iface)
|
||||
|
||||
response = self.client.patch(
|
||||
url,
|
||||
{
|
||||
'mac_address': str(mac.mac_address),
|
||||
'primary_mac_address': {'mac_address': str(mac.mac_address)},
|
||||
'description': 'round-trip edit',
|
||||
},
|
||||
format='json',
|
||||
**self.header
|
||||
)
|
||||
self.assertHttpStatus(response, status.HTTP_200_OK)
|
||||
iface.refresh_from_db()
|
||||
self.assertEqual(iface.primary_mac_address_id, mac.pk)
|
||||
self.assertEqual(iface.description, 'round-trip edit')
|
||||
|
||||
def test_null_mac_fields_round_trip_accepted(self):
|
||||
"""
|
||||
An interface with no primary MAC round-trips both fields as null; echoing both back is accepted.
|
||||
"""
|
||||
self.add_permissions('dcim.change_interface')
|
||||
iface = Interface.objects.first()
|
||||
iface.primary_mac_address = None
|
||||
iface.save()
|
||||
url = self._get_detail_url(iface)
|
||||
|
||||
response = self.client.patch(
|
||||
url,
|
||||
{'mac_address': None, 'primary_mac_address': None, 'description': 'null round-trip'},
|
||||
format='json',
|
||||
**self.header
|
||||
)
|
||||
self.assertHttpStatus(response, status.HTTP_200_OK)
|
||||
|
||||
def test_combined_update_changelog_does_not_reattribute_other_fields(self):
|
||||
"""
|
||||
A combined PATCH of an unrelated field plus the mac_address shortcut produces two ObjectChange
|
||||
rows (the fields save, then the primary-MAC save). The MAC change's row must record the state
|
||||
after the field save as its prechange, so the unrelated field edit isn't re-reported as part of
|
||||
the MAC change.
|
||||
"""
|
||||
self.add_permissions('dcim.change_interface', 'dcim.add_macaddress')
|
||||
iface = Interface.objects.first()
|
||||
iface.description = 'original'
|
||||
iface.primary_mac_address = None
|
||||
iface.save()
|
||||
url = self._get_detail_url(iface)
|
||||
|
||||
response = self.client.patch(
|
||||
url,
|
||||
{'description': 'updated', 'mac_address': 'AA:BB:CC:DD:EE:10'},
|
||||
format='json',
|
||||
**self.header
|
||||
)
|
||||
self.assertHttpStatus(response, status.HTTP_200_OK)
|
||||
|
||||
changes = ObjectChange.objects.filter(
|
||||
action=ObjectChangeActionChoices.ACTION_UPDATE,
|
||||
changed_object_type=ContentType.objects.get_for_model(Interface),
|
||||
changed_object_id=iface.pk,
|
||||
).order_by('pk')
|
||||
# The MAC change is the row whose postchange records the new primary MAC.
|
||||
mac_change = changes.filter(postchange_data__primary_mac_address__isnull=False).last()
|
||||
self.assertIsNotNone(mac_change)
|
||||
# Its prechange must reflect the already-saved description, so the field edit isn't re-attributed.
|
||||
self.assertEqual(mac_change.prechange_data['description'], 'updated')
|
||||
|
||||
def test_primary_mac_address_must_belong_to_interface(self):
|
||||
"""
|
||||
Setting primary_mac_address (bypassing the shortcut op) to a MAC not assigned to this interface
|
||||
is rejected on update, so the primary MAC can't dangle outside the interface's own MAC set.
|
||||
"""
|
||||
self.add_permissions('dcim.change_interface', 'dcim.change_macaddress')
|
||||
iface = Interface.objects.first()
|
||||
unassigned = MACAddress.objects.create(mac_address='DD:EE:FF:00:11:55')
|
||||
url = self._get_detail_url(iface)
|
||||
|
||||
response = self.client.patch(
|
||||
url,
|
||||
{'primary_mac_address': {'mac_address': str(unassigned.mac_address)}},
|
||||
format='json',
|
||||
**self.header
|
||||
)
|
||||
self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
|
||||
iface.refresh_from_db()
|
||||
self.assertIsNone(iface.primary_mac_address)
|
||||
|
||||
def test_create_with_unassigned_primary_mac_address(self):
|
||||
"""
|
||||
Creating an interface with a nested, as-yet-unassigned primary_mac_address is allowed: the
|
||||
clean() invariant is skipped while adding, and the post_save signal assigns the MAC to the new
|
||||
interface. This pins the create-path carve-out against a future signal regression.
|
||||
"""
|
||||
self.add_permissions(
|
||||
'dcim.add_interface', 'dcim.add_macaddress', 'dcim.change_macaddress', 'dcim.view_macaddress'
|
||||
)
|
||||
device = Device.objects.first()
|
||||
unassigned = MACAddress.objects.create(mac_address='DD:EE:FF:00:11:77')
|
||||
|
||||
data = {
|
||||
'device': device.pk,
|
||||
'name': 'Interface With Primary MAC',
|
||||
'type': '1000base-t',
|
||||
'primary_mac_address': {'mac_address': str(unassigned.mac_address)},
|
||||
}
|
||||
response = self.client.post(self._get_list_url(), data, format='json', **self.header)
|
||||
self.assertHttpStatus(response, status.HTTP_201_CREATED)
|
||||
|
||||
iface = Interface.objects.get(pk=response.data['id'])
|
||||
self.assertEqual(iface.primary_mac_address_id, unassigned.pk)
|
||||
# The signal assigned the MAC to the new interface, so it isn't a dangling primary.
|
||||
unassigned.refresh_from_db()
|
||||
self.assertEqual(unassigned.assigned_object, iface)
|
||||
|
||||
@override_settings(CUSTOM_VALIDATORS={'dcim.macaddress': [{'mac_address': {'regex': '^AA:'}}]})
|
||||
def test_mac_address_custom_validation_returns_400(self):
|
||||
"""
|
||||
A MAC that fails a custom validator on creation returns a 400, not a 500 (the model
|
||||
ValidationError raised inside the serializer is translated to a DRF error).
|
||||
"""
|
||||
self.add_permissions('dcim.add_interface', 'dcim.add_macaddress')
|
||||
device = Device.objects.first()
|
||||
data = {
|
||||
'device': device.pk,
|
||||
'name': 'Interface Custom Validation',
|
||||
'type': '1000base-t',
|
||||
'mac_address': 'BB:CC:DD:EE:FF:00',
|
||||
}
|
||||
response = self.client.post(self._get_list_url(), data, format='json', **self.header)
|
||||
self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
|
||||
|
||||
|
||||
class FrontPortTestCase(APIViewTestCases.APIViewTestCase):
|
||||
model = FrontPort
|
||||
|
|
|
|||
|
|
@ -54,6 +54,67 @@ class MACAddressTestCase(TestCase):
|
|||
self.mac_b.assigned_object = None
|
||||
self.mac_b.clean()
|
||||
|
||||
def test_set_primary_mac_address_assigns(self):
|
||||
self.interface.set_primary_mac_address(self.mac_b)
|
||||
self.interface.refresh_from_db()
|
||||
self.assertEqual(self.interface.primary_mac_address_id, self.mac_b.pk)
|
||||
|
||||
def test_set_primary_mac_address_clears(self):
|
||||
self.interface.set_primary_mac_address(None)
|
||||
self.interface.refresh_from_db()
|
||||
self.assertIsNone(self.interface.primary_mac_address_id)
|
||||
|
||||
def test_set_primary_mac_address_noop_when_already_primary(self):
|
||||
# mac_a is already primary; re-setting it changes nothing and doesn't error.
|
||||
self.interface.set_primary_mac_address(self.mac_a)
|
||||
self.interface.refresh_from_db()
|
||||
self.assertEqual(self.interface.primary_mac_address_id, self.mac_a.pk)
|
||||
|
||||
def test_set_primary_mac_address_from_value_finds_existing(self):
|
||||
# A value already present on the interface is promoted, not duplicated.
|
||||
count_before = self.interface.mac_addresses.count()
|
||||
self.interface.set_primary_mac_address_from_value(str(self.mac_b.mac_address))
|
||||
self.interface.refresh_from_db()
|
||||
self.assertEqual(self.interface.primary_mac_address_id, self.mac_b.pk)
|
||||
self.assertEqual(self.interface.mac_addresses.count(), count_before)
|
||||
|
||||
def test_set_primary_mac_address_from_value_creates(self):
|
||||
count_before = self.interface.mac_addresses.count()
|
||||
self.interface.set_primary_mac_address_from_value('aabbccddeeff')
|
||||
self.interface.refresh_from_db()
|
||||
self.assertEqual(self.interface.mac_addresses.count(), count_before + 1)
|
||||
self.assertEqual(str(self.interface.primary_mac_address.mac_address).lower(), 'aa:bb:cc:dd:ee:ff')
|
||||
|
||||
def test_set_primary_mac_address_rejects_foreign_mac(self):
|
||||
# A MAC assigned to a different interface can't be made primary here.
|
||||
other = Interface.objects.create(
|
||||
device=self.interface.device,
|
||||
name='Interface 2',
|
||||
type=InterfaceTypeChoices.TYPE_1GE_FIXED,
|
||||
)
|
||||
foreign_mac = MACAddress.objects.create(mac_address='ffeeddccbbaa', assigned_object=other)
|
||||
with self.assertRaises(ValidationError):
|
||||
self.interface.set_primary_mac_address(foreign_mac)
|
||||
|
||||
def test_clean_rejects_unassigned_primary_mac_on_update(self):
|
||||
# An existing interface can't point its primary at a MAC that isn't assigned to it.
|
||||
unassigned = MACAddress.objects.create(mac_address='aabbccdd0099')
|
||||
self.interface.primary_mac_address = unassigned
|
||||
with self.assertRaises(ValidationError):
|
||||
self.interface.full_clean()
|
||||
|
||||
def test_clean_allows_unassigned_primary_mac_on_create(self):
|
||||
# On create the MAC is assigned by a post_save signal after clean(), so an as-yet-unassigned
|
||||
# primary MAC must pass validation on a new (adding) instance.
|
||||
mac = MACAddress.objects.create(mac_address='aabbccdd00aa')
|
||||
new_iface = Interface(
|
||||
device=self.interface.device,
|
||||
name='Interface Create Heal',
|
||||
type=InterfaceTypeChoices.TYPE_1GE_FIXED,
|
||||
primary_mac_address=mac,
|
||||
)
|
||||
new_iface.full_clean() # must not raise
|
||||
|
||||
|
||||
class LocationTestCase(TestCase):
|
||||
|
||||
|
|
|
|||
|
|
@ -3,6 +3,7 @@ import datetime
|
|||
import json
|
||||
from decimal import Decimal
|
||||
from io import StringIO
|
||||
from urllib.parse import quote
|
||||
from zoneinfo import ZoneInfo
|
||||
|
||||
import yaml
|
||||
|
|
@ -3789,6 +3790,73 @@ class InterfaceTestCase(ViewTestCases.DeviceComponentViewTestCase):
|
|||
self.assertEqual(wireless_interface.type, InterfaceTypeChoices.TYPE_80211AC)
|
||||
self.assertEqual(wireless_interface.rf_channel_width, Decimal('20.0'))
|
||||
|
||||
@override_settings(EXEMPT_VIEW_PERMISSIONS=['*'], EXEMPT_EXCLUDE_MODELS=[])
|
||||
def test_mac_address_unchanged_edit_without_add_permission(self):
|
||||
"""
|
||||
Editing an unrelated field on an interface with a pre-populated primary MAC, as a user without
|
||||
add_macaddress, succeeds: the untouched pre-populated MAC must not require the create permission.
|
||||
"""
|
||||
self.add_permissions('dcim.change_interface')
|
||||
|
||||
instance = Interface.objects.filter(device_id=self.form_data['device']).first()
|
||||
mac = MACAddress.objects.create(mac_address='AA:BB:CC:DD:EE:FF', assigned_object=instance)
|
||||
instance.primary_mac_address = mac
|
||||
instance.save()
|
||||
|
||||
# Re-submit the current primary MAC unchanged while editing an unrelated field.
|
||||
data = {
|
||||
**self.form_data,
|
||||
'mac_address': 'AA:BB:CC:DD:EE:FF',
|
||||
'description': 'Updated description',
|
||||
'changelog_message': 'test',
|
||||
}
|
||||
response = self.client.post(self._get_url('edit', instance), data=post_data(data))
|
||||
self.assertHttpStatus(response, 302)
|
||||
|
||||
instance.refresh_from_db()
|
||||
self.assertEqual(instance.description, 'Updated description')
|
||||
self.assertEqual(instance.primary_mac_address_id, mac.pk)
|
||||
|
||||
@override_settings(
|
||||
EXEMPT_VIEW_PERMISSIONS=['*'],
|
||||
EXEMPT_EXCLUDE_MODELS=[],
|
||||
CUSTOM_VALIDATORS={'dcim.macaddress': [{'mac_address': {'regex': '^AA:'}}]},
|
||||
)
|
||||
def test_mac_address_shortcut_custom_validation_error(self):
|
||||
"""
|
||||
A MAC that fails a custom validator on the edit form is surfaced as a request error, not a 500.
|
||||
"""
|
||||
self.add_permissions('dcim.change_interface', 'dcim.add_macaddress')
|
||||
|
||||
instance = Interface.objects.filter(device_id=self.form_data['device']).first()
|
||||
|
||||
data = {**self.form_data, 'mac_address': 'BB:CC:DD:EE:FF:00', 'changelog_message': 'test'}
|
||||
response = self.client.post(self._get_url('edit', instance), data=post_data(data))
|
||||
# AbortRequest re-renders the form (200) rather than 500ing; the MAC is not created.
|
||||
self.assertHttpStatus(response, 200)
|
||||
instance.refresh_from_db()
|
||||
self.assertIsNone(instance.primary_mac_address)
|
||||
|
||||
@override_settings(EXEMPT_VIEW_PERMISSIONS=['*'], EXEMPT_EXCLUDE_MODELS=[])
|
||||
def test_edit_interface_with_dangling_primary_mac_does_not_500(self):
|
||||
"""
|
||||
An interface with a primary MAC not assigned to it (reachable via direct ORM) must render a
|
||||
form error on edit, not a 500. The error is non-field because primary_mac_address is not an
|
||||
InterfaceForm field, so a field-keyed error would raise in the form's add_error().
|
||||
"""
|
||||
self.add_permissions('dcim.change_interface')
|
||||
|
||||
instance = Interface.objects.filter(device_id=self.form_data['device']).first()
|
||||
dangling = MACAddress.objects.create(mac_address='AA:BB:CC:DD:EE:AA')
|
||||
instance.primary_mac_address = dangling
|
||||
instance.save()
|
||||
|
||||
data = {**self.form_data, 'description': 'edit attempt', 'changelog_message': 'test'}
|
||||
response = self.client.post(self._get_url('edit', instance), data=post_data(data))
|
||||
# The invalid form re-renders (200) rather than 500ing.
|
||||
self.assertHttpStatus(response, 200)
|
||||
self.assertContains(response, 'Only a MAC address assigned to this interface')
|
||||
|
||||
|
||||
class FrontPortTestCase(ViewTestCases.DeviceComponentViewTestCase):
|
||||
model = FrontPort
|
||||
|
|
@ -5590,6 +5658,150 @@ class MACAddressTestCase(ViewTestCases.PrimaryObjectViewTestCase):
|
|||
mac.assigned_object.refresh_from_db()
|
||||
self.assertIsNone(mac.assigned_object.primary_mac_address)
|
||||
|
||||
def test_set_primary_enforces_object_level_change_permission(self):
|
||||
"""
|
||||
A user with model-level change_interface but a constrained ObjectPermission that excludes the
|
||||
target interface cannot set its primary MAC: object-level constraints are enforced, not just
|
||||
the model-level permission.
|
||||
"""
|
||||
self.add_permissions('dcim.view_macaddress')
|
||||
mac = MACAddress.objects.filter(assigned_object_id__isnull=False).first()
|
||||
interface = mac.assigned_object
|
||||
|
||||
# Grant change_interface constrained to a different interface (id != target), so the target
|
||||
# is invisible to the change-restricted queryset.
|
||||
obj_perm = ObjectPermission(
|
||||
name='Constrained interface change',
|
||||
actions=['change'],
|
||||
constraints={'id__gt': interface.pk},
|
||||
)
|
||||
obj_perm.save()
|
||||
obj_perm.users.add(self.user)
|
||||
obj_perm.object_types.add(ObjectType.objects.get_for_model(Interface))
|
||||
|
||||
url = reverse('dcim:macaddress_set_primary', kwargs={'pk': mac.pk})
|
||||
response = self.client.post(url)
|
||||
|
||||
self.assertHttpStatus(response, 302)
|
||||
self.assertEqual(response['Location'], mac.get_absolute_url())
|
||||
interface.refresh_from_db()
|
||||
self.assertIsNone(interface.primary_mac_address_id)
|
||||
|
||||
def test_set_primary_get_redirects(self):
|
||||
"""
|
||||
A direct GET (bookmark, prefetch) degrades to the MAC's detail page rather than a 405.
|
||||
"""
|
||||
self.add_permissions('dcim.view_macaddress')
|
||||
mac = MACAddress.objects.first()
|
||||
url = reverse('dcim:macaddress_set_primary', kwargs={'pk': mac.pk})
|
||||
response = self.client.get(url)
|
||||
|
||||
self.assertHttpStatus(response, 302)
|
||||
self.assertEqual(response['Location'], mac.get_absolute_url())
|
||||
|
||||
def test_set_primary_anonymous_redirects_to_login(self):
|
||||
"""
|
||||
With LOGIN_REQUIRED, an unauthenticated request is redirected to the login page (via
|
||||
ConditionalLoginRequiredMixin) rather than 404ing or acting, for both GET and POST.
|
||||
"""
|
||||
self.client.logout()
|
||||
mac = MACAddress.objects.first()
|
||||
url = reverse('dcim:macaddress_set_primary', kwargs={'pk': mac.pk})
|
||||
|
||||
with override_settings(LOGIN_REQUIRED=True):
|
||||
for method in (self.client.get, self.client.post):
|
||||
response = method(url)
|
||||
self.assertHttpStatus(response, 302)
|
||||
self.assertTrue(response['Location'].startswith(reverse('login')))
|
||||
|
||||
def test_set_primary_honors_return_url(self):
|
||||
"""
|
||||
With a safe return_url supplied (as the list-view action does), the view redirects there
|
||||
rather than to the interface, so setting a primary MAC from the list keeps the user on it.
|
||||
"""
|
||||
self.add_permissions('dcim.view_macaddress', 'dcim.change_interface')
|
||||
|
||||
mac = MACAddress.objects.first()
|
||||
return_url = reverse('dcim:macaddress_list')
|
||||
url = reverse('dcim:macaddress_set_primary', kwargs={'pk': mac.pk})
|
||||
response = self.client.post(f'{url}?return_url={return_url}')
|
||||
|
||||
self.assertHttpStatus(response, 302)
|
||||
self.assertEqual(response['Location'], return_url)
|
||||
|
||||
@tag('regression') # Issue #18821
|
||||
def test_set_primary_action_list_view_request(self):
|
||||
"""
|
||||
Request-level coverage of the real list-view wiring: GET the MAC list and confirm the
|
||||
table is served inside the bulk-edit <form> with the Set as primary action riding it via
|
||||
formaction, and no nested <form>. This fails if the list view stops wrapping the table in
|
||||
a form or the column's context detection breaks (the class of regression #18821 was).
|
||||
"""
|
||||
self.add_permissions('dcim.view_macaddress')
|
||||
mac = MACAddress.objects.filter(assigned_object_id__isnull=False).first()
|
||||
list_url = reverse('dcim:macaddress_list')
|
||||
set_primary_url = reverse('dcim:macaddress_set_primary', kwargs={'pk': mac.pk})
|
||||
action_url = f'{set_primary_url}?return_url={quote(list_url)}'
|
||||
|
||||
response = self.client.get(list_url)
|
||||
self.assertHttpStatus(response, 200)
|
||||
content = response.content.decode()
|
||||
|
||||
# The action rides the bulk form via a formaction button; it injects no nested <form> of its
|
||||
# own (which the parser would drop, producing the original 405).
|
||||
self.assertInHTML(
|
||||
f'<button type="submit" formaction="{action_url}" formmethod="post" '
|
||||
f'class="dropdown-item"><i class="mdi mdi-star-outline"></i> Set as primary</button>',
|
||||
content,
|
||||
)
|
||||
self.assertNotIn(f'<form method="post" action="{set_primary_url}', content)
|
||||
|
||||
@tag('regression') # Issue #18821
|
||||
def test_set_primary_action_embedded_request(self):
|
||||
"""
|
||||
Request-level coverage of the embedded panel wiring: GET the MAC list as ObjectsTablePanel
|
||||
does (?embedded=True with the parent object's return_url) and confirm the action renders a
|
||||
self-contained <form> (no surrounding form to ride) that returns the user to that object.
|
||||
"""
|
||||
self.add_permissions('dcim.view_macaddress')
|
||||
mac = MACAddress.objects.filter(assigned_object_id__isnull=False).first()
|
||||
interface_url = mac.assigned_object.get_absolute_url()
|
||||
set_primary_url = reverse('dcim:macaddress_set_primary', kwargs={'pk': mac.pk})
|
||||
action_url = f'{set_primary_url}?return_url={quote(interface_url)}'
|
||||
|
||||
response = self.client.get(
|
||||
reverse('dcim:macaddress_list') + f'?embedded=True&return_url={quote(interface_url)}',
|
||||
headers={'hx-request': 'true'},
|
||||
)
|
||||
self.assertHttpStatus(response, 200)
|
||||
content = response.content.decode()
|
||||
|
||||
# A self-contained POST <form> to the returning action URL (valid here, no surrounding form)
|
||||
# wraps the submit button. Assert the button structurally; the form's action carries the
|
||||
# return_url so the user lands back on the interface.
|
||||
self.assertInHTML(
|
||||
'<button type="submit" class="dropdown-item">'
|
||||
'<i class="mdi mdi-star-outline"></i> Set as primary</button>',
|
||||
content,
|
||||
)
|
||||
self.assertIn(f'<form method="post" action="{action_url}">', content)
|
||||
|
||||
@tag('regression') # Issue #18821
|
||||
def test_set_primary_from_embedded_redirects_to_interface(self):
|
||||
"""
|
||||
A set-primary POST with no return_url falls back to the assigned object's detail page, so
|
||||
the action always lands the user on the interface even absent an explicit return target.
|
||||
"""
|
||||
self.add_permissions('dcim.view_macaddress', 'dcim.change_interface')
|
||||
mac = MACAddress.objects.filter(assigned_object_id__isnull=False).first()
|
||||
interface = mac.assigned_object
|
||||
|
||||
url = reverse('dcim:macaddress_set_primary', kwargs={'pk': mac.pk})
|
||||
response = self.client.post(url)
|
||||
|
||||
self.assertHttpStatus(response, 302)
|
||||
self.assertEqual(response['Location'], interface.get_absolute_url())
|
||||
|
||||
@tag('regression') # Issue #20542
|
||||
def test_create_macaddress_via_quickadd(self):
|
||||
"""
|
||||
|
|
|
|||
|
|
@ -1,7 +1,7 @@
|
|||
from django.conf import settings
|
||||
from django.contrib import messages
|
||||
from django.contrib.auth.views import redirect_to_login
|
||||
from django.contrib.contenttypes.models import ContentType
|
||||
from django.core.exceptions import ValidationError
|
||||
from django.core.paginator import EmptyPage, PageNotAnInteger
|
||||
from django.db import router, transaction
|
||||
from django.db.models import Func, IntegerField, Prefetch
|
||||
|
|
@ -41,6 +41,7 @@ from utilities.query import count_related
|
|||
from utilities.query_functions import CollateAsChar
|
||||
from utilities.request import safe_for_redirect
|
||||
from utilities.views import (
|
||||
ConditionalLoginRequiredMixin,
|
||||
GetRelatedModelsMixin,
|
||||
GetReturnURLMixin,
|
||||
ObjectPermissionRequiredMixin,
|
||||
|
|
@ -5855,13 +5856,16 @@ class MACAddressDeleteView(generic.ObjectDeleteView):
|
|||
|
||||
|
||||
@register_model_view(MACAddress, 'set_primary')
|
||||
class MACAddressSetPrimaryView(View):
|
||||
class MACAddressSetPrimaryView(ConditionalLoginRequiredMixin, GetReturnURLMixin, View):
|
||||
queryset = MACAddress.objects.all()
|
||||
|
||||
def post(self, request, pk):
|
||||
if not request.user.is_authenticated:
|
||||
return redirect_to_login(request.get_full_path())
|
||||
def get(self, request, pk):
|
||||
# Degrade a direct GET (bookmark, prefetch) to the MAC's detail page rather than a 405,
|
||||
# matching DataSourceSyncView.
|
||||
mac = get_object_or_404(self.queryset.restrict(request.user, 'view'), pk=pk)
|
||||
return redirect(mac.get_absolute_url())
|
||||
|
||||
def post(self, request, pk):
|
||||
mac = get_object_or_404(self.queryset.restrict(request.user, 'view'), pk=pk)
|
||||
assigned_object = mac.assigned_object
|
||||
|
||||
|
|
@ -5869,26 +5873,28 @@ class MACAddressSetPrimaryView(View):
|
|||
messages.error(request, _('This MAC address is not assigned to an interface.'))
|
||||
return redirect(mac.get_absolute_url())
|
||||
|
||||
perm = get_permission_for_model(assigned_object, 'change')
|
||||
if not request.user.has_perm(perm):
|
||||
# Re-fetch the interface through its change-restricted queryset so object-level permissions
|
||||
# are enforced, not just the model-level change permission.
|
||||
model = assigned_object._meta.model
|
||||
interface = model.objects.restrict(request.user, 'change').filter(pk=assigned_object.pk).first()
|
||||
if interface is None:
|
||||
messages.error(
|
||||
request,
|
||||
_('You do not have permission to modify {object}.').format(object=assigned_object)
|
||||
)
|
||||
return redirect(mac.get_absolute_url())
|
||||
|
||||
if assigned_object.primary_mac_address_id != mac.pk:
|
||||
assigned_object.snapshot()
|
||||
assigned_object.primary_mac_address = mac
|
||||
assigned_object.save()
|
||||
messages.success(
|
||||
request,
|
||||
_('Set {mac} as primary MAC address for {interface}.').format(
|
||||
mac=mac, interface=assigned_object
|
||||
)
|
||||
)
|
||||
try:
|
||||
interface.set_primary_mac_address(mac)
|
||||
except ValidationError as e:
|
||||
messages.error(request, ', '.join(e.messages))
|
||||
return redirect(mac.get_absolute_url())
|
||||
|
||||
return redirect(assigned_object.get_absolute_url())
|
||||
messages.success(
|
||||
request,
|
||||
_('Set {mac} as primary MAC address for {interface}.').format(mac=mac, interface=interface)
|
||||
)
|
||||
return redirect(self.get_return_url(request, interface))
|
||||
|
||||
|
||||
@register_model_view(MACAddress, 'bulk_import', path='import', detail=False)
|
||||
|
|
|
|||
|
|
@ -145,6 +145,7 @@ class VMInterfaceSerializer(MACAddressShortcutMixin, OwnerMixin, NetBoxModelSeri
|
|||
mac_address = _UNSET
|
||||
if isinstance(data, dict):
|
||||
mac_address = data.pop('mac_address', _UNSET)
|
||||
self._validate_no_mac_conflict(data, mac_address)
|
||||
|
||||
if not self.nested and isinstance(data, dict) and mac_address not in (_UNSET, None):
|
||||
try:
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
import logging
|
||||
|
||||
from django.test import tag
|
||||
from django.test import override_settings, tag
|
||||
from django.urls import reverse
|
||||
from netaddr import IPNetwork
|
||||
from rest_framework import status
|
||||
|
|
@ -832,6 +832,82 @@ class VMInterfaceTestCase(APIViewTestCases.APIViewTestCase):
|
|||
self.assertEqual(iface.primary_mac_address.pk, mac2.pk)
|
||||
self.assertEqual(iface.mac_addresses.count(), mac_count_before)
|
||||
|
||||
def test_mac_address_conflicts_with_primary_mac_address(self):
|
||||
"""
|
||||
Supplying both mac_address and primary_mac_address in one request is rejected on VMInterface
|
||||
too (the shared shortcut mixin applies to both interface types).
|
||||
"""
|
||||
from dcim.models import MACAddress
|
||||
|
||||
self.add_permissions(
|
||||
'virtualization.change_vminterface', 'dcim.add_macaddress', 'dcim.change_macaddress'
|
||||
)
|
||||
iface = VMInterface.objects.first()
|
||||
mac = MACAddress.objects.create(mac_address='DD:EE:FF:00:11:22', assigned_object=iface)
|
||||
url = self._get_detail_url(iface)
|
||||
|
||||
response = self.client.patch(
|
||||
url,
|
||||
{'mac_address': 'DD:EE:FF:00:11:33', 'primary_mac_address': {'mac_address': str(mac.mac_address)}},
|
||||
format='json',
|
||||
**self.header
|
||||
)
|
||||
self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
|
||||
|
||||
def test_mac_address_conflicts_with_explicit_null_primary(self):
|
||||
"""
|
||||
The presence-based conflict guard applies to VMInterface too: mac_address shortcut plus an
|
||||
explicit primary_mac_address=null is rejected.
|
||||
"""
|
||||
self.add_permissions('virtualization.change_vminterface', 'dcim.add_macaddress')
|
||||
iface = VMInterface.objects.first()
|
||||
url = self._get_detail_url(iface)
|
||||
|
||||
response = self.client.patch(
|
||||
url,
|
||||
{'mac_address': 'DD:EE:FF:00:11:44', 'primary_mac_address': None},
|
||||
format='json',
|
||||
**self.header
|
||||
)
|
||||
self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
|
||||
|
||||
def test_primary_mac_address_must_belong_to_interface(self):
|
||||
"""
|
||||
Setting primary_mac_address to a MAC not assigned to this VMInterface is rejected on update,
|
||||
so the primary MAC can't dangle outside the interface's own MAC set.
|
||||
"""
|
||||
from dcim.models import MACAddress
|
||||
|
||||
self.add_permissions('virtualization.change_vminterface', 'dcim.change_macaddress')
|
||||
iface = VMInterface.objects.first()
|
||||
unassigned = MACAddress.objects.create(mac_address='DD:EE:FF:00:11:55')
|
||||
url = self._get_detail_url(iface)
|
||||
|
||||
response = self.client.patch(
|
||||
url,
|
||||
{'primary_mac_address': {'mac_address': str(unassigned.mac_address)}},
|
||||
format='json',
|
||||
**self.header
|
||||
)
|
||||
self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
|
||||
iface.refresh_from_db()
|
||||
self.assertIsNone(iface.primary_mac_address)
|
||||
|
||||
@override_settings(CUSTOM_VALIDATORS={'dcim.macaddress': [{'mac_address': {'regex': '^AA:'}}]})
|
||||
def test_mac_address_custom_validation_returns_400(self):
|
||||
"""
|
||||
A MAC that fails a custom validator on VMInterface creation returns 400, not 500.
|
||||
"""
|
||||
self.add_permissions('virtualization.add_vminterface', 'dcim.add_macaddress')
|
||||
vm = VirtualMachine.objects.first()
|
||||
data = {
|
||||
'virtual_machine': vm.pk,
|
||||
'name': 'VMInterface Custom Validation',
|
||||
'mac_address': 'BB:CC:DD:EE:FF:00',
|
||||
}
|
||||
response = self.client.post(self._get_list_url(), data, format='json', **self.header)
|
||||
self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
|
||||
|
||||
|
||||
class VirtualDiskTestCase(APIViewTestCases.APIViewTestCase):
|
||||
model = VirtualDisk
|
||||
|
|
|
|||
Loading…
Reference in New Issue