Преглед на файлове

Reject a duplicate IP address termination in form validation

Model forms skip unique constraints on fields they don't include, so a
second termination of the same L2VPN to the same IP address reached the
database and raised an IntegrityError. Check for it in clean(), as is
already done for interfaces and VLANs.
Brian Tiemann преди 1 ден
родител
ревизия
1aa09fa41b
променени са 3 файла, в които са добавени 43 реда и са изтрити 5 реда
  1. 15 4
      netbox/vpn/models/l2vpn.py
  2. 27 0
      netbox/vpn/tests/test_forms.py
  3. 1 1
      netbox/vpn/tests/test_models.py

+ 15 - 4
netbox/vpn/models/l2vpn.py

@@ -123,12 +123,23 @@ class L2VPNTermination(NetBoxModel):
         return super().__str__()
         return super().__str__()
 
 
     def clean(self):
     def clean(self):
-        # An IP address (e.g. a pseudowire's far-end PE) may terminate many L2VPNs, but other objects only one.
-        if self.assigned_object and not isinstance(self.assigned_object, IPAddress):
+        # Only check is assigned_object is set.  Required otherwise we have an Integrity Error thrown.
+        if self.assigned_object:
             obj_id = self.assigned_object.pk
             obj_id = self.assigned_object.pk
             obj_type = ObjectType.objects.get_for_model(self.assigned_object)
             obj_type = ObjectType.objects.get_for_model(self.assigned_object)
-            terminations = L2VPNTermination.objects.filter(assigned_object_id=obj_id, assigned_object_type=obj_type)
-            if terminations.exclude(pk=self.pk).exists():
+            terminations = L2VPNTermination.objects.filter(
+                assigned_object_id=obj_id, assigned_object_type=obj_type
+            ).exclude(pk=self.pk)
+            # An IP address (e.g. a pseudowire's far-end PE) may terminate many L2VPNs, but each only once. Other
+            # objects may terminate only one L2VPN.
+            if isinstance(self.assigned_object, IPAddress):
+                if terminations.filter(l2vpn_id=self.l2vpn_id).exists():
+                    raise ValidationError(
+                        _('{assigned_object} already terminates {l2vpn}').format(
+                            assigned_object=self.assigned_object, l2vpn=self.l2vpn
+                        )
+                    )
+            elif terminations.exists():
                 raise ValidationError(
                 raise ValidationError(
                     _('L2VPN Termination already assigned ({assigned_object})').format(
                     _('L2VPN Termination already assigned ({assigned_object})').format(
                         assigned_object=self.assigned_object
                         assigned_object=self.assigned_object

+ 27 - 0
netbox/vpn/tests/test_forms.py

@@ -0,0 +1,27 @@
+from django.test import TestCase
+
+from ipam.models import IPAddress
+from vpn.choices import L2VPNTypeChoices
+from vpn.forms import L2VPNTerminationForm
+from vpn.models import L2VPN, L2VPNTermination
+
+
+class L2VPNTerminationFormTestCase(TestCase):
+
+    @classmethod
+    def setUpTestData(cls):
+        cls.ip_address = IPAddress.objects.create(address='192.0.2.1/32')
+        cls.l2vpns = (
+            L2VPN.objects.create(name='L2VPN 1', slug='l2vpn-1', type=L2VPNTypeChoices.TYPE_VPLS),
+            L2VPN.objects.create(name='L2VPN 2', slug='l2vpn-2', type=L2VPNTypeChoices.TYPE_VPLS),
+        )
+        L2VPNTermination.objects.create(l2vpn=cls.l2vpns[0], assigned_object=cls.ip_address)
+
+    def test_ipaddress_terminates_another_l2vpn(self):
+        form = L2VPNTerminationForm(data={'l2vpn': self.l2vpns[1].pk, 'ipaddress': self.ip_address.pk})
+        self.assertTrue(form.is_valid(), form.errors)
+
+    def test_duplicate_ipaddress_termination(self):
+        form = L2VPNTerminationForm(data={'l2vpn': self.l2vpns[0].pk, 'ipaddress': self.ip_address.pk})
+        self.assertFalse(form.is_valid())
+        self.assertIn('already terminates', str(form.errors))

+ 1 - 1
netbox/vpn/tests/test_models.py

@@ -95,7 +95,7 @@ class L2VPNTerminationTestCase(TestCase):
 
 
         L2VPNTermination.objects.create(l2vpn=l2vpn, assigned_object=ip_address)
         L2VPNTermination.objects.create(l2vpn=l2vpn, assigned_object=ip_address)
         duplicate = L2VPNTermination(l2vpn=l2vpn, assigned_object=ip_address)
         duplicate = L2VPNTermination(l2vpn=l2vpn, assigned_object=ip_address)
-        self.assertRaises(ValidationError, duplicate.full_clean)
+        self.assertRaises(ValidationError, duplicate.clean)
 
 
     def test_ipaddress_termination_parent(self):
     def test_ipaddress_termination_parent(self):
         interface = Interface.objects.first()
         interface = Interface.objects.first()