#8157 - Final work on L2VPN model

netbox/dcim/api/serializers.py

@@ -10,6 +10,7 @@ from dcim.constants import *
 from dcim.models import *
 from ipam.api.nested_serializers import (
     NestedASNSerializer, NestedIPAddressSerializer, NestedVLANSerializer, NestedVRFSerializer,
+    NestedL2VPNTerminationSerializer,
 )
 from ipam.models import ASN, VLAN
 from netbox.api import ChoiceField, ContentTypeField, SerializedPKRelatedField
@@ -823,6 +824,7 @@ class InterfaceSerializer(NetBoxModelSerializer, LinkTerminationSerializer, Conn
         many=True
     )
     vrf = NestedVRFSerializer(required=False, allow_null=True)
+    l2vpn_termination = NestedL2VPNTerminationSerializer(read_only=True)
     cable = NestedCableSerializer(read_only=True)
     wireless_link = NestedWirelessLinkSerializer(read_only=True)
     wireless_lans = SerializedPKRelatedField(
@@ -841,7 +843,7 @@ class InterfaceSerializer(NetBoxModelSerializer, LinkTerminationSerializer, Conn
             'mtu', 'mac_address', 'speed', 'duplex', 'wwn', 'mgmt_only', 'description', 'mode', 'rf_role', 'rf_channel',
             'poe_mode', 'poe_type', 'rf_channel_frequency', 'rf_channel_width', 'tx_power', 'untagged_vlan',
             'tagged_vlans', 'mark_connected', 'cable', 'wireless_link', 'link_peer', 'link_peer_type', 'wireless_lans',
-            'vrf', 'connected_endpoint', 'connected_endpoint_type', 'connected_endpoint_reachable', 'tags',
+            'vrf', 'l2vpn_termination', 'connected_endpoint', 'connected_endpoint_type', 'connected_endpoint_reachable', 'tags',
             'custom_fields', 'created', 'last_updated', 'count_ipaddresses', 'count_fhrp_groups', '_occupied',
         ]
 

netbox/dcim/models/device_components.py

@@ -649,10 +649,11 @@ class Interface(ModularComponentModel, BaseInterface, LinkTermination, PathEndpo
         object_id_field='interface_id',
         related_query_name='+'
     )
-    l2vpn = GenericRelation(
+    l2vpn_terminations = GenericRelation(
         to='ipam.L2VPNTermination',
         content_type_field='assigned_object_type',
         object_id_field='assigned_object_id',
+        related_query_name='interface',
     )
 
     clone_fields = ['device', 'parent', 'bridge', 'lag', 'type', 'mgmt_only', 'poe_mode', 'poe_type']
@@ -828,6 +829,10 @@ class Interface(ModularComponentModel, BaseInterface, LinkTermination, PathEndpo
     def link(self):
         return self.cable or self.wireless_link
 
+    @property
+    def l2vpn_termination(self):
+        return self.l2vpn_terminations.first()
+
 
 #
 # Pass-through ports

netbox/ipam/api/nested_serializers.py

@@ -11,6 +11,8 @@ __all__ = [
     'NestedFHRPGroupAssignmentSerializer',
     'NestedIPAddressSerializer',
     'NestedIPRangeSerializer',
+    'NestedL2VPNSerializer',
+    'NestedL2VPNTerminationSerializer',
     'NestedPrefixSerializer',
     'NestedRIRSerializer',
     'NestedRoleSerializer',
@@ -203,17 +205,17 @@ class NestedL2VPNSerializer(WritableNestedSerializer):
     class Meta:
         model = L2VPN
         fields = [
-            'id', 'url', 'display', 'name', 'type'
+            'id', 'url', 'display', 'identifier', 'name', 'slug', 'type'
         ]
 
 
 class NestedL2VPNTerminationSerializer(WritableNestedSerializer):
-    url = serializers.HyperlinkedIdentityField(view_name='ipam-api:l2vpn_termination-detail')
+    url = serializers.HyperlinkedIdentityField(view_name='ipam-api:l2vpntermination-detail')
     l2vpn = NestedL2VPNSerializer()
 
     class Meta:
         model = L2VPNTermination
         fields = [
-            'id', 'url', 'display', 'l2vpn', 'assigned_object'
+            'id', 'url', 'display', 'l2vpn'
         ]
 

netbox/ipam/api/serializers.py

@@ -207,13 +207,14 @@ class VLANSerializer(NetBoxModelSerializer):
     tenant = NestedTenantSerializer(required=False, allow_null=True)
     status = ChoiceField(choices=VLANStatusChoices, required=False)
     role = NestedRoleSerializer(required=False, allow_null=True)
+    l2vpn_termination = NestedL2VPNTerminationSerializer(read_only=True)
     prefix_count = serializers.IntegerField(read_only=True)
 
     class Meta:
         model = VLAN
         fields = [
-            'id', 'url', 'display', 'site', 'group', 'vid', 'name', 'tenant', 'status', 'role', 'description', 'tags',
-            'custom_fields', 'created', 'last_updated', 'prefix_count',
+            'id', 'url', 'display', 'site', 'group', 'vid', 'name', 'tenant', 'status', 'role', 'description',
+            'l2vpn_termination', 'tags', 'custom_fields', 'created', 'last_updated', 'prefix_count',
         ]
 
 

netbox/ipam/api/views.py

@@ -165,7 +165,7 @@ class L2VPNViewSet(NetBoxModelViewSet):
 
 
 class L2VPNTerminationViewSet(NetBoxModelViewSet):
-    queryset = L2VPNTermination.objects
+    queryset = L2VPNTermination.objects.prefetch_related('assigned_object')
     serializer_class = serializers.L2VPNTerminationSerializer
     filterset_class = filtersets.L2VPNTerminationFilterSet
 

netbox/ipam/filtersets.py

@@ -957,7 +957,7 @@ class L2VPNFilterSet(NetBoxModelFilterSet, TenancyFilterSet):
 
     class Meta:
         model = L2VPN
-        fields = ['identifier', 'name', 'type', 'description']
+        fields = ['id', 'identifier', 'name', 'type', 'description']
 
     def search(self, queryset, name, value):
         if not value.strip():
@@ -977,13 +977,60 @@ class L2VPNTerminationFilterSet(NetBoxModelFilterSet):
         to_field_name='name',
         label='L2VPN (name)',
     )
+    device = MultiValueCharFilter(
+        method='filter_device',
+        field_name='name',
+        label='Device (name)',
+    )
+    device_id = MultiValueNumberFilter(
+        method='filter_device',
+        field_name='pk',
+        label='Device (ID)',
+    )
+    interface = django_filters.ModelMultipleChoiceFilter(
+        field_name='interface__name',
+        queryset=Interface.objects.all(),
+        to_field_name='name',
+        label='Interface (name)',
+    )
+    interface_id = django_filters.ModelMultipleChoiceFilter(
+        field_name='interface',
+        queryset=Interface.objects.all(),
+        label='Interface (ID)',
+    )
+    vlan = django_filters.ModelMultipleChoiceFilter(
+        field_name='vlan__name',
+        queryset=VLAN.objects.all(),
+        to_field_name='name',
+        label='VLAN (name)',
+    )
+    vlan_vid = django_filters.NumberFilter(
+        field_name='vlan__vid',
+        label='VLAN number (1-4094)',
+    )
+    vlan_id = django_filters.ModelMultipleChoiceFilter(
+        field_name='vlan',
+        queryset=VLAN.objects.all(),
+        label='VLAN (ID)',
+    )
 
     class Meta:
         model = L2VPNTermination
-        fields = ['l2vpn']
+        fields = ['id', ]
 
     def search(self, queryset, name, value):
         if not value.strip():
             return queryset
         qs_filter = Q(l2vpn__name__icontains=value)
         return queryset.filter(qs_filter)
+
+    def filter_device(self, queryset, name, value):
+        devices = Device.objects.filter(**{'{}__in'.format(name): value})
+        if not devices.exists():
+            return queryset.none()
+        interface_ids = []
+        for device in devices:
+            interface_ids.extend(device.vc_interfaces().values_list('id', flat=True))
+        return queryset.filter(
+            interface__in=interface_ids
+        )

netbox/ipam/forms/bulk_edit.py

@@ -19,6 +19,7 @@ __all__ = (
     'IPAddressBulkEditForm',
     'IPRangeBulkEditForm',
     'L2VPNBulkEditForm',
+    'L2VPNTerminationBulkEditForm',
     'PrefixBulkEditForm',
     'RIRBulkEditForm',
     'RoleBulkEditForm',
@@ -458,3 +459,7 @@ class L2VPNBulkEditForm(NetBoxModelBulkEditForm):
         (None, ('tenant', 'description')),
     )
     nullable_fields = ('tenant', 'description',)
+
+
+class L2VPNTerminationBulkEditForm(NetBoxModelBulkEditForm):
+    model = L2VPN

netbox/ipam/graphql/schema.py

@@ -17,6 +17,12 @@ class IPAMQuery(graphene.ObjectType):
     ip_range = ObjectField(IPRangeType)
     ip_range_list = ObjectListField(IPRangeType)
 
+    l2vpn = ObjectField(L2VPNType)
+    l2vpn_list = ObjectListField(L2VPNType)
+
+    l2vpn_termination = ObjectField(L2VPNTerminationType)
+    l2vpn_termination_list = ObjectListField(L2VPNTerminationType)
+
     prefix = ObjectField(PrefixType)
     prefix_list = ObjectListField(PrefixType)
 

netbox/ipam/graphql/types.py

@@ -11,6 +11,8 @@ __all__ = (
     'FHRPGroupAssignmentType',
     'IPAddressType',
     'IPRangeType',
+    'L2VPNType',
+    'L2VPNTerminationType',
     'PrefixType',
     'RIRType',
     'RoleType',
@@ -151,3 +153,17 @@ class VRFType(NetBoxObjectType):
         model = models.VRF
         fields = '__all__'
         filterset_class = filtersets.VRFFilterSet
+
+
+class L2VPNType(NetBoxObjectType):
+    class Meta:
+        model = models.L2VPN
+        fields = '__all__'
+        filtersets_class = filtersets.L2VPNFilterSet
+
+
+class L2VPNTerminationType(NetBoxObjectType):
+    class Meta:
+        model = models.L2VPNTermination
+        fields = '__all__'
+        filtersets_class = filtersets.L2VPNTerminationFilterSet

netbox/ipam/models/vlans.py

@@ -174,10 +174,11 @@ class VLAN(NetBoxModel):
         blank=True
     )
 
-    l2vpn = GenericRelation(
+    l2vpn_terminations = GenericRelation(
         to='ipam.L2VPNTermination',
         content_type_field='assigned_object_type',
         object_id_field='assigned_object_id',
+        related_query_name='vlan'
     )
 
     objects = VLANQuerySet.as_manager()
@@ -234,3 +235,7 @@ class VLAN(NetBoxModel):
             Q(untagged_vlan_id=self.pk) |
             Q(tagged_vlans=self.pk)
         ).distinct()
+
+    @property
+    def l2vpn_termination(self):
+        return self.l2vpn_terminations.first()

netbox/ipam/tests/test_api.py

@@ -947,28 +947,28 @@ class L2VPNTest(APIViewTestCases.APIViewTestCase):
     def setUpTestData(cls):
 
         l2vpns = (
-            L2VPN(name='L2VPN 1', type='vxlan', identifier=650001),
-            L2VPN(name='L2VPN 2', type='vpws', identifier=650002),
-            L2VPN(name='L2VPN 3', type='vpls'),  # No RD
+            L2VPN(name='L2VPN 1', slug='l2vpn-1', type='vxlan', identifier=650001),
+            L2VPN(name='L2VPN 2', slug='l2vpn-2', type='vpws', identifier=650002),
+            L2VPN(name='L2VPN 3', slug='l2vpn-3', type='vpls'),  # No RD
         )
         L2VPN.objects.bulk_create(l2vpns)
 
 
 class L2VPNTerminationTest(APIViewTestCases.APIViewTestCase):
     model = L2VPNTermination
-    brief_fields = ['display', 'id', 'l2vpn', 'assigned_object', 'assigned_object_id', 'assigned_object_type', 'url']
+    brief_fields = ['display', 'id', 'l2vpn', 'url']
 
     @classmethod
     def setUpTestData(cls):
 
         vlans = (
-            VLAN(name='VLAN 1', vid=650001),
-            VLAN(name='VLAN 2', vid=650002),
-            VLAN(name='VLAN 3', vid=650003),
-            VLAN(name='VLAN 4', vid=650004),
-            VLAN(name='VLAN 5', vid=650005),
-            VLAN(name='VLAN 6', vid=650006),
-            VLAN(name='VLAN 7', vid=650007)
+            VLAN(name='VLAN 1', vid=651),
+            VLAN(name='VLAN 2', vid=652),
+            VLAN(name='VLAN 3', vid=653),
+            VLAN(name='VLAN 4', vid=654),
+            VLAN(name='VLAN 5', vid=655),
+            VLAN(name='VLAN 6', vid=656),
+            VLAN(name='VLAN 7', vid=657)
         )
 
         VLAN.objects.bulk_create(vlans)
@@ -986,24 +986,26 @@ class L2VPNTerminationTest(APIViewTestCases.APIViewTestCase):
             L2VPNTermination(l2vpn=l2vpns[0], assigned_object=vlans[2])
         )
 
+        L2VPNTermination.objects.bulk_create(l2vpnterminations)
+
         cls.create_data = [
             {
-                'l2vpn': l2vpns[0],
+                'l2vpn': l2vpns[0].pk,
                 'assigned_object_type': 'ipam.vlan',
-                'assigned_object_id': vlans[3],
+                'assigned_object_id': vlans[3].pk,
             },
             {
-                'l2vpn': l2vpns[0],
+                'l2vpn': l2vpns[0].pk,
                 'assigned_object_type': 'ipam.vlan',
-                'assigned_object_id': vlans[4],
+                'assigned_object_id': vlans[4].pk,
             },
             {
-                'l2vpn': l2vpns[0],
+                'l2vpn': l2vpns[0].pk,
                 'assigned_object_type': 'ipam.vlan',
-                'assigned_object_id': vlans[5],
+                'assigned_object_id': vlans[5].pk,
             },
         ]
 
         cls.bulk_update_data = {
-            'l2vpn': l2vpns[2]
+            'l2vpn': l2vpns[2].pk
         }

netbox/ipam/tests/test_filtersets.py

@@ -1465,8 +1465,7 @@ class ServiceTestCase(TestCase, ChangeLoggedFilterSetTests):
         self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
 
 
-class L2VPNTest(TestCase, ChangeLoggedFilterSetTests):
-    # TODO: L2VPN Tests
+class L2VPNTestCase(TestCase, ChangeLoggedFilterSetTests):
     queryset = L2VPN.objects.all()
     filterset = L2VPNFilterSet
 
@@ -1480,20 +1479,8 @@ class L2VPNTest(TestCase, ChangeLoggedFilterSetTests):
         )
         L2VPN.objects.bulk_create(l2vpns)
 
-    def test_created(self):
-        from datetime import date, date
-        pk_list = self.queryset.values_list('pk', flat=True)[:2]
-        print(pk_list)
-        self.queryset.filter(pk__in=pk_list).update(created=datetime(2021, 1, 1, 0, 0, 0, tzinfo=timezone.utc))
-        params = {'created': '2021-01-01T00:00:00'}
-        fs = self.filterset({}, self.queryset).qs.all()
-        for res in fs:
-            print(f'{res.name}:{res.created}')
-        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
-
 
-class L2VPNTerminationTest(TestCase, ChangeLoggedFilterSetTests):
-    # TODO: L2VPN Termination Tests
+class L2VPNTerminationTestCase(TestCase, ChangeLoggedFilterSetTests):
     queryset = L2VPNTermination.objects.all()
     filterset = L2VPNTerminationFilterSet
 
@@ -1511,22 +1498,24 @@ class L2VPNTerminationTest(TestCase, ChangeLoggedFilterSetTests):
             device_role=device_role,
             status='active'
         )
