Address further review feedback on L2VPN IP address terminations

netbox/ipam/graphql/filters.py

@@ -27,7 +27,7 @@ if TYPE_CHECKING:
     from core.graphql.filters import ContentTypeFilter
     from dcim.graphql.filters import SiteFilter
     from netbox.graphql.filter_lookups import BigIntegerLookup, IntegerLookup, IntegerRangeArrayLookup
-    from vpn.graphql.filters import L2VPNFilter
+    from vpn.graphql.filters import L2VPNFilter, L2VPNTerminationFilter
 
     from .enums import *
 
@@ -198,6 +198,9 @@ class IPAddressFilter(ContactFilterMixin, TenancyFilterMixin, PrimaryModelFilter
     )
     nat_outside_id: ID | None = strawberry_django.filter_field()
     dns_name: StrFilterLookup | None = strawberry_django.filter_field()
+    l2vpn_terminations: Annotated['L2VPNTerminationFilter', strawberry.lazy('vpn.graphql.filters')] | None = (
+        strawberry_django.filter_field()
+    )
 
     @strawberry_django.filter_field()
     def assigned(self, value: bool, prefix) -> Q:

netbox/vpn/constants.py

@@ -6,3 +6,8 @@ L2VPN_ASSIGNMENT_MODELS = Q(
     Q(app_label='ipam', model='vlan') |
     Q(app_label='virtualization', model='vminterface')
 )
+
+# Models which may terminate more than one L2VPN (e.g. an IP address which is the far-end PE of several pseudowires)
+L2VPN_MULTI_TERMINATION_MODELS = (
+    'ipam.ipaddress',
+)

netbox/vpn/filtersets.py

@@ -5,6 +5,7 @@ from django.utils.translation import gettext as _
 from core.models import ObjectType
 from dcim.models import Device, Interface
 from ipam.models import VLAN, IPAddress, RouteTarget
+from ipam.utils import parse_inet_addresses
 from netbox.filtersets import NetBoxModelFilterSet, OrganizationalModelFilterSet, PrimaryModelFilterSet
 from tenancy.filtersets import ContactModelFilterSet, TenancyFilterSet
 from utilities.filters import (
@@ -484,10 +485,8 @@ class L2VPNTerminationFilterSet(NetBoxModelFilterSet):
         queryset=VLAN.objects.all(),
         label=_('VLAN (ID)'),
     )
-    ipaddress = django_filters.ModelMultipleChoiceFilter(
-        field_name='ipaddress__address',
-        queryset=IPAddress.objects.all(),
-        to_field_name='address',
+    ipaddress = MultiValueCharFilter(
+        method='filter_ipaddress',
         label=_('IP address'),
     )
     ipaddress_id = django_filters.ModelMultipleChoiceFilter(
@@ -530,6 +529,13 @@ class L2VPNTerminationFilterSet(NetBoxModelFilterSet):
         )
         return qs
 
+    def filter_ipaddress(self, queryset, name, value):
+        # Match an address given without a mask regardless of the IP address's mask. Discard any invalid addresses.
+        value = parse_inet_addresses(value)
+        if not value:
+            return queryset.none()
+        return queryset.filter(ipaddress__address__net_in=value)
+
     def filter_device(self, queryset, name, value):
         if not value:
             return queryset

netbox/vpn/forms/bulk_import.py

@@ -2,7 +2,7 @@ from django.core.exceptions import ValidationError
 from django.utils.translation import gettext_lazy as _
 
 from dcim.models import Device, Interface
-from ipam.models import VLAN, IPAddress
+from ipam.models import VLAN, VRF, IPAddress
 from netbox.forms import NetBoxModelImportForm, OrganizationalModelImportForm, PrimaryModelImportForm
 from tenancy.models import Tenant
 from utilities.forms.fields import CSVChoiceField, CSVModelChoiceField, CSVModelMultipleChoiceField
@@ -289,14 +289,14 @@ class L2VPNTerminationImportForm(NetBoxModelImportForm):
         queryset=Device.objects.all(),
         required=False,
         to_field_name='name',
-        help_text=_('Parent device (for interface)')
+        help_text=_('Parent device (for interface or IP address)')
     )
     virtual_machine = CSVModelChoiceField(
         label=_('Virtual machine'),
         queryset=VirtualMachine.objects.all(),
         required=False,
         to_field_name='name',
-        help_text=_('Parent virtual machine (for interface)')
+        help_text=_('Parent virtual machine (for interface or IP address)')
     )
     interface = CSVModelChoiceField(
         label=_('Interface'),
@@ -319,25 +319,47 @@ class L2VPNTerminationImportForm(NetBoxModelImportForm):
         to_field_name='address',
         help_text=_('Assigned IP address, with mask (e.g. the far-end PE of a pseudowire)')
     )
