Add device field to L2VPNTermination for assigning vlan terminations to devices

netbox/ipam/api/serializers.py

@@ -516,6 +516,7 @@ class L2VPNSerializer(NetBoxModelSerializer):
 class L2VPNTerminationSerializer(NetBoxModelSerializer):
     url = serializers.HyperlinkedIdentityField(view_name='ipam-api:l2vpntermination-detail')
     l2vpn = NestedL2VPNSerializer()
+    device = NestedDeviceSerializer(required=False)
     assigned_object_type = ContentTypeField(
         queryset=ContentType.objects.all()
     )
@@ -524,7 +525,7 @@ class L2VPNTerminationSerializer(NetBoxModelSerializer):
     class Meta:
         model = L2VPNTermination
         fields = [
-            'id', 'url', 'display', 'l2vpn', 'assigned_object_type', 'assigned_object_id',
+            'id', 'url', 'display', 'l2vpn', 'device', 'assigned_object_type', 'assigned_object_id',
             'assigned_object', 'tags', 'custom_fields', 'created', 'last_updated'
         ]
 

netbox/ipam/filtersets.py

@@ -1131,16 +1131,15 @@ class L2VPNTerminationFilterSet(NetBoxModelFilterSet):
         field_name='pk',
         label=_('Site (ID)'),
     )
-    device = django_filters.ModelMultipleChoiceFilter(
+    device = MultiValueCharFilter(
-        field_name='interface__device__name',
-        queryset=Device.objects.all(),
-        to_field_name='name',
         label=_('Device (name)'),
+        field_name='name',
+        method='filter_device',
     )
-    device_id = django_filters.ModelMultipleChoiceFilter(
+    device_id = MultiValueNumberFilter(
-        field_name='interface__device',
-        queryset=Device.objects.all(),
         label=_('Device (ID)'),
+        field_name='pk',
+        method='filter_device',
     )
     virtual_machine = django_filters.ModelMultipleChoiceFilter(
         field_name='vminterface__virtual_machine__name',
@@ -1227,3 +1226,13 @@ class L2VPNTerminationFilterSet(NetBoxModelFilterSet):
             )
         )
         return qs
+
+    def filter_device(self, queryset, name, value):
+        qs = queryset.filter(
+            Q(
+                Q(**{'device__{}__in'.format(name): value}) |
+                Q(**{'interface__device__{}__in'.format(name): value})
+            )
+        )
+
+        return qs

netbox/ipam/forms/model_forms.py

@@ -794,6 +794,12 @@ class L2VPNTerminationForm(NetBoxModelForm):
         label=_('L2VPN'),
         fetch_trigger='open'
     )
+    device = DynamicModelChoiceField(
+        queryset=Device.objects.all(),
+        required=False,
+        selector=True,
+        label=_('Device')
+    )
     vlan = DynamicModelChoiceField(
         queryset=VLAN.objects.all(),
         required=False,
@@ -815,7 +821,7 @@ class L2VPNTerminationForm(NetBoxModelForm):
 
     class Meta:
         model = L2VPNTermination
-        fields = ('l2vpn', )
+        fields = ('l2vpn', 'device')
 
     def __init__(self, *args, **kwargs):
         instance = kwargs.get('instance')

netbox/ipam/migrations/0068_l2vpn_add_device_for_vlans.py

@@ -0,0 +1,32 @@
+# Generated by Django 4.1.10 on 2023-08-28 20:23
+
+from django.db import migrations, models
+import django.db.models.deletion
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('dcim', '0181_rename_device_role_device_role'),
+        ('ipam', '0067_ipaddress_index_host'),
+    ]
+
+    operations = [
+        migrations.RemoveConstraint(
+            model_name='l2vpntermination',
+            name='ipam_l2vpntermination_assigned_object',
+        ),
+        migrations.AddField(
+            model_name='l2vpntermination',
+            name='device',
+            field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.CASCADE, related_name='l2vpns', to='dcim.device'),
+        ),
+        migrations.AddConstraint(
+            model_name='l2vpntermination',
+            constraint=models.UniqueConstraint(fields=('device', 'assigned_object_type', 'assigned_object_id'), name='ipam_l2vpntermination_device_assigned_object'),
+        ),
+        migrations.AddConstraint(
+            model_name='l2vpntermination',
+            constraint=models.UniqueConstraint(condition=models.Q(('device__isnull', True)), fields=('assigned_object_type', 'assigned_object_id'), name='ipam_l2vpntermination_assigned_object', violation_error_message='This object is already assigned to this l2vpn without a device specified'),
+        ),
+    ]

