mirror/netbox-community-netbox
1
0
Fork 0
mirror of https://github.com/netbox-community/netbox.git synced 2024-05-10 07:54:54 +00:00
Code Activity

L2VPN Clean Tree

Browse Source
This commit is contained in:
Daniel Sheppard
2022-06-27 23:24:50 -05:00
parent 7dd5f9e720
commit 03f1584d3a
27 changed files with 1130 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
netbox/dcim/models/device_components.py
View File

@ -649,6 +649,11 @@ class Interface(ModularComponentModel, BaseInterface, LinkTermination, PathEndpo
object_id_field='interface_id',
related_query_name='+'
)
l2vpn = GenericRelation(
to='ipam.L2VPNTermination',
content_type_field='assigned_object_type',
object_id_field='assigned_object_id',
)
clone_fields = ['device', 'parent', 'bridge', 'lag', 'type', 'mgmt_only', 'poe_mode', 'poe_type']

27
netbox/ipam/api/nested_serializers.py
View File

@ -1,6 +1,7 @@
from rest_framework import serializers
from ipam import models
from ipam.models.l2vpn import L2VPNTermination, L2VPN
from netbox.api import WritableNestedSerializer
__all__ = [
@ -190,3 +191,29 @@ class NestedServiceSerializer(WritableNestedSerializer):
class Meta:
model = models.Service
fields = ['id', 'url', 'display', 'name', 'protocol', 'ports']
#
# Virtual Circuits
#
class NestedL2VPNSerializer(WritableNestedSerializer):
url = serializers.HyperlinkedIdentityField(view_name='ipam-api:l2vpn-detail')
class Meta:
model = L2VPN
fields = [
'id', 'url', 'display', 'name', 'type'
]
class NestedL2VPNTerminationSerializer(WritableNestedSerializer):
url = serializers.HyperlinkedIdentityField(view_name='ipam-api:l2vpn_termination-detail')
l2vpn = NestedL2VPNSerializer()
class Meta:
model = L2VPNTermination
fields = [
'id', 'url', 'display', 'l2vpn', 'assigned_object'
]

58
netbox/ipam/api/serializers.py
View File

@ -19,6 +19,9 @@ from .nested_serializers import *
#
# ASNs
#
from .nested_serializers import NestedL2VPNSerializer
from ..models.l2vpn import L2VPNTermination, L2VPN
class ASNSerializer(NetBoxModelSerializer):
url = serializers.HyperlinkedIdentityField(view_name='ipam-api:asn-detail')
@ -433,3 +436,58 @@ class ServiceSerializer(NetBoxModelSerializer):
'id', 'url', 'display', 'device', 'virtual_machine', 'name', 'ports', 'protocol', 'ipaddresses',
'description', 'tags', 'custom_fields', 'created', 'last_updated',
]
#
# Virtual Circuits
#
class L2VPNSerializer(NetBoxModelSerializer):
url = serializers.HyperlinkedIdentityField(view_name='ipam-api:l2vpn-detail')
type = ChoiceField(choices=L2VPNTypeChoices, required=False)
import_targets = SerializedPKRelatedField(
queryset=RouteTarget.objects.all(),
serializer=NestedRouteTargetSerializer,
required=False,
many=True
)
export_targets = SerializedPKRelatedField(
queryset=RouteTarget.objects.all(),
serializer=NestedRouteTargetSerializer,
required=False,
many=True
)
tenant = NestedTenantSerializer(required=False, allow_null=True)
class Meta:
model = L2VPN
fields = [
'id', 'url', 'display', 'identifier', 'name', 'slug', 'type', 'import_targets', 'export_targets',
'description', 'tenant',
# Extra Fields
'tags', 'custom_fields', 'created', 'last_updated'
]
class L2VPNTerminationSerializer(NetBoxModelSerializer):
url = serializers.HyperlinkedIdentityField(view_name='ipam-api:l2vpntermination-detail')
l2vpn = NestedL2VPNSerializer()
assigned_object_type = ContentTypeField(
queryset=ContentType.objects.all()
)
assigned_object = serializers.SerializerMethodField(read_only=True)
class Meta:
model = L2VPNTermination
fields = [
'id', 'url', 'display', 'l2vpn', 'assigned_object_type', 'assigned_object_id',
'assigned_object',
# Extra Fields
'tags', 'custom_fields', 'created', 'last_updated'
]
@swagger_serializer_method(serializer_or_field=serializers.DictField)
def get_assigned_object(self, instance):
serializer = get_serializer_for_model(instance.assigned_object, prefix='Nested')
context = {'request': self.context['request']}
return serializer(instance.assigned_object, context=context).data

4
netbox/ipam/api/urls.py
View File