+    ipaddress_vrf = CSVModelChoiceField(
+        label=_('IP address VRF'),
+        queryset=VRF.objects.all(),
+        required=False,
+        to_field_name='name',
+        help_text=_('VRF of the assigned IP address')
+    )
 
     class Meta:
         model = L2VPNTermination
-        fields = ('l2vpn', 'device', 'virtual_machine', 'interface', 'vlan', 'ipaddress', 'tags')
+        fields = ('l2vpn', 'device', 'virtual_machine', 'interface', 'vlan', 'ipaddress', 'ipaddress_vrf', 'tags')
 
     def __init__(self, data=None, *args, **kwargs):
         super().__init__(data, *args, **kwargs)
 
         if data:
 
-            # Limit interface queryset by device or VM
+            # Limit interface and IP address querysets by device or VM
             if data.get('device'):
                 self.fields['interface'].queryset = Interface.objects.filter(
                     **{f"device__{self.fields['device'].to_field_name}": data['device']}
                 )
+                self.fields['ipaddress'].queryset = IPAddress.objects.filter(
+                    **{f"interface__device__{self.fields['device'].to_field_name}": data['device']}
+                )
             elif data.get('virtual_machine'):
                 self.fields['interface'].queryset = VMInterface.objects.filter(
                     **{f"virtual_machine__{self.fields['virtual_machine'].to_field_name}": data['virtual_machine']}
                 )
+                self.fields['ipaddress'].queryset = IPAddress.objects.filter(
+                    **{
+                        f"vminterface__virtual_machine__{self.fields['virtual_machine'].to_field_name}":
+                            data['virtual_machine']
+                    }
+                )
+
+            # Limit IP address queryset by VRF
+            if data.get('ipaddress_vrf'):
+                self.fields['ipaddress'].queryset = self.fields['ipaddress'].queryset.filter(
+                    **{f"vrf__{self.fields['ipaddress_vrf'].to_field_name}": data['ipaddress_vrf']}
+                )
 
     def clean(self):
         super().clean()

netbox/vpn/models/l2vpn.py

@@ -9,6 +9,7 @@ from ipam.models import VLAN, IPAddress
 from netbox.models import NetBoxModel, PrimaryModel
 from netbox.models.features import ContactsMixin
 from vpn.choices import L2VPNStatusChoices, L2VPNTypeChoices
+from vpn.constants import L2VPN_MULTI_TERMINATION_MODELS
 
 __all__ = (
     'L2VPN',
@@ -130,9 +131,8 @@ class L2VPNTermination(NetBoxModel):
             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):