netbox/ipam/models/l2vpn.py

@@ -85,6 +85,13 @@ class L2VPNTermination(NetBoxModel):
         on_delete=models.CASCADE,
         related_name='terminations'
     )
+    device = models.ForeignKey(
+        to='dcim.Device',
+        on_delete=models.CASCADE,
+        related_name='l2vpns',
+        null=True,
+        blank=True,
+    )
     assigned_object_type = models.ForeignKey(
         to=ContentType,
         limit_choices_to=L2VPN_ASSIGNMENT_MODELS,
@@ -105,9 +112,15 @@ class L2VPNTermination(NetBoxModel):
     class Meta:
         ordering = ('l2vpn',)
         constraints = (
+            models.UniqueConstraint(
+                fields=('device', 'assigned_object_type', 'assigned_object_id'),
+                name='ipam_l2vpntermination_device_assigned_object',
+            ),
             models.UniqueConstraint(
                 fields=('assigned_object_type', 'assigned_object_id'),
-                name='ipam_l2vpntermination_assigned_object'
+                name='ipam_l2vpntermination_assigned_object',
+                condition=models.Q(('device__isnull', True)),
+                violation_error_message=_("This object is already assigned to this l2vpn without a device specified")
             ),
         )
         verbose_name = _('L2VPN termination')
@@ -115,6 +128,8 @@ class L2VPNTermination(NetBoxModel):
 
     def __str__(self):
         if self.pk is not None:
+            if self.device is not None:
+                return f'{self.assigned_object} ({self.device}) <> {self.l2vpn}'
             return f'{self.assigned_object} <> {self.l2vpn}'
         return super().__str__()
 
@@ -126,13 +141,25 @@ class L2VPNTermination(NetBoxModel):
         if self.assigned_object:
             obj_id = self.assigned_object.pk
             obj_type = ContentType.objects.get_for_model(self.assigned_object)
-            if L2VPNTermination.objects.filter(assigned_object_id=obj_id, assigned_object_type=obj_type).\
+            if L2VPNTermination.objects.filter(
-                    exclude(pk=self.pk).count() > 0:
+                    device__isnull=True,
+                    assigned_object_id=obj_id,
+                    assigned_object_type=obj_type
+            ).exclude(pk=self.pk).count() > 0:
                 raise ValidationError(
                     _('L2VPN Termination already assigned ({assigned_object})').format(
                         assigned_object=self.assigned_object
                     )
                 )
+            elif L2VPNTermination.objects.filter(
+                    models.Q(device=self.device, assigned_object_id=obj_id, assigned_object_type=obj_type) |
+                    models.Q(device__isnull=True, assigned_object_id=obj_id, assigned_object_type=obj_type)
+            ).exclude(pk=self.pk).count() > 0:
+                raise ValidationError(
+                    _('L2VPN Termination already assigned to this device ({assigned_object})').format(
+                        assigned_object=self.assigned_object
+                    )
+                )
 
         # Only check if L2VPN is set and is of type P2P
         if hasattr(self, 'l2vpn') and self.l2vpn.type in L2VPNTypeChoices.P2P:

netbox/ipam/tables/l2vpn.py

@@ -55,6 +55,11 @@ class L2VPNTerminationTable(NetBoxTable):
         verbose_name=_('L2VPN'),
         linkify=True
     )