@ -45,6 +45,10 @@ router.register('vlans', views.VLANViewSet)
router.register('service-templates', views.ServiceTemplateViewSet)
router.register('services', views.ServiceViewSet)
# L2VPN
router.register('l2vpn', views.L2VPNViewSet)
router.register('l2vpn-termination', views.L2VPNTerminationViewSet)
app_name = 'ipam-api'
urlpatterns = [

13
netbox/ipam/api/views.py
View File

@ -18,6 +18,7 @@ from netbox.config import get_config
from utilities.constants import ADVISORY_LOCK_KEYS
from utilities.utils import count_related
from . import serializers
from ..models.l2vpn import L2VPN, L2VPNTermination
class IPAMRootView(APIRootView):
@ -157,6 +158,18 @@ class ServiceViewSet(NetBoxModelViewSet):
filterset_class = filtersets.ServiceFilterSet
class L2VPNViewSet(NetBoxModelViewSet):
queryset = L2VPN.objects
serializer_class = serializers.L2VPNSerializer
filterset_class = filtersets.L2VPNFilterSet
class L2VPNTerminationViewSet(NetBoxModelViewSet):
queryset = L2VPNTermination.objects
serializer_class = serializers.L2VPNTerminationSerializer
filterset_class = filtersets.L2VPNTerminationFilterSet
#
# Views
#

50
netbox/ipam/choices.py
View File

@ -170,3 +170,53 @@ class ServiceProtocolChoices(ChoiceSet):
(PROTOCOL_UDP, 'UDP'),
(PROTOCOL_SCTP, 'SCTP'),
)
class L2VPNTypeChoices(ChoiceSet):
TYPE_VPLS = 'vpls'
TYPE_VPWS = 'vpws'
TYPE_EPL = 'epl'
TYPE_EVPL = 'evpl'
TYPE_EPLAN = 'ep-lan'
TYPE_EVPLAN = 'evp-lan'
TYPE_EPTREE = 'ep-tree'
TYPE_EVPTREE = 'evp-tree'
TYPE_VXLAN = 'vxlan'
TYPE_VXLAN_EVPN = 'vxlan-evpn'
TYPE_MPLS_EVPN = 'mpls-evpn'
TYPE_PBB_EVPN = 'pbb-evpn'
CHOICES = (
('VPLS', (
(TYPE_VPWS, 'VPWS'),
(TYPE_VPLS, 'VPLS'),
)),
('E-Line', (
(TYPE_EPL, 'EPL'),
(TYPE_EVPL, 'EVPL'),
)),
('E-LAN', (
(TYPE_EPLAN, 'Ethernet Private LAN'),
(TYPE_EVPLAN, 'Ethernet Virtual Private LAN'),
)),
('E-Tree', (
(TYPE_EPTREE, 'Ethernet Private Tree'),
(TYPE_EVPTREE, 'Ethernet Virtual Private Tree'),
)),
('VXLAN', (
(TYPE_VXLAN, 'VXLAN'),
(TYPE_VXLAN_EVPN, 'VXLAN-EVPN'),
)),
('L2VPN E-VPN', (
(TYPE_MPLS_EVPN, 'MPLS EVPN'),
(TYPE_PBB_EVPN, 'PBB EVPN'),
))
)
P2P = (
TYPE_VPWS,
TYPE_EPL,
TYPE_EPLAN,
TYPE_EPTREE
)

6
netbox/ipam/constants.py
View File