-        interfaces = Interface.objects.bulk_create(
-            Interface(name='GigabitEthernet1/0/1', device=device, type='1000baset'),
-            Interface(name='GigabitEthernet1/0/2', device=device, type='1000baset'),
-            Interface(name='GigabitEthernet1/0/3', device=device, type='1000baset'),
-            Interface(name='GigabitEthernet1/0/4', device=device, type='1000baset'),
-            Interface(name='GigabitEthernet1/0/5', device=device, type='1000baset'),
+
+        interfaces = (
+            Interface(name='Interface 1', device=device, type='1000baset'),
+            Interface(name='Interface 2', device=device, type='1000baset'),
+            Interface(name='Interface 3', device=device, type='1000baset'),
+            Interface(name='Interface 4', device=device, type='1000baset'),
+            Interface(name='Interface 5', device=device, type='1000baset'),
+            Interface(name='Interface 6', device=device, type='1000baset')
         )
 
+        Interface.objects.bulk_create(interfaces)
+
         vlans = (
-            VLAN(name='VLAN 1', vid=650001),
-            VLAN(name='VLAN 2', vid=650002),
-            VLAN(name='VLAN 3', vid=650003),
-            VLAN(name='VLAN 4', vid=650004),
-            VLAN(name='VLAN 5', vid=650005),
-            VLAN(name='VLAN 6', vid=650006),
-            VLAN(name='VLAN 7', vid=650007)
+            VLAN(name='VLAN 1', vid=651),
+            VLAN(name='VLAN 2', vid=652),
+            VLAN(name='VLAN 3', vid=653),
+            VLAN(name='VLAN 4', vid=654),
+            VLAN(name='VLAN 5', vid=655)
         )
 
         VLAN.objects.bulk_create(vlans)
@@ -1534,26 +1523,33 @@ class L2VPNTerminationTest(TestCase, ChangeLoggedFilterSetTests):
         l2vpns = (
             L2VPN(name='L2VPN 1', type='vxlan', identifier=650001),
             L2VPN(name='L2VPN 2', type='vpws', identifier=650002),
-            L2VPN(name='L2VPN 3', type='vpls'),  # No RD
+            L2VPN(name='L2VPN 3', type='vpls'),  # No RD,
         )
         L2VPN.objects.bulk_create(l2vpns)
 
         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(l2vpn=l2vpns[1], assigned_object=vlans[1]),
+            L2VPNTermination(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]),
         )
 
