Address review feedback on L2VPN IP address terminations

- Extend the termination site, region, device and VM filters to IP addresses assigned to interfaces
- Show a VLAN termination's site, and drop a duplicate branch in assigned_object_parent
- Reject a termination import row with no assigned object instead of failing on insert
- Import L2VPNTerminationType for type checking in the IP address GraphQL type

netbox/ipam/graphql/types.py

@@ -34,7 +34,7 @@ if TYPE_CHECKING:
     )
     from tenancy.graphql.types import TenantType
     from virtualization.graphql.types import ClusterGroupType, ClusterType, VirtualMachineType, VMInterfaceType
-    from vpn.graphql.types import L2VPNType, TunnelTerminationType
+    from vpn.graphql.types import L2VPNTerminationType, L2VPNType, TunnelTerminationType
     from wireless.graphql.types import WirelessLANType
 
 __all__ = (

netbox/vpn/filtersets.py

@@ -417,24 +417,24 @@ class L2VPNTerminationFilterSet(NetBoxModelFilterSet):
         label=_('Site (ID)'),
     )
     device = django_filters.ModelMultipleChoiceFilter(
-        field_name='interface__device__name',
+        method='filter_device',
         queryset=Device.objects.all(),
         to_field_name='name',
         label=_('Device (name)'),
     )
     device_id = django_filters.ModelMultipleChoiceFilter(
-        field_name='interface__device',
+        method='filter_device',
         queryset=Device.objects.all(),
         label=_('Device (ID)'),
     )
     virtual_machine = django_filters.ModelMultipleChoiceFilter(
-        field_name='vminterface__virtual_machine__name',
+        method='filter_virtual_machine',
         queryset=VirtualMachine.objects.all(),
         to_field_name='name',
         label=_('Virtual machine (name)'),
     )
     virtual_machine_id = django_filters.ModelMultipleChoiceFilter(
-        field_name='vminterface__virtual_machine',
+        method='filter_virtual_machine',
         queryset=VirtualMachine.objects.all(),
         label=_('Virtual machine (ID)'),
     )
@@ -514,17 +514,37 @@ class L2VPNTerminationFilterSet(NetBoxModelFilterSet):
             Q(
                 Q(**{'vlan__site__{}__in'.format(name): value}) |
                 Q(**{'interface__device__site__{}__in'.format(name): value}) |
-                Q(**{'vminterface__virtual_machine__site__{}__in'.format(name): value})
+                Q(**{'vminterface__virtual_machine__site__{}__in'.format(name): value}) |
+                Q(**{'ipaddress__interface__device__site__{}__in'.format(name): value}) |
+                Q(**{'ipaddress__vminterface__virtual_machine__site__{}__in'.format(name): value})
             )
         )
         return qs
 
+    def filter_device(self, queryset, name, value):
+        if not value:
+            return queryset
+        return queryset.filter(
+            Q(interface__device__in=value) |
+            Q(ipaddress__interface__device__in=value)
+        )
+
+    def filter_virtual_machine(self, queryset, name, value):
+        if not value:
+            return queryset
+        return queryset.filter(
+            Q(vminterface__virtual_machine__in=value) |
+            Q(ipaddress__vminterface__virtual_machine__in=value)
+        )
+
     def filter_region(self, queryset, name, value):
         qs = queryset.filter(
             Q(
                 Q(**{'vlan__site__region__{}__in'.format(name): value}) |
                 Q(**{'interface__device__site__region__{}__in'.format(name): value}) |
-                Q(**{'vminterface__virtual_machine__site__region__{}__in'.format(name): value})
+                Q(**{'vminterface__virtual_machine__site__region__{}__in'.format(name): value}) |
+                Q(**{'ipaddress__interface__device__site__region__{}__in'.format(name): value}) |
+                Q(**{'ipaddress__vminterface__virtual_machine__site__region__{}__in'.format(name): value})
             )
         )
         return qs

netbox/vpn/forms/bulk_import.py

@@ -351,7 +351,7 @@ class L2VPNTerminationImportForm(NetBoxModelImportForm):
                 self.cleaned_data.get('ipaddress'),
             ) if obj
         ]
-        if not self.instance and not assigned_objects:
+        if not self.instance.pk and not assigned_objects:
             raise ValidationError(_('Each termination must specify an interface, a VLAN, or an IP address.'))
         if len(assigned_objects) > 1:
             raise ValidationError(_('Cannot assign more than one of an interface, a VLAN, and an IP address.'))