@ -90,3 +90,9 @@ VLANGROUP_SCOPE_TYPES = (
# 16-bit port number
SERVICE_PORT_MIN = 1
SERVICE_PORT_MAX = 65535
L2VPN_ASSIGNMENT_MODELS = Q(
Q(app_label='dcim', model='interface') |
Q(app_label='ipam', model='vlan') |
Q(app_label='virtualization', model='vminterface')
)

65
netbox/ipam/filtersets.py
View File

@ -23,6 +23,8 @@ __all__ = (
'FHRPGroupFilterSet',
'IPAddressFilterSet',
'IPRangeFilterSet',
'L2VPNFilterSet',
'L2VPNTerminationFilterSet',
'PrefixFilterSet',
'RIRFilterSet',
'RoleFilterSet',
@ -922,3 +924,66 @@ class ServiceFilterSet(NetBoxModelFilterSet):
return queryset
qs_filter = Q(name__icontains=value) | Q(description__icontains=value)
return queryset.filter(qs_filter)
#
# L2VPN
#
class L2VPNFilterSet(NetBoxModelFilterSet, TenancyFilterSet):
import_target_id = django_filters.ModelMultipleChoiceFilter(
field_name='import_targets',
queryset=RouteTarget.objects.all(),
label='Import target',
)
import_target = django_filters.ModelMultipleChoiceFilter(
field_name='import_targets__name',
queryset=RouteTarget.objects.all(),
to_field_name='name',
label='Import target (name)',
)
export_target_id = django_filters.ModelMultipleChoiceFilter(
field_name='export_targets',
queryset=RouteTarget.objects.all(),
label='Export target',
)
export_target = django_filters.ModelMultipleChoiceFilter(
field_name='export_targets__name',
queryset=RouteTarget.objects.all(),
to_field_name='name',
label='Export target (name)',
)
class Meta:
model = L2VPN
fields = ['identifier', 'name', 'type', 'description']
def search(self, queryset, name, value):
if not value.strip():
return queryset
qs_filter = Q(identifier=value) | Q(name__icontains=value) | Q(description__icontains=value)
return queryset.filter(qs_filter)
class L2VPNTerminationFilterSet(NetBoxModelFilterSet):
l2vpn_id = django_filters.ModelMultipleChoiceFilter(
queryset=L2VPN.objects.all(),
label='L2VPN (ID)',
)
l2vpn = django_filters.ModelMultipleChoiceFilter(
field_name='l2vpn__name',
queryset=L2VPN.objects.all(),
to_field_name='name',
label='L2VPN (name)',
)
class Meta:
model = L2VPNTermination
fields = ['l2vpn']
def search(self, queryset, name, value):
if not value.strip():
return queryset
qs_filter = Q(l2vpn__name__icontains=value)
return queryset.filter(qs_filter)

18
netbox/ipam/forms/bulk_edit.py
View File

@ -18,6 +18,7 @@ __all__ = (
'FHRPGroupBulkEditForm',
'IPAddressBulkEditForm',
'IPRangeBulkEditForm',
'L2VPNBulkEditForm',
'PrefixBulkEditForm',
'RIRBulkEditForm',
'RoleBulkEditForm',
@ -440,3 +441,20 @@ class ServiceTemplateBulkEditForm(NetBoxModelBulkEditForm):
class ServiceBulkEditForm(ServiceTemplateBulkEditForm):
model = Service
class L2VPNBulkEditForm(NetBoxModelBulkEditForm):
tenant = DynamicModelChoiceField(
queryset=Tenant.objects.all(),
required=False
)
description = forms.CharField(
max_length=100,
required=False
)
model = L2VPN
fieldsets = (
(None, ('tenant', 'description')),
)
nullable_fields = ('tenant', 'description',)

74
netbox/ipam/forms/bulk_import.py
View File

@ -1,5 +1,6 @@
from django import forms
from django.contrib.contenttypes.models import ContentType
from django.core.exceptions import ValidationError
from dcim.models import Device, Interface, Site
from ipam.choices import *
@ -16,6 +17,8 @@ __all__ = (
'FHRPGroupCSVForm',
'IPAddressCSVForm',
'IPRangeCSVForm',
'L2VPNCSVForm',
'L2VPNTerminationCSVForm',
'PrefixCSVForm',
'RIRCSVForm',
'RoleCSVForm',
@ -425,3 +428,74 @@ class ServiceCSVForm(NetBoxModelCSVForm):
class Meta:
model = Service
fields = ('device', 'virtual_machine', 'name', 'protocol', 'ports', 'description')
class L2VPNCSVForm(NetBoxModelCSVForm):
tenant = CSVModelChoiceField(
queryset=Tenant.objects.all(),
required=False,
to_field_name='name',
)
type = CSVChoiceField(
choices=L2VPNTypeChoices,
help_text='IP protocol'
)
class Meta:
model = L2VPN
fields = ('identifier', 'name', 'slug', 'type', 'description')
class L2VPNTerminationCSVForm(NetBoxModelCSVForm):
l2vpn = CSVModelChoiceField(
queryset=L2VPN.objects.all(),
required=True,
to_field_name='name',
label='L2VPN',
)
device = CSVModelChoiceField(
queryset=Device.objects.all(),
required=False,
to_field_name='name',
help_text='Required if assigned to a interface'
)
interface = CSVModelChoiceField(
queryset=Interface.objects.all(),
required=False,
to_field_name='name',
help_text='Required if not assigned to a vlan'
)
vlan = CSVModelChoiceField(
queryset=VLAN.objects.all(),
required=False,
to_field_name='name',
help_text='Required if not assigned to a interface'
)
class Meta:
model = L2VPNTermination
fields = ('l2vpn', 'device', 'interface', 'vlan')
def __init__(self, data=None, *args, **kwargs):
super().__init__(data, *args, **kwargs)
if data:
# Limit interface queryset by assigned device
if data.get('device'):
self.fields['interface'].queryset = Interface.objects.filter(
**{f"device__{self.fields['device'].to_field_name}": data['device']}
)
def clean(self):
super().clean()
if not (self.cleaned_data.get('interface') or self.cleaned_data.get('vlan')):
raise ValidationError('You must have either a interface or a VLAN')
if self.cleaned_data.get('interface') and self.cleaned_data.get('vlan'):
raise ValidationError('Cannot assign both a interface and vlan')
# Set Assigned Object
self.instance.assigned_object = self.cleaned_data.get('interface') or self.cleaned_data.get('vlan')

29
netbox/ipam/forms/filtersets.py
View File

@ -19,6 +19,8 @@ __all__ = (
'FHRPGroupFilterForm',
'IPAddressFilterForm',
'IPRangeFilterForm',
'L2VPNFilterForm',
'L2VPNTerminationFilterForm',
'PrefixFilterForm',
'RIRFilterForm',
'RoleFilterForm',
@ -463,3 +465,30 @@ class ServiceTemplateFilterForm(NetBoxModelFilterSetForm):
class ServiceFilterForm(ServiceTemplateFilterForm):
model = Service
class L2VPNFilterForm(TenancyFilterForm, NetBoxModelFilterSetForm):
model = L2VPN
fieldsets = (
(None, ('type', )),
('Tenant', ('tenant_group_id', 'tenant_id')),
)
type = forms.ChoiceField(
choices=add_blank_choice(L2VPNTypeChoices),
required=False,
widget=StaticSelect()
)
class L2VPNTerminationFilterForm(NetBoxModelFilterSetForm):
model = L2VPNTermination
fieldsets = (
(None, ('l2vpn', )),
)
l2vpn = DynamicModelChoiceField(
queryset=L2VPN.objects.all(),
required=True,
query_params={},
label='L2VPN',
fetch_trigger='open'
)

96
netbox/ipam/forms/models.py
View File

@ -1,5 +1,6 @@
from django import forms
from django.contrib.contenttypes.models import ContentType
from django.core.exceptions import ValidationError
from dcim.models import Device, Interface, Location, Rack, Region, Site, SiteGroup
from extras.models import Tag
@ -8,8 +9,10 @@ from ipam.constants import *
from ipam.formfields import IPNetworkFormField
from ipam.models import *
from ipam.models import ASN
from ipam.models.l2vpn import L2VPN, L2VPNTermination
from netbox.forms import NetBoxModelForm
from tenancy.forms import TenancyForm
from tenancy.models import Tenant
from utilities.exceptions import PermissionsViolation
from utilities.forms import (
add_blank_choice, BootstrapMixin, ContentTypeChoiceField, DatePicker, DynamicModelChoiceField,
@ -26,6 +29,8 @@ __all__ = (
'IPAddressBulkAddForm',
'IPAddressForm',
'IPRangeForm',
'L2VPNForm',
'L2VPNTerminationForm',
'PrefixForm',
'RIRForm',
'RoleForm',
@ -861,3 +866,94 @@ class ServiceCreateForm(ServiceForm):
self.cleaned_data['description'] = service_template.description
elif not all(self.cleaned_data[f] for f in ('name', 'protocol', 'ports')):
raise forms.ValidationError("Must specify name, protocol, and port(s) if not using a service template.")
#
# L2VPN
#
class L2VPNForm(TenancyForm, NetBoxModelForm):
slug = SlugField()
import_targets = DynamicModelMultipleChoiceField(
queryset=RouteTarget.objects.all(),
required=False
)
export_targets = DynamicModelMultipleChoiceField(
queryset=RouteTarget.objects.all(),
required=False
)
fieldsets = (
('L2VPN', ('name', 'slug', 'type', 'identifier', 'description', 'tags')),
('Route Targets', ('import_targets', 'export_targets')),
('Tenancy', ('tenant_group', 'tenant')),
)
class Meta:
model = L2VPN
fields = (
'name', 'slug', 'type', 'identifier', 'description', 'import_targets', 'export_targets', 'tenant', 'tags'
)
class L2VPNTerminationForm(NetBoxModelForm):
l2vpn = DynamicModelChoiceField(
queryset=L2VPN.objects.all(),
required=True,
query_params={},
label='L2VPN',
fetch_trigger='open'
)
device = DynamicModelChoiceField(
queryset=Device.objects.all(),
required=False,
query_params={}
)
vlan = DynamicModelChoiceField(
queryset=VLAN.objects.all(),
required=False,
query_params={
'available_on_device': '$device'
}
)
interface = DynamicModelChoiceField(
queryset=Interface.objects.all(),
required=False,
query_params={
'device_id': '$device'
}
)
class Meta:
model = L2VPNTermination
fields = ('l2vpn', )
def __init__(self, *args, **kwargs):
instance = kwargs.get('instance')
initial = kwargs.get('initial', {}).copy()
if instance:
if type(instance.assigned_object) is Interface:
initial['device'] = instance.assigned_object.parent
initial['interface'] = instance.assigned_object
elif type(instance.assigned_object) is VLAN:
initial['vlan'] = instance.assigned_object
kwargs['initial'] = initial
super().__init__(*args, **kwargs)
def clean(self):
super().clean()
if not (self.cleaned_data.get('interface') or self.cleaned_data.get('vlan')):
raise ValidationError('You must have either a interface or a VLAN')
if self.cleaned_data.get('interface') and self.cleaned_data.get('vlan'):
raise ValidationError('Cannot assign both a interface and vlan')
obj = self.cleaned_data.get('interface') or self.cleaned_data.get('vlan')
self.instance.assigned_object = obj

3
netbox/ipam/models/__init__.py
View File

@ -2,6 +2,7 @@
from .fhrp import *
from .vrfs import *
from .ip import *
from .l2vpn import *
from .services import *
from .vlans import *
@ -12,6 +13,8 @@ __all__ = (
'IPRange',
'FHRPGroup',
'FHRPGroupAssignment',
'L2VPN',
'L2VPNTermination',
'Prefix',
'RIR',
'Role',

116
netbox/ipam/models/l2vpn.py Normal file
View File

@ -0,0 +1,116 @@
from django.contrib.contenttypes.fields import GenericRelation, GenericForeignKey
from django.contrib.contenttypes.models import ContentType
from django.core.exceptions import ValidationError
from django.db import models
from django.urls import reverse
from ipam.choices import L2VPNTypeChoices
from ipam.constants import L2VPN_ASSIGNMENT_MODELS
from netbox.models import NetBoxModel
class L2VPN(NetBoxModel):
name = models.CharField(
max_length=100,
unique=True
)
slug = models.SlugField()
type = models.CharField(max_length=50, choices=L2VPNTypeChoices)
identifier = models.BigIntegerField(
null=True,
blank=True,
unique=True
)
import_targets = models.ManyToManyField(
to='ipam.RouteTarget',
related_name='importing_l2vpns',
blank=True,
)
export_targets = models.ManyToManyField(
to='ipam.RouteTarget',
related_name='exporting_l2vpns',
blank=True
)
description = models.TextField(null=True, blank=True)
tenant = models.ForeignKey(
to='tenancy.Tenant',
on_delete=models.SET_NULL,
related_name='l2vpns',
blank=True,
null=True
)
contacts = GenericRelation(
to='tenancy.ContactAssignment'
)
class Meta:
ordering = ('identifier', 'name')
verbose_name = 'L2VPN'
def __str__(self):
if self.identifier:
return f'{self.name} ({self.identifier})'
return f'{self.name}'
def get_absolute_url(self):
return reverse('ipam:l2vpn', args=[self.pk])
class L2VPNTermination(NetBoxModel):
l2vpn = models.ForeignKey(
to='ipam.L2VPN',
on_delete=models.CASCADE,
related_name='terminations',
blank=False,
null=False
)
assigned_object_type = models.ForeignKey(
to=ContentType,
limit_choices_to=L2VPN_ASSIGNMENT_MODELS,
on_delete=models.PROTECT,
related_name='+',
blank=True,
null=True
)
assigned_object_id = models.PositiveBigIntegerField(
blank=True,
null=True
)
assigned_object = GenericForeignKey(
ct_field='assigned_object_type',
fk_field='assigned_object_id'
)
class Meta:
ordering = ('l2vpn',)
verbose_name = 'L2VPN Termination'
constraints = (
models.UniqueConstraint(
fields=('assigned_object_type', 'assigned_object_id'),
name='ipam_l2vpntermination_assigned_object'
),
)
def __str__(self):
if self.pk is not None:
return f'{self.assigned_object} <> {self.l2vpn}'
return ''
def get_absolute_url(self):
return reverse('ipam:l2vpntermination', args=[self.pk])
def clean(self):
# Only check is assigned_object is set
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).\
exclude(pk=self.pk).count() > 0:
raise ValidationError(f'L2VPN Termination already assigned ({self.assigned_object})')
# Only check if L2VPN is set and is of type P2P
if self.l2vpn and self.l2vpn.type in L2VPNTypeChoices.P2P:
if L2VPNTermination.objects.filter(l2vpn=self.l2vpn).exclude(pk=self.pk).count() >= 2:
raise ValidationError(f'P2P Type L2VPNs can only have 2 terminations; first delete a termination')

9
netbox/ipam/models/vlans.py
View File

@ -1,4 +1,4 @@
from django.contrib.contenttypes.fields import GenericForeignKey
from django.contrib.contenttypes.fields import GenericForeignKey, GenericRelation
from django.contrib.contenttypes.models import ContentType
from django.core.exceptions import ValidationError
from django.core.validators import MaxValueValidator, MinValueValidator
@ -8,6 +8,7 @@ from django.urls import reverse
from dcim.models import Interface
from ipam.choices import *
from ipam.constants import *
from ipam.models import L2VPNTermination
from ipam.querysets import VLANQuerySet
from netbox.models import OrganizationalModel, NetBoxModel
from virtualization.models import VMInterface
@ -173,6 +174,12 @@ class VLAN(NetBoxModel):
blank=True
)
l2vpn = GenericRelation(
to='ipam.L2VPNTermination',
content_type_field='assigned_object_type',
object_id_field='assigned_object_id',
)
objects = VLANQuerySet.as_manager()
clone_fields = [

1
netbox/ipam/tables/__init__.py
View File

@ -1,5 +1,6 @@
from .fhrp import *
from .ip import *
from .l2vpn import *
from .services import *
from .vlans import *
from .vrfs import *

38
netbox/ipam/tables/l2vpn.py Normal file
View File

@ -0,0 +1,38 @@
import django_tables2 as tables
from ipam.models import *
from ipam.models.l2vpn import L2VPN, L2VPNTermination
from netbox.tables import NetBoxTable, columns
__all__ = (
'L2VPNTable',
'L2VPNTerminationTable',
)
class L2VPNTable(NetBoxTable):
pk = columns.ToggleColumn()
name = tables.Column(
linkify=True
)
class Meta(NetBoxTable.Meta):
model = L2VPN
fields = ('pk', 'name', 'description', 'slug', 'type', 'tenant', 'actions')
default_columns = ('pk', 'name', 'description', 'actions')
class L2VPNTerminationTable(NetBoxTable):
pk = columns.ToggleColumn()
assigned_object_type = columns.ContentTypeColumn(
verbose_name='Object Type'
)
assigned_object = tables.Column(
linkify=True,
orderable=False
)
class Meta(NetBoxTable.Meta):
model = L2VPNTermination
fields = ('pk', 'l2vpn', 'assigned_object_type', 'assigned_object', 'actions')
default_columns = ('pk', 'l2vpn', 'assigned_object_type', 'assigned_object', 'actions')

93
netbox/ipam/tests/test_api.py
View File

@ -914,3 +914,96 @@ class ServiceTest(APIViewTestCases.APIViewTestCase):
'ports': [6],
},
]
class L2VPNTest(APIViewTestCases.APIViewTestCase):
model = L2VPN
brief_fields = ['display', 'id', 'identifier', 'name', 'slug', 'type', 'url']
create_data = [
{
'name': 'L2VPN 4',
'slug': 'l2vpn-4',
'type': 'vxlan',
'identifier': 33343344
},
{
'name': 'L2VPN 5',
'slug': 'l2vpn-5',
'type': 'vxlan',
'identifier': 33343345
},
{
'name': 'L2VPN 6',
'slug': 'l2vpn-6',
'type': 'vpws',
'identifier': 33343346
},
]
bulk_update_data = {
'description': 'New description',
}
@classmethod
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.objects.bulk_create(l2vpns)
class L2VPNTerminationTest(APIViewTestCases.APIViewTestCase):
model = L2VPNTermination
brief_fields = ['display', 'id', 'l2vpn', 'assigned_object', 'assigned_object_id', 'assigned_object_type', '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.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])
)
cls.create_data = [
{
'l2vpn': l2vpns[0],
'assigned_object_type': 'ipam.vlan',
'assigned_object_id': vlans[3],
},
{
'l2vpn': l2vpns[0],
'assigned_object_type': 'ipam.vlan',
'assigned_object_id': vlans[4],
},
{
'l2vpn': l2vpns[0],
'assigned_object_type': 'ipam.vlan',
'assigned_object_id': vlans[5],
},
]
cls.bulk_update_data = {
'l2vpn': l2vpns[2]
}

101
netbox/ipam/tests/test_filtersets.py
View File

@ -1463,3 +1463,104 @@ class ServiceTestCase(TestCase, ChangeLoggedFilterSetTests):
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
params = {'virtual_machine': [vms[0].name, vms[1].name]}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
class L2VPNTest(TestCase, ChangeLoggedFilterSetTests):
# TODO: L2VPN Tests
queryset = L2VPN.objects.all()
filterset = L2VPNFilterSet
@classmethod
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.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
queryset = L2VPNTermination.objects.all()
filterset = L2VPNTerminationFilterSet
@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.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'),
)
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.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])
)
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)
params = {'l2vpn': ['L2VPN 1', 'L2VPN 2']}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
def test_interfaces(self):
interfaces = Interface.objects.all()[:2]
params = {'interface_id': [interfaces[0].pk, interfaces[1].pk]}
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)
def test_vlans(self):
vlans = VLAN.objects.all()[:2]
params = {'vlan_id': [vlans[0].pk, vlans[1].pk]}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
params = {'vlan': ['VLAN 1', 'VLAN 2']}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)