+        L2VPNTermination.objects.bulk_create(l2vpnterminations)
+
     def test_l2vpns(self):
         l2vpns = L2VPN.objects.all()[:2]
         params = {'l2vpn_id': [l2vpns[0].pk, l2vpns[1].pk]}
-        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
+        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 4)
         params = {'l2vpn': ['L2VPN 1', 'L2VPN 2']}
-        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
+        self.assertEqual(self.filterset(params, self.queryset).qs.count(), 4)
 
     def test_interfaces(self):
         interfaces = Interface.objects.all()[:2]
         params = {'interface_id': [interfaces[0].pk, interfaces[1].pk]}
+        qs = self.filterset(params, self.queryset).qs
+        results = qs.all()
         self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
         params = {'interface': ['Interface 1', 'Interface 2']}
         self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)

netbox/ipam/tests/test_models.py

@@ -2,8 +2,9 @@ from netaddr import IPNetwork, IPSet
 from django.core.exceptions import ValidationError
 from django.test import TestCase, override_settings
 
+from dcim.models import Interface, Device, DeviceRole, DeviceType, Manufacturer, Site
 from ipam.choices import IPAddressRoleChoices, PrefixStatusChoices
-from ipam.models import Aggregate, IPAddress, IPRange, Prefix, RIR, VLAN, VLANGroup, VRF
+from ipam.models import Aggregate, IPAddress, IPRange, Prefix, RIR, VLAN, VLANGroup, VRF, L2VPN, L2VPNTermination
 
 
 class TestAggregate(TestCase):