+            # Some objects may terminate many L2VPNs, but each only once. Others may terminate only one L2VPN.
+            if self.assigned_object._meta.label_lower in L2VPN_MULTI_TERMINATION_MODELS:
                 if terminations.filter(l2vpn_id=self.l2vpn_id).exists():
                     raise ValidationError(
                         _('{assigned_object} already terminates {l2vpn}').format(
@@ -159,16 +159,11 @@ class L2VPNTermination(NetBoxModel):
 
     @property
     def assigned_object_parent(self):
-        obj_type = ObjectType.objects.get_for_model(self.assigned_object)
-        if obj_type.model == 'ipaddress':
-            # The device or virtual machine whose interface the IP address is assigned to, if any
-            interface = self.assigned_object.assigned_object
-            return getattr(interface, 'parent_object', None)
-        if obj_type.model == 'vminterface':
-            return self.assigned_object.virtual_machine
-        if obj_type.model == 'interface':
-            return self.assigned_object.device
-        return None
+        obj = self.assigned_object
+        if isinstance(obj, IPAddress):
+            # The interface the IP address is assigned to, if any
+            obj = obj.assigned_object
+        return getattr(obj, 'parent_object', None)
 
     @property
     def assigned_object_site(self):

netbox/vpn/tests/query_counts.json

@@ -12,7 +12,7 @@
   "l2vpn:api_list_objects": 14,
   "l2vpn:list_objects_with_permission": 16,
   "l2vpntermination:api_list_objects": 15,
-  "l2vpntermination:list_objects_with_permission": 21,
+  "l2vpntermination:list_objects_with_permission": 20,
   "tunnel:api_list_objects": 13,
   "tunnel:list_objects_with_permission": 17,
   "tunnelgroup:api_list_objects": 12,

netbox/vpn/tests/test_api.py

@@ -1,3 +1,5 @@
+import json
+
 from django.urls import reverse
 from rest_framework import status
 
@@ -711,3 +713,17 @@ class L2VPNTerminationTestCase(APIViewTestCases.APIViewTestCase):
         cls.bulk_update_invalid_data = {
             'l2vpn': 99999,
         }
+
+    def test_graphql_ip_address_l2vpn_terminations_filter(self):
+        """IP addresses can be filtered over GraphQL by the L2VPNs they terminate."""
+        self.add_permissions('ipam.view_ipaddress')
+        IPAddress.objects.create(address='192.0.2.2/32')
+        l2vpn = L2VPN.objects.get(name='L2VPN 2')
+        query = (
+            f'{{ ip_address_list(filters: {{l2vpn_terminations: {{l2vpn_id: "{l2vpn.pk}"}}}}) {{ address }} }}'
+        )
+        response = self.client.post(reverse('graphql'), data={'query': query}, format='json', **self.header)
+        self.assertHttpStatus(response, status.HTTP_200_OK)
+        data = json.loads(response.content)
+        self.assertNotIn('errors', data)
+        self.assertEqual([ip['address'] for ip in data['data']['ip_address_list']], ['192.0.2.1/32'])

netbox/vpn/tests/test_filtersets.py

@@ -969,6 +969,10 @@ class L2VPNTerminationTestCase(TestCase, ChangeLoggedFilterSetTestMixin):
         self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
         params = {'ipaddress': ['192.0.2.1/32', '192.0.2.2/32']}
         self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
+        params = {'ipaddress': ['192.0.2.1', '192.0.2.2']}
+        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
+        params = {'ipaddress': ['invalid']}
+        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 0)
 
     def test_vlan_shared_name(self):
         # A VLAN name shared by several groups should resolve to a single filter value

netbox/vpn/tests/test_forms.py

@@ -1,6 +1,8 @@
 from django.test import TestCase
 
-from ipam.models import IPAddress
+from dcim.models import Interface
+from ipam.models import VRF, IPAddress
+from utilities.testing import create_test_device
 from vpn.choices import L2VPNTypeChoices
 from vpn.forms import L2VPNTerminationForm, L2VPNTerminationImportForm
 from vpn.models import L2VPN, L2VPNTermination
@@ -33,7 +35,48 @@ class L2VPNTerminationImportFormTestCase(TestCase):
     def setUpTestData(cls):
         L2VPN.objects.create(name='L2VPN 1', slug='l2vpn-1', type=L2VPNTypeChoices.TYPE_VPLS)
 
+        # The same address exists globally, in a VRF, and on a device's interface in another VRF
+        cls.vrfs = (
+            VRF.objects.create(name='VRF 1'),
+            VRF.objects.create(name='VRF 2'),
+        )
+        cls.device = create_test_device('Device 1')
+        interface = Interface.objects.create(device=cls.device, name='Interface 1', type='1000baset')
+        cls.ip_addresses = (
+            IPAddress.objects.create(address='192.0.2.1/32'),
+            IPAddress.objects.create(address='192.0.2.1/32', vrf=cls.vrfs[0]),
+            IPAddress.objects.create(address='192.0.2.1/32', vrf=cls.vrfs[1], assigned_object=interface),
+        )
+
     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))
+
+    def get_assigned_object(self, data):
+        form = L2VPNTerminationImportForm(data={'l2vpn': 'L2VPN 1', **data}, instance=L2VPNTermination())
+        self.assertTrue(form.is_valid(), form.errors)
+        return form.instance.assigned_object
+
+    def test_ipaddress_ambiguous(self):
+        form = L2VPNTerminationImportForm(
+            data={'l2vpn': 'L2VPN 1', 'ipaddress': '192.0.2.1/32'}, instance=L2VPNTermination()
+        )
+        self.assertFalse(form.is_valid())
+        self.assertIn('ipaddress', form.errors)
+
+    def test_ipaddress_vrf(self):
+        assigned_object = self.get_assigned_object({'ipaddress': '192.0.2.1/32', 'ipaddress_vrf': 'VRF 1'})
+        self.assertEqual(assigned_object, self.ip_addresses[1])
+
+    def test_ipaddress_device(self):
+        assigned_object = self.get_assigned_object({'ipaddress': '192.0.2.1/32', 'device': 'Device 1'})
+        self.assertEqual(assigned_object, self.ip_addresses[2])
+
+    def test_ipaddress_not_on_device(self):
+        form = L2VPNTerminationImportForm(
+            data={'l2vpn': 'L2VPN 1', 'ipaddress': '192.0.2.1/32', 'ipaddress_vrf': 'VRF 1', 'device': 'Device 1'},
+            instance=L2VPNTermination()
+        )
+        self.assertFalse(form.is_valid())
+        self.assertIn('ipaddress', form.errors)