10
netbox/ipam/tests/test_models.py
View File

@ -538,3 +538,13 @@ class TestVLANGroup(TestCase):
VLAN.objects.create(name='VLAN 104', vid=104, group=vlangroup)
self.assertEqual(vlangroup.get_next_available_vid(), 105)
class TestL2VPN(TestCase):
# TODO: L2VPN Tests
pass
class TestL2VPNTermination(TestCase):
# TODO: L2VPN Termination Tests
pass

10
netbox/ipam/tests/test_views.py
View File

@ -746,3 +746,13 @@ class ServiceTestCase(ViewTestCases.PrimaryObjectViewTestCase):
self.assertEqual(instance.protocol, service_template.protocol)
self.assertEqual(instance.ports, service_template.ports)
self.assertEqual(instance.description, service_template.description)
class L2VPNTestCase(ViewTestCases.PrimaryObjectViewTestCase):
# TODO: L2VPN Tests
pass
class L2VPNTerminationTestCase(ViewTestCases.PrimaryObjectViewTestCase):
# TODO: L2VPN Termination Tests
pass

21
netbox/ipam/urls.py
View File

@ -186,4 +186,25 @@ urlpatterns = [
path('services/<int:pk>/changelog/', ObjectChangeLogView.as_view(), name='service_changelog', kwargs={'model': Service}),
path('services/<int:pk>/journal/', ObjectJournalView.as_view(), name='service_journal', kwargs={'model': Service}),
# L2VPN
path('l2vpn/', views.L2VPNListView.as_view(), name='l2vpn_list'),
path('l2vpn/add/', views.L2VPNEditView.as_view(), name='l2vpn_add'),
path('l2vpn/import/', views.L2VPNBulkImportView.as_view(), name='l2vpn_import'),
path('l2vpn/edit/', views.L2VPNBulkEditView.as_view(), name='l2vpn_bulk_edit'),
path('l2vpn/delete/', views.L2VPNBulkDeleteView.as_view(), name='l2vpn_bulk_delete'),
path('l2vpn/<int:pk>/', views.L2VPNView.as_view(), name='l2vpn'),
path('l2vpn/<int:pk>/edit/', views.L2VPNEditView.as_view(), name='l2vpn_edit'),
path('l2vpn/<int:pk>/delete/', views.L2VPNDeleteView.as_view(), name='l2vpn_delete'),
path('l2vpn/<int:pk>/changelog/', ObjectChangeLogView.as_view(), name='l2vpn_changelog', kwargs={'model': L2VPN}),
path('l2vpn/<int:pk>/journal/', ObjectJournalView.as_view(), name='l2vpn_journal', kwargs={'model': L2VPN}),
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/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'),
path('l2vpn-termination/<int:pk>/delete/', views.L2VPNTerminationDeleteView.as_view(), name='l2vpntermination_delete'),
path('l2vpn-termination/<int:pk>/changelog/', ObjectChangeLogView.as_view(), name='l2vpntermination_changelog', kwargs={'model': L2VPNTermination}),
path('l2vpn-termination/<int:pk>/journal/', ObjectJournalView.as_view(), name='l2vpntermination_journal', kwargs={'model': L2VPNTermination}),
]