@@ -540,11 +541,75 @@ class TestVLANGroup(TestCase):
         self.assertEqual(vlangroup.get_next_available_vid(), 105)
 
 
-class TestL2VPN(TestCase):
-    # TODO: L2VPN Tests
-    pass
+class TestL2VPNTermination(TestCase):
+
+    @classmethod
+    def setUpTestData(cls):
 
+        site = Site.objects.create(name='Site 1')
+        manufacturer = Manufacturer.objects.create(name='Manufacturer 1')
+        device_type = DeviceType.objects.create(model='Device Type 1', manufacturer=manufacturer)
+        device_role = DeviceRole.objects.create(name='Switch')
+        device = Device.objects.create(
+            name='Device 1',
+            site=site,
+            device_type=device_type,
+            device_role=device_role,
+            status='active'
+        )
+
+        interfaces = (
+            Interface(name='Interface 1', device=device, type='1000baset'),
+            Interface(name='Interface 2', device=device, type='1000baset'),
+            Interface(name='Interface 3', device=device, type='1000baset'),
+            Interface(name='Interface 4', device=device, type='1000baset'),
+            Interface(name='Interface 5', device=device, type='1000baset'),
+        )
+
+        Interface.objects.bulk_create(interfaces)
+
+        vlans = (
+            VLAN(name='VLAN 1', vid=651),
+            VLAN(name='VLAN 2', vid=652),
+            VLAN(name='VLAN 3', vid=653),
+            VLAN(name='VLAN 4', vid=654),
+            VLAN(name='VLAN 5', vid=655),
+            VLAN(name='VLAN 6', vid=656),
+            VLAN(name='VLAN 7', vid=657)
+        )
+
+        VLAN.objects.bulk_create(vlans)
+
+        l2vpns = (
+            L2VPN(name='L2VPN 1', type='vxlan', identifier=650001),
+            L2VPN(name='L2VPN 2', type='vpws', identifier=650002),
+            L2VPN(name='L2VPN 3', type='vpls'),  # No RD
+        )
+        L2VPN.objects.bulk_create(l2vpns)
+
+        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.objects.bulk_create(l2vpnterminations)
+
+    def test_duplicate_interface_terminations(self):
+        device = Device.objects.first()
+        interface = Interface.objects.filter(device=device).first()
+        l2vpn = L2VPN.objects.first()
+
+        L2VPNTermination.objects.create(l2vpn=l2vpn, assigned_object=interface)
+        duplicate = L2VPNTermination(l2vpn=l2vpn, assigned_object=interface)
+
+        self.assertRaises(ValidationError, duplicate.clean)
+
+    def test_duplicate_vlan_terminations(self):
+        vlan = Interface.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)
 