netbox/vpn/models/l2vpn.py

@@ -5,7 +5,7 @@ from django.utils.functional import cached_property
 from django.utils.translation import gettext_lazy as _
 
 from core.models import ObjectType
-from ipam.models import IPAddress
+from ipam.models import VLAN, IPAddress
 from netbox.models import NetBoxModel, PrimaryModel
 from netbox.models.features import ContactsMixin
 from vpn.choices import L2VPNStatusChoices, L2VPNTypeChoices
@@ -168,12 +168,12 @@ class L2VPNTermination(NetBoxModel):
             return self.assigned_object.virtual_machine
         if obj_type.model == 'interface':
             return self.assigned_object.device
-        if obj_type.model == 'vminterface':
-            return self.assigned_object.virtual_machine
         return None
 
     @property
     def assigned_object_site(self):
+        if isinstance(self.assigned_object, VLAN):
+            return self.assigned_object.site
         if parent := self.assigned_object_parent:
             return parent.site
         return None

netbox/vpn/tests/test_filtersets.py

@@ -888,8 +888,8 @@ class L2VPNTerminationTestCase(TestCase, ChangeLoggedFilterSetTestMixin):
         L2VPN.objects.bulk_create(l2vpns)
 
         ip_addresses = (
-            IPAddress(address='192.0.2.1/32'),
-            IPAddress(address='192.0.2.2/32'),
+            IPAddress(address='192.0.2.1/32', assigned_object=interfaces[0]),
+            IPAddress(address='192.0.2.2/32', assigned_object=vminterfaces[0]),
             IPAddress(address='192.0.2.3/32'),
         )
         IPAddress.objects.bulk_create(ip_addresses)
@@ -947,20 +947,20 @@ class L2VPNTerminationTestCase(TestCase, ChangeLoggedFilterSetTestMixin):
     def test_site(self):
         site = Site.objects.all().first()
         params = {'site_id': [site.pk]}
-        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 3)
+        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 4)
         params = {'site': ['site-1']}
-        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 3)
+        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 4)
 
     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()
         params = {'virtual_machine_id': [virtual_machine.pk]}
-        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 3)
+        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 4)
         params = {'virtual_machine': ['Virtual Machine 1']}
-        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 3)
+        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 4)

netbox/vpn/tests/test_forms.py

@@ -2,7 +2,7 @@ from django.test import TestCase
 
 from ipam.models import IPAddress
 from vpn.choices import L2VPNTypeChoices
-from vpn.forms import L2VPNTerminationForm
+from vpn.forms import L2VPNTerminationForm, L2VPNTerminationImportForm
 from vpn.models import L2VPN, L2VPNTermination
 
 
@@ -25,3 +25,15 @@ class L2VPNTerminationFormTestCase(TestCase):
         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))
+
+
+class L2VPNTerminationImportFormTestCase(TestCase):
+
+    @classmethod
+    def setUpTestData(cls):
+        L2VPN.objects.create(name='L2VPN 1', slug='l2vpn-1', type=L2VPNTypeChoices.TYPE_VPLS)
+
+    def test_missing_assigned_object(self):
+        form = L2VPNTerminationImportForm(data={'l2vpn': 'L2VPN 1'}, instance=L2VPNTermination())
+        self.assertFalse(form.is_valid())
+        self.assertIn('must specify an interface, a VLAN, or an IP address', str(form.errors))

netbox/vpn/tests/test_models.py

@@ -111,6 +111,13 @@ class L2VPNTerminationTestCase(TestCase):
         self.assertIsNone(termination.assigned_object_parent)
         self.assertIsNone(termination.assigned_object_site)
 
+    def test_vlan_termination_site(self):
+        site = Site.objects.first()
+        vlan = VLAN.objects.create(name='VLAN 8', vid=658, site=site)
+        termination = L2VPNTermination.objects.create(l2vpn=L2VPN.objects.first(), assigned_object=vlan)
+        self.assertIsNone(termination.assigned_object_parent)
+        self.assertEqual(termination.assigned_object_site, site)
+
     def test_delete_ipaddress_deletes_terminations(self):
         ip_address = IPAddress.objects.create(address='192.0.2.1/32')
         termination = L2VPNTermination.objects.create(l2vpn=L2VPN.objects.first(), assigned_object=ip_address)