96
netbox/ipam/views.py
View File

@ -17,6 +17,7 @@ from . import filtersets, forms, tables
from .constants import *
from .models import *
from .models import ASN
from .tables.l2vpn import L2VPNTable, L2VPNTerminationTable
from .utils import add_requested_prefixes, add_available_ipaddresses, add_available_vlans
@ -1140,6 +1141,101 @@ class ServiceBulkEditView(generic.BulkEditView):
queryset = Service.objects.prefetch_related('device', 'virtual_machine')
filterset = filtersets.ServiceFilterSet
table = tables.ServiceTable
# L2VPN
class L2VPNListView(generic.ObjectListView):
queryset = L2VPN.objects.all()
table = L2VPNTable
filterset = filtersets.L2VPNFilterSet
filterset_form = forms.L2VPNFilterForm
class L2VPNView(generic.ObjectView):
queryset = L2VPN.objects.all()
def get_extra_context(self, request, instance):
terminations = L2VPNTermination.objects.restrict(request.user, 'view').filter(l2vpn=instance)
terminations_table = tables.L2VPNTerminationTable(terminations, user=request.user, exclude=('l2vpn', ))
terminations_table.configure(request)
import_targets_table = tables.RouteTargetTable(
instance.import_targets.prefetch_related('tenant'),
orderable=False
)
export_targets_table = tables.RouteTargetTable(
instance.export_targets.prefetch_related('tenant'),
orderable=False
)
return {
'terminations_table': terminations_table,
'import_targets_table': import_targets_table,
'export_targets_table': export_targets_table,
}
class L2VPNEditView(generic.ObjectEditView):
queryset = L2VPN.objects.all()
form = forms.L2VPNForm
class L2VPNDeleteView(generic.ObjectDeleteView):
queryset = L2VPN.objects.all()
class L2VPNBulkImportView(generic.BulkImportView):
queryset = L2VPN.objects.all()
model_form = forms.L2VPNCSVForm
table = tables.L2VPNTable
class L2VPNBulkEditView(generic.BulkEditView):
queryset = L2VPN.objects.all()
filterset = filtersets.L2VPNFilterSet
table = tables.L2VPNTable
form = forms.L2VPNBulkEditForm
class L2VPNBulkDeleteView(generic.BulkDeleteView):
queryset = L2VPN.objects.all()
filterset = filtersets.L2VPNFilterSet
table = tables.L2VPNTable
class L2VPNTerminationListView(generic.ObjectListView):
queryset = L2VPNTermination.objects.all()
table = L2VPNTerminationTable
filterset = filtersets.L2VPNTerminationFilterSet
filterset_form = forms.L2VPNTerminationFilterForm
class L2VPNTerminationView(generic.ObjectView):
queryset = L2VPNTermination.objects.all()
class L2VPNTerminationEditView(generic.ObjectEditView):
queryset = L2VPNTermination.objects.all()
form = forms.L2VPNTerminationForm
template_name = 'ipam/l2vpntermination_edit.html'
class L2VPNTerminationDeleteView(generic.ObjectDeleteView):
queryset = L2VPNTermination.objects.all()
class L2VPNTerminationBulkImportView(generic.BulkImportView):
queryset = L2VPNTermination.objects.all()
model_form = forms.L2VPNTerminationCSVForm
table = tables.L2VPNTerminationTable
class L2VPNTerminationBulkDeleteView(generic.BulkDeleteView):
queryset = L2VPNTermination.objects.all()
filterset = filtersets.L2VPNTerminationFilterSet
table = tables.L2VPNTerminationTable
form = forms.ServiceBulkEditForm