-class TestL2VPNTermination(TestCase):
-    # TODO: L2VPN Termination Tests
-    pass

netbox/ipam/tests/test_views.py

@@ -1,14 +1,18 @@
 import datetime
 
+from django.contrib.contenttypes.models import ContentType
 from django.test import override_settings
 from django.urls import reverse
 from netaddr import IPNetwork
 
-from dcim.models import Device, DeviceRole, DeviceType, Manufacturer, Site
+from dcim.models import Device, DeviceRole, DeviceType, Manufacturer, Site, Interface
+from extras.choices import ObjectChangeActionChoices
+from extras.models import ObjectChange
 from ipam.choices import *
 from ipam.models import *
 from tenancy.models import Tenant
-from utilities.testing import ViewTestCases, create_tags
+from users.models import ObjectPermission
+from utilities.testing import ViewTestCases, create_tags, post_data
 
 
 class ASNTestCase(ViewTestCases.PrimaryObjectViewTestCase):
@@ -749,10 +753,130 @@ class ServiceTestCase(ViewTestCases.PrimaryObjectViewTestCase):
 
 
 class L2VPNTestCase(ViewTestCases.PrimaryObjectViewTestCase):
-    # TODO: L2VPN Tests
-    pass
+    model = L2VPN
+    csv_data = (
+        'name,slug,type,identifier',
+        'L2VPN 5,l2vpn-5,vxlan,456',
+        'L2VPN 6,l2vpn-6,vxlan,444',
+    )
+    bulk_edit_data = {
+        'description': 'New Description',
+    }
 