+    device = tables.Column(
+        linkify=True,
+        orderable=False,
+        verbose_name=_('Device')
+    )
     assigned_object_type = columns.ContentTypeColumn(
         verbose_name=_('Object Type')
     )
@@ -63,11 +68,6 @@ class L2VPNTerminationTable(NetBoxTable):
         orderable=False,
         verbose_name=_('Object')
     )
-    assigned_object_parent = tables.Column(
-        linkify=True,
-        orderable=False,
-        verbose_name=_('Object Parent')
-    )
     assigned_object_site = tables.Column(
         linkify=True,
         orderable=False,
@@ -77,9 +77,8 @@ class L2VPNTerminationTable(NetBoxTable):
     class Meta(NetBoxTable.Meta):
         model = L2VPNTermination
         fields = (
-            'pk', 'l2vpn', 'assigned_object_type', 'assigned_object', 'assigned_object_parent', 'assigned_object_site',
+            'pk', 'l2vpn', 'device', 'assigned_object_type', 'assigned_object', 'assigned_object_site', 'actions',
-            'actions',
         )
         default_columns = (
-            'pk', 'l2vpn', 'assigned_object_type', 'assigned_object_parent', 'assigned_object', 'actions',
+            'pk', 'l2vpn', 'device', 'assigned_object_type', 'assigned_object', 'actions',
         )

netbox/ipam/tests/test_api.py

@@ -1090,6 +1090,16 @@ class L2VPNTerminationTest(APIViewTestCases.APIViewTestCase):
 
     @classmethod
     def setUpTestData(cls):
+        site = Site.objects.create(name='Site 1', slug='site-1')
+        manufacturer = Manufacturer.objects.create(name='Manufacturer 1', slug='manufacturer-1')
+        devicetype = DeviceType.objects.create(manufacturer=manufacturer, model='Device Type 1')
+        role = DeviceRole.objects.create(name='Device Role 1', slug='device-role-1')
+
+        devices = (
+            Device(name='Device 1', site=site, device_type=devicetype, role=role),
+            Device(name='Device 2', site=site, device_type=devicetype, role=role),
+        )
+        Device.objects.bulk_create(devices)
 
         vlans = (
             VLAN(name='VLAN 1', vid=651),
@@ -1112,7 +1122,7 @@ class L2VPNTerminationTest(APIViewTestCases.APIViewTestCase):
         l2vpnterminations = (
             L2VPNTermination(l2vpn=l2vpns[0], assigned_object=vlans[0]),
             L2VPNTermination(l2vpn=l2vpns[0], assigned_object=vlans[1]),
-            L2VPNTermination(l2vpn=l2vpns[0], assigned_object=vlans[2])
+            L2VPNTermination(device=devices[0], l2vpn=l2vpns[0], assigned_object=vlans[2])
         )
         L2VPNTermination.objects.bulk_create(l2vpnterminations)
 
@@ -1129,6 +1139,7 @@ class L2VPNTerminationTest(APIViewTestCases.APIViewTestCase):
             },
             {
                 'l2vpn': l2vpns[0].pk,
+                'device': devices[1].pk,
                 'assigned_object_type': 'ipam.vlan',
                 'assigned_object_id': vlans[5].pk,
             },

netbox/ipam/tests/test_filtersets.py

@@ -1710,7 +1710,7 @@ class L2VPNTerminationTestCase(TestCase, ChangeLoggedFilterSetTests):
         l2vpnterminations = (
             L2VPNTermination(l2vpn=l2vpns[0], assigned_object=vlans[0]),
             L2VPNTermination(l2vpn=l2vpns[1], assigned_object=vlans[1]),
-            L2VPNTermination(l2vpn=l2vpns[2], assigned_object=vlans[2]),
+            L2VPNTermination(device=device, l2vpn=l2vpns[2], assigned_object=vlans[2]),
             L2VPNTermination(l2vpn=l2vpns[0], assigned_object=interfaces[0]),
             L2VPNTermination(l2vpn=l2vpns[1], assigned_object=interfaces[1]),
             L2VPNTermination(l2vpn=l2vpns[2], assigned_object=interfaces[2]),
@@ -1758,9 +1758,9 @@ class L2VPNTerminationTestCase(TestCase, ChangeLoggedFilterSetTests):
     def test_device(self):
         device = Device.objects.all().first()
         params = {'device_id': [device.pk]}
-        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 3)
+        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 4)
         params = {'device': ['Device 1']}