7
netbox/netbox/navigation_menu.py
View File

@ -260,6 +260,13 @@ IPAM_MENU = Menu(
get_model_item('ipam', 'vlangroup', 'VLAN Groups'),
),
),
MenuGroup(
label='L2VPNs',
items=(
get_model_item('ipam', 'l2vpn', 'L2VPN'),
get_model_item('ipam', 'l2vpntermination', 'Terminations'),
),
),
MenuGroup(
label='Other',
items=(

111
netbox/templates/ipam/l2vpn.html Normal file
View File

@ -0,0 +1,111 @@
{% extends 'generic/object.html' %}
{% load helpers %}
{% load plugins %}
{% load render_table from django_tables2 %}
{% block content %}
<div class="row mb-3">
<div class="col col-md-6">
<div class="card">
<h5 class="card-header">
L2VPN Attributes
</h5>
<div class="card-body">
<table class="table table-hover attr-table
<tr>
<th scope="row">Name</th>
<td>{{ object.name|placeholder }}</td>
</tr>
<tr>
<th scope="row">Slug</th>
<td>{{ object.slug|placeholder }}</td>
</tr>
<tr>
<th scope="row">Identifier</th>
<td>{{ object.identifier|placeholder }}</td>
</tr>
<tr>
<th scope="row">Type</th>
<td>{{ object.get_type_display }}</td>
</tr>
<tr>
<th scope="row">Description</th>
<td>{{ object.description|placeholder }}</td>
</tr>
<tr>
<th scope="row">Tenant</th>
<td>{{ object.tenant|placeholder }}</td>
</tr>
</table>
</div>
</div>
{% include 'inc/panels/contacts.html' %}
{% plugin_left_page object %}
</div>
<div class="col col-md-6">
{% include 'inc/panels/tags.html' with tags=object.tags.all url='circuits:circuit_list' %}
{% include 'inc/panels/custom_fields.html' %}
{% plugin_right_page object %}
</div>
</div>
<div class="row mb-3">
<div class="col col-md-6">
{% include 'inc/panel_table.html' with table=import_targets_table heading="Import Route Targets" %}
</div>
<div class="col col-md-6">
{% include 'inc/panel_table.html' with table=export_targets_table heading="Export Route Targets" %}
</div>
</div>
<div class="row mb-3">
<div class="col col-md-12">
<div class="card">
<h5 class="card-header">L2VPN Terminations</h5>
<div class="card-body">
{% with terminations=object.terminations.all %}
{% if terminations.exists %}
<table class="table table-hover">
<tr>
<th>Termination Type</th>
<th>Termination</th>
<th></th>
</tr>
{% for termination in terminations %}
<tr>
<td>{{ termination.assigned_object|meta:"verbose_name" }}</td>
<td>{{ termination.assigned_object|linkify }}</td>
<td class="text-end noprint">
{% if perms.ipam.change_l2vpntermination %}
<a href="{% url 'ipam:l2vpntermination_edit' pk=termination.pk %}?return_url={{ object.get_absolute_url }}" class="btn btn-warning btn-sm lh-1" title="Edit">
<i class="mdi mdi-pencil" aria-hidden="true"></i>
</a>
{% endif %}
{% if perms.ipam.delete_l2vpntermination %}
<a href="{% url 'ipam:l2vpntermination_delete' pk=termination.pk %}?return_url={{ object.get_absolute_url }}" class="btn btn-danger btn-sm lh-1" title="Delete">
<i class="mdi mdi-trash-can-outline" aria-hidden="true"></i>
</a>
{% endif %}
</td>
</tr>
{% endfor %}
</table>
{% else %}
<div class="text-muted">None</div>
{% endif %}
{% endwith %}
</div>
{% if perms.ipam.add_l2vpntermination %}
<div class="card-footer text-end noprint">
<a href="{% url 'ipam:l2vpntermination_add' %}?l2vpn={{ object.pk }}&return_url={{ object.get_absolute_url }}" class="btn btn-primary btn-sm">
<i class="mdi mdi-plus-thick" aria-hidden="true"></i> Add a Termination
</a>
</div>
{% endif %}
</div>
</div>
</div>
<div class="row mb-3">
<div class="col col-md-12">
{% plugin_full_width_page object %}
</div>
</div>
{% endblock %}

31
netbox/templates/ipam/l2vpntermination.html Normal file
View File

@ -0,0 +1,31 @@
{% extends 'generic/object.html' %}
{% load helpers %}
{% block content %}
<div class="row">
<div class="col col-md-6">
<div class="card">
<h5 class="card-header">
L2VPN Attributes
</h5>
<div class="card-body">
<table class="table table-hover">
<tr>
<th scope="row">L2vPN</th>
<td>{{ object.l2vpn.name|placeholder }}</td>
</tr>
<tr>
<th scope="row">Assigned Object</th>
<td>{{ object.assigned_object.name|placeholder }}</td>
</tr>
</table>
</div>
</div>
</div>
<div class="col col-md-6">
{% include 'inc/panels/custom_fields.html' %}
{% include 'inc/panels/tags.html' with tags=object.tags.all url='ipam:l2vpntermination_list' %}
</div>
</div>
{% endblock %}

39
netbox/templates/ipam/l2vpntermination_edit.html Normal file
View File

@ -0,0 +1,39 @@
{% extends 'generic/object_edit.html' %}
{% load helpers %}
{% load form_helpers %}
{% block form %}
<div class="field-group my-5">
<div class="row mb-2">
<h5 class="offset-sm-3">L2VPN Termination</h5>
</div>
{% render_field form.l2vpn %}
<div class="row mb-3">
<div class="offset-sm-3">
<ul class="nav nav-pills" role="tablist">
<li role="presentation" class="nav-item">
<button role="tab" type="button" id="vlan_tab" data-bs-toggle="tab" aria-controls="vlan" data-bs-target="#vlan" class="nav-link {% if not form.initial.interface %}active{% endif %}">
VLAN
</button>
</li>
<li role="presentation" class="nav-item">
<button role="tab" type="button" id="interface_tab" data-bs-toggle="tab" aria-controls="interface" data-bs-target="#interface" class="nav-link {% if form.initial.interface %}active{% endif %}">
Interface
</button>
</li>
</ul>
</div>
</div>
<div class="row mb-3">
<div class="tab-content p-0 border-0">
{% render_field form.device %}
<div class="tab-pane {% if not form.initial.interface %}active{% endif %}" id="vlan" role="tabpanel" aria-labeled-by="vlan_tab">
{% render_field form.vlan %}
</div>
<div class="tab-pane {% if form.initial.interface %}active{% endif %}" id="interface" role="tabpanel" aria-labeled-by="interface_tab">
{% render_field form.interface %}
</div>
</div>
</div>
</div>
{% endblock %}