+    @classmethod
+    def setUpTestData(cls):
+        rts = (
+            RouteTarget(name='64534:123'),
+            RouteTarget(name='64534:321')
+        )
+        RouteTarget.objects.bulk_create(rts)
+
+        l2vpns = (
+            L2VPN(name='L2VPN 1', slug='l2vpn-1', type='vxlan', identifier='650001'),
+            L2VPN(name='L2VPN 2', slug='l2vpn-2', type='vxlan', identifier='650002'),
+            L2VPN(name='L2VPN 3', slug='l2vpn-3', type='vxlan', identifier='650003')
+        )
+
+        L2VPN.objects.bulk_create(l2vpns)
+
+        cls.form_data = {
+            'name': 'L2VPN 8',
+            'slug': 'l2vpn-8',
+            'type': 'vxlan',
+            'identifier': 123,
+            'description': 'Description',
+            'import_targets': [rts[0].pk],
+            'export_targets': [rts[1].pk]
+        }
+
+        print(cls.form_data)
+
+
+class L2VPNTerminationTestCase(
+        ViewTestCases.GetObjectViewTestCase,
+        ViewTestCases.GetObjectChangelogViewTestCase,
+        ViewTestCases.CreateObjectViewTestCase,
+        ViewTestCases.EditObjectViewTestCase,
+        ViewTestCases.DeleteObjectViewTestCase,
+        ViewTestCases.ListObjectsViewTestCase,
+        ViewTestCases.BulkImportObjectsViewTestCase,
+        ViewTestCases.BulkDeleteObjectsViewTestCase,
+):
+
+    model = L2VPNTermination
+
+    @classmethod
+    def setUpTestData(cls):
+        site = Site.objects.create(name='Site 1')
+        manufacturer = Manufacturer.objects.create(name='Manufacturer 1')
+        device_type = DeviceType.objects.create(model='Device Type 1', manufacturer=manufacturer)
+        device_role = DeviceRole.objects.create(name='Switch')
+        device = Device.objects.create(
+            name='Device 1',
+            site=site,
+            device_type=device_type,
+            device_role=device_role,
+            status='active'
+        )
+
+        interface = Interface.objects.create(name='Interface 1', device=device, type='1000baset')
+        l2vpn = L2VPN.objects.create(name='L2VPN 1', type='vxlan', identifier=650001)
+        l2vpn_vlans = L2VPN.objects.create(name='L2VPN 2', type='vxlan', identifier=650002)
+
+        vlans = (
+            VLAN(name='Vlan 1', vid=1001),
+            VLAN(name='Vlan 2', vid=1002),
+            VLAN(name='Vlan 3', vid=1003),
+            VLAN(name='Vlan 4', vid=1004),
+            VLAN(name='Vlan 5', vid=1005),
+            VLAN(name='Vlan 6', vid=1006)
+        )
+        VLAN.objects.bulk_create(vlans)
+
+        terminations = (
+            L2VPNTermination(l2vpn=l2vpn, assigned_object=vlans[0]),
+            L2VPNTermination(l2vpn=l2vpn, assigned_object=vlans[1]),
+            L2VPNTermination(l2vpn=l2vpn, assigned_object=vlans[2])
+        )
+        L2VPNTermination.objects.bulk_create(terminations)
+
+        cls.form_data = {
+            'l2vpn': l2vpn.pk,
+            'device': device.pk,
+            'interface': interface.pk,
+        }
+
+        cls.csv_data = (
+            "l2vpn,vlan",
+            "L2VPN 2,Vlan 4",
+            "L2VPN 2,Vlan 5",
+            "L2VPN 2,Vlan 6",
+        )
+
+        cls.bulk_edit_data = {}
+
+    #
+    # Custom assertions
+    #
+
+    def assertInstanceEqual(self, instance, data, exclude=None, api=False):
+        """
+        Override parent
+        """
+        if exclude is None:
+            exclude = []
+
+        fields = [k for k in data.keys() if k not in exclude]
+        model_dict = self.model_to_dict(instance, fields=fields, api=api)
+
+        # Omit any dictionary keys which are not instance attributes or have been excluded
+        relevant_data = {
+            k: v for k, v in data.items() if hasattr(instance, k) and k not in exclude
+        }
+
+        # Handle relations on the model
+        for k, v in model_dict.items():
+            if isinstance(v, object) and hasattr(v, 'first'):
+                model_dict[k] = v.first().pk
 