netbox/vpn/tests/test_views.py

@@ -1,7 +1,12 @@
+from django.db import connection
+from django.test.utils import CaptureQueriesContext
+from django.urls import reverse
+
 from dcim.choices import InterfaceTypeChoices
 from dcim.models import Interface
 from ipam.models import VLAN, IPAddress, RouteTarget
-from utilities.testing import ViewTestCases, create_tags, create_test_device
+from utilities.testing import ViewTestCases, create_tags, create_test_device, create_test_virtualmachine
+from virtualization.models import VMInterface
 from vpn.choices import *
 from vpn.models import *
 
@@ -686,6 +691,43 @@ class L2VPNTerminationTestCase(
     def test_bulk_update_objects_with_permission(self):
         pass
 
+    def test_list_objects_query_count(self):
+        """
+        Listing terminations should not need extra queries for each row to display its parent and site.
+        """
+        self.add_permissions('vpn.view_l2vpntermination')
+        self.user.config.set(
+            'tables.L2VPNTerminationTable.columns',
+            ['l2vpn', 'assigned_object_type', 'assigned_object', 'assigned_object_parent', 'assigned_object_site'],
+            commit=True
+        )
+        device = create_test_device('Device 2')
+        virtual_machine = create_test_virtualmachine('Virtual Machine 1')
+
+        def get_queries(count):
+            l2vpn = L2VPN.objects.create(
+                name=f'L2VPN {count}x', slug=f'l2vpn-{count}x', type=L2VPNTypeChoices.TYPE_VPLS
+            )
+            for i in range(count):
+                interface = Interface.objects.create(device=device, name=f'{count}x {i}', type='1000baset')
+                vminterface = VMInterface.objects.create(virtual_machine=virtual_machine, name=f'{count}x {i}')
+                vlan = VLAN.objects.create(name=f'VLAN {count}x {i}', vid=100 * count + i, site=device.site)
+                for obj in (
+                    interface,
+                    vminterface,
+                    vlan,
+                    IPAddress.objects.create(address=f'198.51.{count}.{i}/32', assigned_object=interface),
+                    IPAddress.objects.create(address=f'203.0.{count}.{i}/32', assigned_object=vminterface),
+                ):
+                    L2VPNTermination.objects.create(l2vpn=l2vpn, assigned_object=obj)
+            with CaptureQueriesContext(connection) as queries:
+                response = self.client.get(f"{reverse('vpn:l2vpntermination_list')}?l2vpn_id={l2vpn.pk}")
+            self.assertHttpStatus(response, 200)
+            self.assertContains(response, device.site.name)
+            return len(queries)
+
+        self.assertEqual(get_queries(1), get_queries(3))
+
     #
     # Custom assertions
     #

netbox/vpn/views.py

@@ -1,6 +1,9 @@
+from django.contrib.contenttypes.prefetch import GenericPrefetch
 from django.utils.translation import gettext_lazy as _
 
+from dcim.models import Interface
 from extras.ui.panels import CustomFieldsPanel, TagsPanel
+from ipam.models import VLAN, IPAddress
 from ipam.tables import RouteTargetTable
 from netbox.object_actions import AddObject, BulkDelete, BulkEdit, BulkExport, BulkImport
 from netbox.ui import actions, layout
@@ -14,6 +17,7 @@ from netbox.ui.panels import (
 from netbox.views import generic
 from utilities.query import count_related
 from utilities.views import GetRelatedModelsMixin, register_model_view
+from virtualization.models import VMInterface
 
 from . import filtersets, forms, tables
 from .models import *
@@ -761,7 +765,20 @@ class L2VPNBulkDeleteView(generic.BulkDeleteView):
 
 @register_model_view(L2VPNTermination, 'list', path='', detail=False)
 class L2VPNTerminationListView(generic.ObjectListView):
-    queryset = L2VPNTermination.objects.all()
+    # Fetch each terminating object's parent and site, which the table may display
+    queryset = L2VPNTermination.objects.prefetch_related(
+        GenericPrefetch('assigned_object', [
+            Interface.objects.select_related('device__site'),
+            VMInterface.objects.select_related('virtual_machine__site'),
+            VLAN.objects.select_related('site'),
+            IPAddress.objects.prefetch_related(
+                GenericPrefetch('assigned_object', [
+                    Interface.objects.select_related('device__site'),
+                    VMInterface.objects.select_related('virtual_machine__site'),
+                ])
+            ),
+        ])
+    )
     table = tables.L2VPNTerminationTable
     filterset = filtersets.L2VPNTerminationFilterSet
     filterset_form = forms.L2VPNTerminationFilterForm