-        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 3)
+        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 4)
 
     def test_virtual_machine(self):
         virtual_machine = VirtualMachine.objects.all().first()

netbox/ipam/tests/test_models.py

@@ -1,5 +1,6 @@
 from netaddr import IPNetwork, IPSet
 from django.core.exceptions import ValidationError
+from django.db.utils import IntegrityError
 from django.test import TestCase, override_settings
 
 from dcim.models import Interface, Device, DeviceRole, DeviceType, Manufacturer, Site
@@ -595,6 +596,32 @@ class TestL2VPNTermination(TestCase):
 
         L2VPNTermination.objects.bulk_create(l2vpnterminations)
 
+    def test_vlan_device_creation(self):
+        device = Device.objects.first()
+        vlan = Interface.objects.first()
+        l2vpn = L2VPN.objects.first()
+        termination = L2VPNTermination.objects.first()
+
+        termination.device = device
+        termination.save()
+
+        device = Device.objects.last()
+
+        L2VPNTermination.objects.create(device=device, l2vpn=l2vpn, assigned_object=vlan)
+
+        self.assertEqual(L2VPNTermination.objects.all().count(), 4)
+
+    def test_duplicate_vlan_device_creation(self):
+        device = Device.objects.first()
+        vlan = Interface.objects.first()
+        l2vpn = L2VPN.objects.first()
+
+        L2VPNTermination.objects.create(device=device, l2vpn=l2vpn, assigned_object=vlan)
+
+        duplicate = L2VPNTermination(device=device, l2vpn=l2vpn, assigned_object=vlan)
+        self.assertRaises(ValidationError, duplicate.clean)
+        self.assertRaises(IntegrityError, duplicate.save)
+
     def test_duplicate_interface_terminations(self):
         device = Device.objects.first()
         interface = Interface.objects.filter(device=device).first()
@@ -604,11 +631,12 @@ class TestL2VPNTermination(TestCase):
         duplicate = L2VPNTermination(l2vpn=l2vpn, assigned_object=interface)
 
         self.assertRaises(ValidationError, duplicate.clean)
+        self.assertRaises(IntegrityError, duplicate.save)
 
     def test_duplicate_vlan_terminations(self):
-        vlan = Interface.objects.first()
+        vlan = VLAN.objects.first()
         l2vpn = L2VPN.objects.first()
 
-        L2VPNTermination.objects.create(l2vpn=l2vpn, assigned_object=vlan)
         duplicate = L2VPNTermination(l2vpn=l2vpn, assigned_object=vlan)
         self.assertRaises(ValidationError, duplicate.clean)
+        self.assertRaises(IntegrityError, duplicate.save)

netbox/templates/ipam/l2vpntermination_edit.html

@@ -33,6 +33,7 @@
     <div class="row mb-3">
       <div class="tab-content p-0 border-0">
         <div class="tab-pane {% if not form.initial.interface or form.initial.vminterface %}active{% endif %}" id="vlan" role="tabpanel" aria-labeled-by="vlan_tab">
+          {% render_field form.device %}
           {% render_field form.vlan %}
         </div>
         <div class="tab-pane {% if form.initial.interface %}active{% endif %}" id="interface" role="tabpanel" aria-labeled-by="interface_tab">