-class L2VPNTerminationTestCase(ViewTestCases.PrimaryObjectViewTestCase):
-    # TODO: L2VPN Termination Tests
-    pass
+        self.assertDictEqual(model_dict, relevant_data)

netbox/ipam/urls.py

@@ -201,6 +201,7 @@ urlpatterns = [
     path('l2vpn-termination/', views.L2VPNTerminationListView.as_view(), name='l2vpntermination_list'),
     path('l2vpn-termination/add/', views.L2VPNTerminationEditView.as_view(), name='l2vpntermination_add'),
     path('l2vpn-termination/import/', views.L2VPNTerminationBulkImportView.as_view(), name='l2vpntermination_import'),
+    path('l2vpn-termination/edit/', views.L2VPNTerminationBulkEditView.as_view(), name='l2vpntermination_bulk_edit'),
     path('l2vpn-termination/delete/', views.L2VPNTerminationBulkDeleteView.as_view(), name='l2vpntermination_bulk_delete'),
     path('l2vpn-termination/<int:pk>/', views.L2VPNTerminationView.as_view(), name='l2vpntermination'),
     path('l2vpn-termination/<int:pk>/edit/', views.L2VPNTerminationEditView.as_view(), name='l2vpntermination_edit'),

netbox/ipam/views.py

@@ -1141,6 +1141,13 @@ class ServiceBulkEditView(generic.BulkEditView):
     queryset = Service.objects.prefetch_related('device', 'virtual_machine')
     filterset = filtersets.ServiceFilterSet
     table = tables.ServiceTable
+    form = forms.ServiceBulkEditForm
+
+
+class ServiceBulkDeleteView(generic.BulkDeleteView):
+    queryset = Service.objects.prefetch_related('device', 'virtual_machine')
+    filterset = filtersets.ServiceFilterSet
+    table = tables.ServiceTable
 
 
 # L2VPN
@@ -1232,14 +1239,14 @@ class L2VPNTerminationBulkImportView(generic.BulkImportView):
     table = tables.L2VPNTerminationTable
 
 
-class L2VPNTerminationBulkDeleteView(generic.BulkDeleteView):
+class L2VPNTerminationBulkEditView(generic.BulkEditView):
     queryset = L2VPNTermination.objects.all()
     filterset = filtersets.L2VPNTerminationFilterSet
     table = tables.L2VPNTerminationTable
-    form = forms.ServiceBulkEditForm
+    form = forms.L2VPNTerminationBulkEditForm
 
 
-class ServiceBulkDeleteView(generic.BulkDeleteView):
-    queryset = Service.objects.prefetch_related('device', 'virtual_machine')
-    filterset = filtersets.ServiceFilterSet
-    table = tables.ServiceTable
+class L2VPNTerminationBulkDeleteView(generic.BulkDeleteView):
+    queryset = L2VPNTermination.objects.all()
+    filterset = filtersets.L2VPNTerminationFilterSet
+    table = tables.L2VPNTerminationTable

netbox/templates/dcim/interface.html

@@ -104,6 +104,10 @@
               <th scope="row">LAG</th>
               <td>{{ object.lag|linkify|placeholder }}</td>
             </tr>
+            <tr>
+              <th scope="row">L2VPN</th>
+              <td>{{ object.l2vpn_termination.l2vpn|linkify|placeholder }}</td>
+            </tr>
           </table>
         </div>
       </div>

netbox/templates/ipam/vlan.html

@@ -64,6 +64,10 @@
                             <th scope="row">Description</th>
                             <td>{{ object.description|placeholder }}</td>
                         </tr>
+                        <tr>
+                          <th scope="row">L2VPN</th>
+                          <td>{{ object.l2vpn_termination.l2vpn|linkify|placeholder }}</td>
+                        </tr>
                     </table>
                 </div>
             </div>