NeoAnd/netbox
1
0
Fork 0
mirror of https://github.com/netbox-community/netbox.git synced 2025-07-29 11:56:25 -06:00
Code Issues Actions 8 Packages Projects Releases Wiki Activity

Merge feature and rebuild migrations

Browse Source
This commit is contained in:
Brian Tiemann 2024-11-01 11:26:31 -04:00
commit 6d2a0a4968
66 changed files with 1146 additions and 294 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
docs/models/circuits/circuittermination.md
View File

@ -21,13 +21,9 @@ Designates the termination as forming either the A or Z end of the circuit.
If selected, the circuit termination will be considered "connected" even if no cable has been connected to it in NetBox.
### Site
### Termination
The [site](../dcim/site.md) with which this circuit termination is associated. Once created, a cable can be connected between the circuit termination and a device interface (or similar component).
### Provider Network
Circuits which do not connect to a site modeled by NetBox can instead be terminated to a [provider network](./providernetwork.md) representing an unknown network operated by a [provider](./provider.md).
The [region](../dcim/region.md), [site group](../dcim/sitegroup.md), [site](../dcim/site.md), [location](../dcim/location.md) or [provider network](./providernetwork.md) with which this circuit termination is associated. Once created, a cable can be connected between the circuit termination and a device interface (or similar component).
### Port Speed

4
docs/models/dcim/interface.md
View File

@ -119,6 +119,10 @@ The "native" (untagged) VLAN for the interface. Valid only when one of the above
The tagged VLANs which are configured to be carried by this interface. Valid only for the "tagged" 802.1Q mode above.
### Q-in-Q SVLAN
The assigned service VLAN (for Q-in-Q/802.1ad interfaces).
### Wireless Role
Indicates the configured role for wireless interfaces (access point or station).

8
docs/models/ipam/vlan.md
View File

@ -26,3 +26,11 @@ The user-defined functional [role](./role.md) assigned to the VLAN.
### VLAN Group or Site
The [VLAN group](./vlangroup.md) or [site](../dcim/site.md) to which the VLAN is assigned.
### Q-in-Q Role
For VLANs which comprise a Q-in-Q/IEEE 802.1ad topology, this field indicates whether the VLAN is treated as a service or customer VLAN.
### Q-in-Q Service VLAN
The designated parent service VLAN for a Q-in-Q customer VLAN. This may be set only for Q-in-Q custom VLANs.

4
docs/models/virtualization/vminterface.md
View File

@ -52,6 +52,10 @@ The "native" (untagged) VLAN for the interface. Valid only when one of the above
The tagged VLANs which are configured to be carried by this interface. Valid only for the "tagged" 802.1Q mode above.
### Q-in-Q SVLAN
The assigned service VLAN (for Q-in-Q/802.1ad interfaces).
### VRF
The [virtual routing and forwarding](../ipam/vrf.md) instance to which this interface is assigned.

51
netbox/circuits/api/serializers_/circuits.py
View File

@ -1,11 +1,16 @@
from django.contrib.contenttypes.models import ContentType
from drf_spectacular.utils import extend_schema_field
from rest_framework import serializers
from circuits.choices import CircuitPriorityChoices, CircuitStatusChoices
from circuits.constants import CIRCUIT_TERMINATION_TERMINATION_TYPES
from circuits.models import Circuit, CircuitGroup, CircuitGroupAssignment, CircuitTermination, CircuitType
from dcim.api.serializers_.cables import CabledObjectSerializer
from dcim.api.serializers_.sites import SiteSerializer
from netbox.api.fields import ChoiceField, RelatedObjectCountField
from netbox.api.fields import ChoiceField, ContentTypeField, RelatedObjectCountField
from netbox.api.serializers import NetBoxModelSerializer, WritableNestedSerializer
from netbox.choices import DistanceUnitChoices
from tenancy.api.serializers_.tenants import TenantSerializer
from utilities.api import get_serializer_for_model
from .providers import ProviderAccountSerializer, ProviderNetworkSerializer, ProviderSerializer
@ -33,16 +38,33 @@ class CircuitTypeSerializer(NetBoxModelSerializer):
class CircuitCircuitTerminationSerializer(WritableNestedSerializer):
site = SiteSerializer(nested=True, allow_null=True)
termination_type = ContentTypeField(
queryset=ContentType.objects.filter(
model__in=CIRCUIT_TERMINATION_TERMINATION_TYPES
),
allow_null=True,
required=False,
default=None
)
termination_id = serializers.IntegerField(allow_null=True, required=False, default=None)
termination = serializers.SerializerMethodField(read_only=True)
provider_network = ProviderNetworkSerializer(nested=True, allow_null=True)
class Meta:
model = CircuitTermination
fields = [
'id', 'url', 'display_url', 'display', 'site', 'provider_network', 'port_speed', 'upstream_speed',
'id', 'url', 'display_url', 'display', 'termination_type', 'termination_id', 'termination', 'provider_network', 'port_speed', 'upstream_speed',
'xconnect_id', 'description',
]
@extend_schema_field(serializers.JSONField(allow_null=True))
def get_termination(self, obj):
if obj.termination_id is None:
return None
serializer = get_serializer_for_model(obj.termination)
context = {'request': self.context['request']}
return serializer(obj.termination, nested=True, context=context).data
class CircuitGroupSerializer(NetBoxModelSerializer):
tenant = TenantSerializer(nested=True, required=False, allow_null=True)
@ -95,18 +117,35 @@ class CircuitSerializer(NetBoxModelSerializer):
class CircuitTerminationSerializer(NetBoxModelSerializer, CabledObjectSerializer):
circuit = CircuitSerializer(nested=True)
site = SiteSerializer(nested=True, required=False, allow_null=True)
termination_type = ContentTypeField(
queryset=ContentType.objects.filter(
model__in=CIRCUIT_TERMINATION_TERMINATION_TYPES
),
allow_null=True,
required=False,
default=None
)
termination_id = serializers.IntegerField(allow_null=True, required=False, default=None)
termination = serializers.SerializerMethodField(read_only=True)
provider_network = ProviderNetworkSerializer(nested=True, required=False, allow_null=True)
class Meta:
model = CircuitTermination
fields = [
'id', 'url', 'display_url', 'display', 'circuit', 'term_side', 'site', 'provider_network', 'port_speed',
'id', 'url', 'display_url', 'display', 'circuit', 'term_side', 'termination_type', 'termination_id', 'termination', 'provider_network', 'port_speed',
'upstream_speed', 'xconnect_id', 'pp_info', 'description', 'mark_connected', 'cable', 'cable_end',
'link_peers', 'link_peers_type', 'tags', 'custom_fields', 'created', 'last_updated', '_occupied',
]
brief_fields = ('id', 'url', 'display', 'circuit', 'term_side', 'description', 'cable', '_occupied')
@extend_schema_field(serializers.JSONField(allow_null=True))
def get_termination(self, obj):
if obj.termination_id is None:
return None
serializer = get_serializer_for_model(obj.termination)
context = {'request': self.context['request']}
return serializer(obj.termination, nested=True, context=context).data
class CircuitGroupAssignmentSerializer(CircuitGroupAssignmentSerializer_):
circuit = CircuitSerializer(nested=True)

4
netbox/circuits/constants.py Normal file
View File

@ -0,0 +1,4 @@
# models values for ContentTypes which may be CircuitTermination termination types
CIRCUIT_TERMINATION_TERMINATION_TYPES = (
'region', 'sitegroup', 'site', 'location', 'providernetwork',
)

76
netbox/circuits/filtersets.py
View File

@ -3,11 +3,11 @@ from django.db.models import Q
from django.utils.translation import gettext as _
from dcim.filtersets import CabledObjectFilterSet
from dcim.models import Region, Site, SiteGroup
from dcim.models import Location, Region, Site, SiteGroup
from ipam.models import ASN
from netbox.filtersets import NetBoxModelFilterSet, OrganizationalModelFilterSet
from tenancy.filtersets import ContactModelFilterSet, TenancyFilterSet
from utilities.filters import TreeNodeMultipleChoiceFilter
from utilities.filters import ContentTypeFilter, TreeNodeMultipleChoiceFilter
from .choices import *
from .models import *
@ -26,37 +26,37 @@ __all__ = (
class ProviderFilterSet(NetBoxModelFilterSet, ContactModelFilterSet):
region_id = TreeNodeMultipleChoiceFilter(
queryset=Region.objects.all(),
field_name='circuits__terminations__site__region',
field_name='circuits__terminations___region',
lookup_expr='in',
label=_('Region (ID)'),
)
region = TreeNodeMultipleChoiceFilter(
queryset=Region.objects.all(),
field_name='circuits__terminations__site__region',
field_name='circuits__terminations___region',
lookup_expr='in',
to_field_name='slug',
label=_('Region (slug)'),
)
site_group_id = TreeNodeMultipleChoiceFilter(
queryset=SiteGroup.objects.all(),
field_name='circuits__terminations__site__group',
field_name='circuits__terminations___site_group',
lookup_expr='in',
label=_('Site group (ID)'),
)
site_group = TreeNodeMultipleChoiceFilter(
queryset=SiteGroup.objects.all(),
field_name='circuits__terminations__site__group',
field_name='circuits__terminations___site_group',
lookup_expr='in',
to_field_name='slug',
label=_('Site group (slug)'),
)
site_id = django_filters.ModelMultipleChoiceFilter(
field_name='circuits__terminations__site',
field_name='circuits__terminations___site',
queryset=Site.objects.all(),
label=_('Site'),
)
site = django_filters.ModelMultipleChoiceFilter(
field_name='circuits__terminations__site__slug',
field_name='circuits__terminations___site__slug',
queryset=Site.objects.all(),
to_field_name='slug',
label=_('Site (slug)'),
@ -173,7 +173,7 @@ class CircuitFilterSet(NetBoxModelFilterSet, TenancyFilterSet, ContactModelFilte
label=_('Provider account (account)'),
)
provider_network_id = django_filters.ModelMultipleChoiceFilter(
field_name='terminations__provider_network',
field_name='terminations___provider_network',
queryset=ProviderNetwork.objects.all(),
label=_('Provider network (ID)'),
)
@ -193,37 +193,37 @@ class CircuitFilterSet(NetBoxModelFilterSet, TenancyFilterSet, ContactModelFilte
)
region_id = TreeNodeMultipleChoiceFilter(
queryset=Region.objects.all(),
field_name='terminations__site__region',
field_name='terminations___region',
lookup_expr='in',
label=_('Region (ID)'),
)
region = TreeNodeMultipleChoiceFilter(
queryset=Region.objects.all(),
field_name='terminations__site__region',
field_name='terminations___region',
lookup_expr='in',
to_field_name='slug',
label=_('Region (slug)'),
)
site_group_id = TreeNodeMultipleChoiceFilter(
queryset=SiteGroup.objects.all(),
field_name='terminations__site__group',
field_name='terminations___site_group',
lookup_expr='in',
label=_('Site group (ID)'),
)
site_group = TreeNodeMultipleChoiceFilter(
queryset=SiteGroup.objects.all(),
field_name='terminations__site__group',
field_name='terminations___site_group',
lookup_expr='in',
to_field_name='slug',
label=_('Site group (slug)'),
)
site_id = django_filters.ModelMultipleChoiceFilter(
field_name='terminations__site',
field_name='terminations___site',
queryset=Site.objects.all(),
label=_('Site (ID)'),
)
site = django_filters.ModelMultipleChoiceFilter(
field_name='terminations__site__slug',
field_name='terminations___site__slug',
queryset=Site.objects.all(),
to_field_name='slug',
label=_('Site (slug)'),
@ -263,18 +263,60 @@ class CircuitTerminationFilterSet(NetBoxModelFilterSet, CabledObjectFilterSet):
queryset=Circuit.objects.all(),
label=_('Circuit'),
)
termination_type = ContentTypeFilter()
region_id = TreeNodeMultipleChoiceFilter(
queryset=Region.objects.all(),
field_name='_region',
lookup_expr='in',
label=_('Region (ID)'),
)
region = TreeNodeMultipleChoiceFilter(
queryset=Region.objects.all(),
field_name='_region',
lookup_expr='in',
to_field_name='slug',
label=_('Region (slug)'),
)
site_group_id = TreeNodeMultipleChoiceFilter(
queryset=SiteGroup.objects.all(),
field_name='_site_group',
lookup_expr='in',
label=_('Site group (ID)'),
)
site_group = TreeNodeMultipleChoiceFilter(
queryset=SiteGroup.objects.all(),
field_name='_site_group',
lookup_expr='in',
to_field_name='slug',
label=_('Site group (slug)'),
)
site_id = django_filters.ModelMultipleChoiceFilter(
queryset=Site.objects.all(),
field_name='_site',
label=_('Site (ID)'),
)
site = django_filters.ModelMultipleChoiceFilter(
field_name='site__slug',
field_name='_site__slug',
queryset=Site.objects.all(),
to_field_name='slug',
label=_('Site (slug)'),
)
location_id = TreeNodeMultipleChoiceFilter(
queryset=Location.objects.all(),
field_name='_location',
lookup_expr='in',
label=_('Location (ID)'),
)
location = TreeNodeMultipleChoiceFilter(
queryset=Location.objects.all(),
field_name='_location',
lookup_expr='in',
to_field_name='slug',
label=_('Location (slug)'),
)
provider_network_id = django_filters.ModelMultipleChoiceFilter(
queryset=ProviderNetwork.objects.all(),
field_name='_provider_network',
label=_('ProviderNetwork (ID)'),
)
provider_id = django_filters.ModelMultipleChoiceFilter(
@ -292,7 +334,7 @@ class CircuitTerminationFilterSet(NetBoxModelFilterSet, CabledObjectFilterSet):
class Meta:
model = CircuitTermination
fields = (
'id', 'term_side', 'port_speed', 'upstream_speed', 'xconnect_id', 'description', 'mark_connected',
'id', 'termination_id', 'term_side', 'port_speed', 'upstream_speed', 'xconnect_id', 'description', 'mark_connected',
'pp_info', 'cable_end',
)

54
netbox/circuits/forms/bulk_edit.py
View File

@ -1,17 +1,23 @@
from django import forms
from django.contrib.contenttypes.models import ContentType
from django.core.exceptions import ObjectDoesNotExist
from django.utils.translation import gettext_lazy as _
from circuits.choices import CircuitCommitRateChoices, CircuitPriorityChoices, CircuitStatusChoices
from circuits.constants import CIRCUIT_TERMINATION_TERMINATION_TYPES
from circuits.models import *
from dcim.models import Site
from ipam.models import ASN
from netbox.choices import DistanceUnitChoices
from netbox.forms import NetBoxModelBulkEditForm
from tenancy.models import Tenant
from utilities.forms import add_blank_choice
from utilities.forms.fields import ColorField, CommentField, DynamicModelChoiceField, DynamicModelMultipleChoiceField
from utilities.forms.rendering import FieldSet, TabbedGroups
from utilities.forms.widgets import BulkEditNullBooleanSelect, DatePicker, NumberWithOptions
from utilities.forms import add_blank_choice, get_field_value
from utilities.forms.fields import (
ColorField, CommentField, ContentTypeChoiceField, DynamicModelChoiceField, DynamicModelMultipleChoiceField,
)
from utilities.forms.rendering import FieldSet
from utilities.forms.widgets import BulkEditNullBooleanSelect, DatePicker, HTMXSelect, NumberWithOptions
from utilities.templatetags.builtins.filters import bettertitle
__all__ = (
'CircuitBulkEditForm',
@ -197,15 +203,18 @@ class CircuitTerminationBulkEditForm(NetBoxModelBulkEditForm):
max_length=200,
required=False
)
site = DynamicModelChoiceField(
label=_('Site'),
queryset=Site.objects.all(),
required=False
termination_type = ContentTypeChoiceField(
queryset=ContentType.objects.filter(model__in=CIRCUIT_TERMINATION_TERMINATION_TYPES),
widget=HTMXSelect(method='post', attrs={'hx-select': '#form_fields'}),
required=False,
label=_('Termination type')
)
provider_network = DynamicModelChoiceField(
label=_('Provider Network'),
queryset=ProviderNetwork.objects.all(),
required=False
termination = DynamicModelChoiceField(
label=_('Termination'),
queryset=Site.objects.none(), # Initial queryset
required=False,
disabled=True,
selector=True
)
port_speed = forms.IntegerField(
required=False,
@ -225,15 +234,26 @@ class CircuitTerminationBulkEditForm(NetBoxModelBulkEditForm):
fieldsets = (
FieldSet(
'description',
TabbedGroups(
FieldSet('site', name=_('Site')),
FieldSet('provider_network', name=_('Provider Network')),
),
'termination_type', 'termination',
'mark_connected', name=_('Circuit Termination')
),
FieldSet('port_speed', 'upstream_speed', name=_('Termination Details')),
)
nullable_fields = ('description')
nullable_fields = ('description', 'termination')
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
if termination_type_id := get_field_value(self, 'termination_type'):
try:
termination_type = ContentType.objects.get(pk=termination_type_id)
model = termination_type.model_class()
self.fields['termination'].queryset = model.objects.all()
self.fields['termination'].widget.attrs['selector'] = model._meta.label_lower
self.fields['termination'].disabled = False
self.fields['termination'].label = _(bettertitle(model._meta.verbose_name))
except ObjectDoesNotExist:
pass
class CircuitGroupBulkEditForm(NetBoxModelBulkEditForm):

30
netbox/circuits/forms/bulk_import.py
View File

@ -1,13 +1,14 @@
from django import forms
from django.contrib.contenttypes.models import ContentType
from django.utils.translation import gettext_lazy as _
from circuits.choices import *
from circuits.constants import *
from circuits.models import *
from dcim.models import Site
from netbox.choices import DistanceUnitChoices
from netbox.forms import NetBoxModelImportForm
from tenancy.models import Tenant
from utilities.forms.fields import CSVChoiceField, CSVModelChoiceField, SlugField
from utilities.forms.fields import CSVChoiceField, CSVContentTypeField, CSVModelChoiceField, SlugField
__all__ = (
'CircuitImportForm',
@ -127,17 +128,10 @@ class BaseCircuitTerminationImportForm(forms.ModelForm):
label=_('Termination'),
choices=CircuitTerminationSideChoices,
)
site = CSVModelChoiceField(
label=_('Site'),
queryset=Site.objects.all(),
to_field_name='name',
required=False
)
provider_network = CSVModelChoiceField(
label=_('Provider network'),
queryset=ProviderNetwork.objects.all(),
to_field_name='name',
required=False
termination_type = CSVContentTypeField(
queryset=ContentType.objects.filter(model__in=CIRCUIT_TERMINATION_TERMINATION_TYPES),
required=False,
label=_('Termination type (app & model)')
)
@ -145,9 +139,12 @@ class CircuitTerminationImportRelatedForm(BaseCircuitTerminationImportForm):
class Meta:
model = CircuitTermination
fields = [
'circuit', 'term_side', 'site', 'provider_network', 'port_speed', 'upstream_speed', 'xconnect_id',
'circuit', 'term_side', 'termination_type', 'termination_id', 'port_speed', 'upstream_speed', 'xconnect_id',
'pp_info', 'description'
]
labels = {
'termination_id': _('Termination ID'),
}
class CircuitTerminationImportForm(NetBoxModelImportForm, BaseCircuitTerminationImportForm):
@ -155,9 +152,12 @@ class CircuitTerminationImportForm(NetBoxModelImportForm, BaseCircuitTermination
class Meta:
model = CircuitTermination
fields = [
'circuit', 'term_side', 'site', 'provider_network', 'port_speed', 'upstream_speed', 'xconnect_id',
'circuit', 'term_side', 'termination_type', 'termination_id', 'port_speed', 'upstream_speed', 'xconnect_id',
'pp_info', 'description', 'tags'
]
labels = {
'termination_id': _('Termination ID'),
}
class CircuitGroupImportForm(NetBoxModelImportForm):

25
netbox/circuits/forms/filtersets.py
View File

@ -3,7 +3,7 @@ from django.utils.translation import gettext as _
from circuits.choices import CircuitCommitRateChoices, CircuitPriorityChoices, CircuitStatusChoices, CircuitTerminationSideChoices
from circuits.models import *
from dcim.models import Region, Site, SiteGroup
from dcim.models import Location, Region, Site, SiteGroup
from ipam.models import ASN
from netbox.choices import DistanceUnitChoices
from netbox.forms import NetBoxModelFilterSetForm
@ -207,18 +207,29 @@ class CircuitTerminationFilterForm(NetBoxModelFilterSetForm):
fieldsets = (
FieldSet('q', 'filter_id', 'tag'),
FieldSet('circuit_id', 'term_side', name=_('Circuit')),
FieldSet('provider_id', 'provider_network_id', name=_('Provider')),
FieldSet('region_id', 'site_group_id', 'site_id', name=_('Location')),
FieldSet('provider_id', name=_('Provider')),
FieldSet('region_id', 'site_group_id', 'site_id', 'location_id', name=_('Termination')),
)
region_id = DynamicModelMultipleChoiceField(
queryset=Region.objects.all(),
required=False,
label=_('Region')
)
site_group_id = DynamicModelMultipleChoiceField(
queryset=SiteGroup.objects.all(),
required=False,
label=_('Site group')
)
site_id = DynamicModelMultipleChoiceField(
queryset=Site.objects.all(),
required=False,
query_params={
'region_id': '$region_id',
'site_group_id': '$site_group_id',
},
label=_('Site')
)
location_id = DynamicModelMultipleChoiceField(
queryset=Location.objects.all(),
required=False,
label=_('Location')
)
circuit_id = DynamicModelMultipleChoiceField(
queryset=Circuit.objects.all(),
required=False,

63
netbox/circuits/forms/model_forms.py
View File

@ -1,14 +1,19 @@
from django.contrib.contenttypes.models import ContentType
from django.core.exceptions import ObjectDoesNotExist
from django.utils.translation import gettext_lazy as _
from circuits.choices import CircuitCommitRateChoices, CircuitTerminationPortSpeedChoices
from circuits.constants import *
from circuits.models import *
from dcim.models import Site
from ipam.models import ASN
from netbox.forms import NetBoxModelForm
from tenancy.forms import TenancyForm
from utilities.forms.fields import CommentField, DynamicModelChoiceField, DynamicModelMultipleChoiceField, SlugField
from utilities.forms.rendering import FieldSet, InlineFields, TabbedGroups
from utilities.forms.widgets import DatePicker, NumberWithOptions
from utilities.forms import get_field_value
from utilities.forms.fields import CommentField, ContentTypeChoiceField, DynamicModelChoiceField, DynamicModelMultipleChoiceField, SlugField
from utilities.forms.rendering import FieldSet, InlineFields
from utilities.forms.widgets import DatePicker, HTMXSelect, NumberWithOptions
from utilities.templatetags.builtins.filters import bettertitle
__all__ = (
'CircuitForm',
@ -144,26 +149,24 @@ class CircuitTerminationForm(NetBoxModelForm):
queryset=Circuit.objects.all(),
selector=True
)
site = DynamicModelChoiceField(
label=_('Site'),
queryset=Site.objects.all(),
termination_type = ContentTypeChoiceField(
queryset=ContentType.objects.filter(model__in=CIRCUIT_TERMINATION_TERMINATION_TYPES),
widget=HTMXSelect(),
required=False,
selector=True
label=_('Termination type')
)
provider_network = DynamicModelChoiceField(
label=_('Provider network'),
queryset=ProviderNetwork.objects.all(),
termination = DynamicModelChoiceField(
label=_('Termination'),
queryset=Site.objects.none(), # Initial queryset
required=False,
disabled=True,
selector=True
)
fieldsets = (
FieldSet(
'circuit', 'term_side', 'description', 'tags',
TabbedGroups(
FieldSet('site', name=_('Site')),
FieldSet('provider_network', name=_('Provider Network')),
),
'termination_type', 'termination',
'mark_connected', name=_('Circuit Termination')
),
FieldSet('port_speed', 'upstream_speed', 'xconnect_id', 'pp_info', name=_('Termination Details')),
@ -172,7 +175,7 @@ class CircuitTerminationForm(NetBoxModelForm):
class Meta:
model = CircuitTermination
fields = [
'circuit', 'term_side', 'site', 'provider_network', 'mark_connected', 'port_speed', 'upstream_speed',
'circuit', 'term_side', 'termination_type', 'mark_connected', 'port_speed', 'upstream_speed',
'xconnect_id', 'pp_info', 'description', 'tags',
]
widgets = {
@ -184,6 +187,36 @@ class CircuitTerminationForm(NetBoxModelForm):
),
}
def __init__(self, *args, **kwargs):
instance = kwargs.get('instance')
initial = kwargs.get('initial', {})
if instance is not None and instance.termination:
initial['termination'] = instance.termination
kwargs['initial'] = initial
super().__init__(*args, **kwargs)
if termination_type_id := get_field_value(self, 'termination_type'):
try:
termination_type = ContentType.objects.get(pk=termination_type_id)
model = termination_type.model_class()
self.fields['termination'].queryset = model.objects.all()
self.fields['termination'].widget.attrs['selector'] = model._meta.label_lower
self.fields['termination'].disabled = False
self.fields['termination'].label = _(bettertitle(model._meta.verbose_name))
except ObjectDoesNotExist:
pass
if self.instance and termination_type_id != self.instance.termination_type_id:
self.initial['termination'] = None
def clean(self):
super().clean()
# Assign the selected termination (if any)
self.instance.termination = self.cleaned_data.get('termination')
class CircuitGroupForm(TenancyForm, NetBoxModelForm):
slug = SlugField()

16
netbox/circuits/graphql/types.py
View File

@ -1,4 +1,4 @@
from typing import Annotated, List
from typing import Annotated, List, Union
import strawberry
import strawberry_django
@ -59,13 +59,21 @@ class ProviderNetworkType(NetBoxObjectType):
@strawberry_django.type(
models.CircuitTermination,
fields='__all__',
exclude=('termination_type', 'termination_id', '_location', '_region', '_site', '_site_group', '_provider_network'),
filters=CircuitTerminationFilter
)
class CircuitTerminationType(CustomFieldsMixin, TagsMixin, CabledObjectMixin, ObjectType):
circuit: Annotated["CircuitType", strawberry.lazy('circuits.graphql.types')]
provider_network: Annotated["ProviderNetworkType", strawberry.lazy('circuits.graphql.types')] | None
site: Annotated["SiteType", strawberry.lazy('dcim.graphql.types')] | None
@strawberry_django.field
def termination(self) -> Annotated[Union[
Annotated["LocationType", strawberry.lazy('dcim.graphql.types')],
Annotated["RegionType", strawberry.lazy('dcim.graphql.types')],
Annotated["SiteGroupType", strawberry.lazy('dcim.graphql.types')],
Annotated["SiteType", strawberry.lazy('dcim.graphql.types')],
Annotated["ProviderNetworkType", strawberry.lazy('circuits.graphql.types')],
], strawberry.union("CircuitTerminationTerminationType")] | None:
return self.termination
@strawberry_django.type(

56
netbox/circuits/migrations/0047_circuittermination__termination.py Normal file
View File

@ -0,0 +1,56 @@
import django.db.models.deletion
from django.db import migrations, models
def copy_site_assignments(apps, schema_editor):
"""
Copy site ForeignKey values to the Termination GFK.
"""
ContentType = apps.get_model('contenttypes', 'ContentType')
CircuitTermination = apps.get_model('circuits', 'CircuitTermination')
Site = apps.get_model('dcim', 'Site')
CircuitTermination.objects.filter(site__isnull=False).update(
termination_type=ContentType.objects.get_for_model(Site),
termination_id=models.F('site_id')
)
ProviderNetwork = apps.get_model('circuits', 'ProviderNetwork')
CircuitTermination.objects.filter(provider_network__isnull=False).update(
termination_type=ContentType.objects.get_for_model(ProviderNetwork),
termination_id=models.F('provider_network_id')
)
class Migration(migrations.Migration):
dependencies = [
('circuits', '0046_charfield_null_choices'),
('contenttypes', '0002_remove_content_type_name'),
('dcim', '0193_poweroutlet_color'),
]
operations = [
migrations.AddField(
model_name='circuittermination',
name='termination_id',
field=models.PositiveBigIntegerField(blank=True, null=True),
),
migrations.AddField(
model_name='circuittermination',
name='termination_type',
field=models.ForeignKey(
blank=True,
limit_choices_to=models.Q(('model__in', ('region', 'sitegroup', 'site', 'location', 'providernetwork'))),
null=True,
on_delete=django.db.models.deletion.PROTECT,
related_name='+',
to='contenttypes.contenttype',
),
),
# Copy over existing site assignments
migrations.RunPython(
code=copy_site_assignments,
reverse_code=migrations.RunPython.noop
),
]

90
netbox/circuits/migrations/0048_circuitterminations_cached_relations.py Normal file
View File

@ -0,0 +1,90 @@
# Generated by Django 5.0.9 on 2024-10-21 17:34
import django.db.models.deletion
from django.db import migrations, models
def populate_denormalized_fields(apps, schema_editor):
"""
Copy site ForeignKey values to the Termination GFK.
"""
CircuitTermination = apps.get_model('circuits', 'CircuitTermination')
terminations = CircuitTermination.objects.filter(site__isnull=False).prefetch_related('site')
for termination in terminations:
termination._region_id = termination.site.region_id
termination._site_group_id = termination.site.group_id
termination._site_id = termination.site_id
# Note: Location cannot be set prior to migration
CircuitTermination.objects.bulk_update(terminations, ['_region', '_site_group', '_site'])
class Migration(migrations.Migration):
dependencies = [
('circuits', '0047_circuittermination__termination'),
]
operations = [
migrations.AddField(
model_name='circuittermination',
name='_location',
field=models.ForeignKey(
blank=True,
null=True,
on_delete=django.db.models.deletion.CASCADE,
related_name='circuit_terminations',
to='dcim.location',
),
),
migrations.AddField(
model_name='circuittermination',
name='_region',
field=models.ForeignKey(
blank=True,
null=True,
on_delete=django.db.models.deletion.CASCADE,
related_name='circuit_terminations',
to='dcim.region',
),
),
migrations.AddField(
model_name='circuittermination',
name='_site',
field=models.ForeignKey(
blank=True,
null=True,
on_delete=django.db.models.deletion.CASCADE,
related_name='circuit_terminations',
to='dcim.site',
),
),
migrations.AddField(
model_name='circuittermination',
name='_site_group',
field=models.ForeignKey(
blank=True,
null=True,
on_delete=django.db.models.deletion.CASCADE,
related_name='circuit_terminations',
to='dcim.sitegroup',
),
),
# Populate denormalized FK values
migrations.RunPython(
code=populate_denormalized_fields,
reverse_code=migrations.RunPython.noop
),
# Delete the site ForeignKey
migrations.RemoveField(
model_name='circuittermination',
name='site',
),
migrations.RenameField(
model_name='circuittermination',
old_name='provider_network',
new_name='_provider_network',
),
]

94
netbox/circuits/models/circuits.py
View File

@ -1,9 +1,13 @@
from django.apps import apps
from django.contrib.contenttypes.fields import GenericForeignKey
from django.core.exceptions import ValidationError
from django.db import models
from django.db.models import Q
from django.urls import reverse
from django.utils.translation import gettext_lazy as _
from circuits.choices import *
from circuits.constants import *
from dcim.models import CabledObjectModel
from netbox.models import ChangeLoggedModel, OrganizationalModel, PrimaryModel
from netbox.models.mixins import DistanceMixin
@ -230,22 +234,24 @@ class CircuitTermination(
term_side = models.CharField(
max_length=1,
choices=CircuitTerminationSideChoices,
verbose_name=_('termination')
verbose_name=_('termination side')
)
site = models.ForeignKey(
to='dcim.Site',
termination_type = models.ForeignKey(
to='contenttypes.ContentType',
on_delete=models.PROTECT,
related_name='circuit_terminations',
limit_choices_to=Q(model__in=CIRCUIT_TERMINATION_TERMINATION_TYPES),
related_name='+',
blank=True,
null=True
)
provider_network = models.ForeignKey(
to='circuits.ProviderNetwork',
on_delete=models.PROTECT,
related_name='circuit_terminations',
termination_id = models.PositiveBigIntegerField(
blank=True,
null=True
)
termination = GenericForeignKey(
ct_field='termination_type',
fk_field='termination_id'
)
port_speed = models.PositiveIntegerField(
verbose_name=_('port speed (Kbps)'),
blank=True,
@ -276,6 +282,43 @@ class CircuitTermination(
blank=True
)
# Cached associations to enable efficient filtering
_provider_network = models.ForeignKey(
to='circuits.ProviderNetwork',
on_delete=models.PROTECT,
related_name='circuit_terminations',
blank=True,
null=True
)
_location = models.ForeignKey(
to='dcim.Location',
on_delete=models.CASCADE,
related_name='circuit_terminations',
blank=True,
null=True
)
_site = models.ForeignKey(
to='dcim.Site',
on_delete=models.CASCADE,
related_name='circuit_terminations',
blank=True,
null=True
)
_region = models.ForeignKey(
to='dcim.Region',
on_delete=models.CASCADE,
related_name='circuit_terminations',
blank=True,
null=True
)
_site_group = models.ForeignKey(
to='dcim.SiteGroup',
on_delete=models.CASCADE,
related_name='circuit_terminations',
blank=True,
null=True
)
class Meta:
ordering = ['circuit', 'term_side']
constraints = (
@ -297,10 +340,35 @@ class CircuitTermination(
super().clean()
# Must define either site *or* provider network
if self.site is None and self.provider_network is None:
raise ValidationError(_("A circuit termination must attach to either a site or a provider network."))
if self.site and self.provider_network:
raise ValidationError(_("A circuit termination cannot attach to both a site and a provider network."))
if self.termination is None:
raise ValidationError(_("A circuit termination must attach to termination."))
def save(self, *args, **kwargs):
# Cache objects associated with the terminating object (for filtering)
self.cache_related_objects()
super().save(*args, **kwargs)
def cache_related_objects(self):
self._provider_network = self._region = self._site_group = self._site = self._location = None
if self.termination_type:
termination_type = self.termination_type.model_class()
if termination_type == apps.get_model('dcim', 'region'):
self._region = self.termination
elif termination_type == apps.get_model('dcim', 'sitegroup'):
self._site_group = self.termination
elif termination_type == apps.get_model('dcim', 'site'):
self._region = self.termination.region
self._site_group = self.termination.group
self._site = self.termination
elif termination_type == apps.get_model('dcim', 'location'):
self._region = self.termination.site.region
self._site_group = self.termination.site.group
self._site = self.termination.site
self._location = self.termination
elif termination_type == apps.get_model('circuits', 'providernetwork'):
self._provider_network = self.termination
cache_related_objects.alters_data = True
def to_objectchange(self, action):
objectchange = super().to_objectchange(action)
@ -314,7 +382,7 @@ class CircuitTermination(
def get_peer_termination(self):
peer_side = 'Z' if self.term_side == 'A' else 'A'
try:
return CircuitTermination.objects.prefetch_related('site').get(
return CircuitTermination.objects.prefetch_related('termination').get(
circuit=self.circuit,
term_side=peer_side
)

52
netbox/circuits/tables/circuits.py
View File

@ -18,10 +18,8 @@ __all__ = (
CIRCUITTERMINATION_LINK = """
{% if value.site %}
<a href="{{ value.site.get_absolute_url }}">{{ value.site }}</a>
{% elif value.provider_network %}
<a href="{{ value.provider_network.get_absolute_url }}">{{ value.provider_network }}</a>
{% if value.termination %}
<a href="{{ value.termination.get_absolute_url }}">{{ value.termination }}</a>
{% endif %}
"""
@ -63,12 +61,12 @@ class CircuitTable(TenancyColumnsMixin, ContactsColumnMixin, NetBoxTable):
verbose_name=_('Account')
)
status = columns.ChoiceFieldColumn()
termination_a = tables.TemplateColumn(
termination_a = columns.TemplateColumn(
template_code=CIRCUITTERMINATION_LINK,
orderable=False,
verbose_name=_('Side A')
)
termination_z = tables.TemplateColumn(
termination_z = columns.TemplateColumn(
template_code=CIRCUITTERMINATION_LINK,
orderable=False,
verbose_name=_('Side Z')
@ -110,22 +108,54 @@ class CircuitTerminationTable(NetBoxTable):
linkify=True,
accessor='circuit.provider'
)
term_side = tables.Column(
verbose_name=_('Side')
)
termination_type = columns.ContentTypeColumn(
verbose_name=_('Termination Type'),
)
termination = tables.Column(
verbose_name=_('Termination Point'),
linkify=True
)
# Termination types
site = tables.Column(
verbose_name=_('Site'),
linkify=True
linkify=True,
accessor='_site'
)
site_group = tables.Column(
verbose_name=_('Site Group'),
linkify=True,
accessor='_sitegroup'
)
region = tables.Column(
verbose_name=_('Region'),
linkify=True,
accessor='_region'
)
location = tables.Column(
verbose_name=_('Location'),
linkify=True,
accessor='_location'
)
provider_network = tables.Column(
verbose_name=_('Provider Network'),
linkify=True
linkify=True,
accessor='_provider_network'
)
class Meta(NetBoxTable.Meta):
model = CircuitTermination
fields = (
'pk', 'id', 'circuit', 'provider', 'term_side', 'site', 'provider_network', 'port_speed', 'upstream_speed',
'xconnect_id', 'pp_info', 'description', 'created', 'last_updated', 'actions',
'pk', 'id', 'circuit', 'provider', 'term_side', 'termination_type', 'termination', 'site_group', 'region',
'site', 'location', 'provider_network', 'port_speed', 'upstream_speed', 'xconnect_id', 'pp_info',
'description', 'created', 'last_updated', 'actions',
)
default_columns = (
'pk', 'id', 'circuit', 'provider', 'term_side', 'termination_type', 'termination', 'description',
)
default_columns = ('pk', 'id', 'circuit', 'provider', 'term_side', 'description')
class CircuitGroupTable(NetBoxTable):

14
netbox/circuits/tests/test_api.py
View File

@ -181,10 +181,10 @@ class CircuitTerminationTest(APIViewTestCases.APIViewTestCase):
Circuit.objects.bulk_create(circuits)
circuit_terminations = (
CircuitTermination(circuit=circuits[0], term_side=SIDE_A, site=sites[0]),
CircuitTermination(circuit=circuits[0], term_side=SIDE_Z, provider_network=provider_networks[0]),
CircuitTermination(circuit=circuits[1], term_side=SIDE_A, site=sites[1]),
CircuitTermination(circuit=circuits[1], term_side=SIDE_Z, provider_network=provider_networks[1]),
CircuitTermination(circuit=circuits[0], term_side=SIDE_A, termination=sites[0]),
CircuitTermination(circuit=circuits[0], term_side=SIDE_Z, termination=provider_networks[0]),
CircuitTermination(circuit=circuits[1], term_side=SIDE_A, termination=sites[1]),
CircuitTermination(circuit=circuits[1], term_side=SIDE_Z, termination=provider_networks[1]),
)
CircuitTermination.objects.bulk_create(circuit_terminations)
@ -192,13 +192,15 @@ class CircuitTerminationTest(APIViewTestCases.APIViewTestCase):
{
'circuit': circuits[2].pk,
'term_side': SIDE_A,
'site': sites[0].pk,
'termination_type': 'dcim.site',
'termination_id': sites[0].pk,
'port_speed': 200000,
},
{
'circuit': circuits[2].pk,
'term_side': SIDE_Z,
'provider_network': provider_networks[0].pk,
'termination_type': 'circuits.providernetwork',
'termination_id': provider_networks[0].pk,
'port_speed': 200000,
},
]

48
netbox/circuits/tests/test_filtersets.py
View File

@ -70,10 +70,12 @@ class ProviderTestCase(TestCase, ChangeLoggedFilterSetTests):
)
Circuit.objects.bulk_create(circuits)
CircuitTermination.objects.bulk_create((
CircuitTermination(circuit=circuits[0], site=sites[0], term_side='A'),
CircuitTermination(circuit=circuits[1], site=sites[0], term_side='A'),
))
circuit_terminations = (
CircuitTermination(circuit=circuits[0], termination=sites[0], term_side='A'),
CircuitTermination(circuit=circuits[1], termination=sites[0], term_side='A'),
)
for ct in circuit_terminations:
ct.save()
def test_q(self):
params = {'q': 'foobar1'}
@ -233,14 +235,15 @@ class CircuitTestCase(TestCase, ChangeLoggedFilterSetTests):
Circuit.objects.bulk_create(circuits)
circuit_terminations = ((
CircuitTermination(circuit=circuits[0], site=sites[0], term_side='A'),
CircuitTermination(circuit=circuits[1], site=sites[1], term_side='A'),
CircuitTermination(circuit=circuits[2], site=sites[2], term_side='A'),
CircuitTermination(circuit=circuits[3], provider_network=provider_networks[0], term_side='A'),
CircuitTermination(circuit=circuits[4], provider_network=provider_networks[1], term_side='A'),
CircuitTermination(circuit=circuits[5], provider_network=provider_networks[2], term_side='A'),
CircuitTermination(circuit=circuits[0], termination=sites[0], term_side='A'),
CircuitTermination(circuit=circuits[1], termination=sites[1], term_side='A'),
CircuitTermination(circuit=circuits[2], termination=sites[2], term_side='A'),
CircuitTermination(circuit=circuits[3], termination=provider_networks[0], term_side='A'),
CircuitTermination(circuit=circuits[4], termination=provider_networks[1], term_side='A'),
CircuitTermination(circuit=circuits[5], termination=provider_networks[2], term_side='A'),
))
CircuitTermination.objects.bulk_create(circuit_terminations)
for ct in circuit_terminations:
ct.save()
def test_q(self):
params = {'q': 'foobar1'}
@ -384,18 +387,19 @@ class CircuitTerminationTestCase(TestCase, ChangeLoggedFilterSetTests):
Circuit.objects.bulk_create(circuits)
circuit_terminations = ((
CircuitTermination(circuit=circuits[0], site=sites[0], term_side='A', port_speed=1000, upstream_speed=1000, xconnect_id='ABC', description='foobar1'),
CircuitTermination(circuit=circuits[0], site=sites[1], term_side='Z', port_speed=1000, upstream_speed=1000, xconnect_id='DEF', description='foobar2'),
CircuitTermination(circuit=circuits[1], site=sites[1], term_side='A', port_speed=2000, upstream_speed=2000, xconnect_id='GHI'),
CircuitTermination(circuit=circuits[1], site=sites[2], term_side='Z', port_speed=2000, upstream_speed=2000, xconnect_id='JKL'),
CircuitTermination(circuit=circuits[2], site=sites[2], term_side='A', port_speed=3000, upstream_speed=3000, xconnect_id='MNO'),
CircuitTermination(circuit=circuits[2], site=sites[0], term_side='Z', port_speed=3000, upstream_speed=3000, xconnect_id='PQR'),
CircuitTermination(circuit=circuits[3], provider_network=provider_networks[0], term_side='A'),
CircuitTermination(circuit=circuits[4], provider_network=provider_networks[1], term_side='A'),
CircuitTermination(circuit=circuits[5], provider_network=provider_networks[2], term_side='A'),
CircuitTermination(circuit=circuits[6], provider_network=provider_networks[0], term_side='A', mark_connected=True),
CircuitTermination(circuit=circuits[0], termination=sites[0], term_side='A', port_speed=1000, upstream_speed=1000, xconnect_id='ABC', description='foobar1'),
CircuitTermination(circuit=circuits[0], termination=sites[1], term_side='Z', port_speed=1000, upstream_speed=1000, xconnect_id='DEF', description='foobar2'),
CircuitTermination(circuit=circuits[1], termination=sites[1], term_side='A', port_speed=2000, upstream_speed=2000, xconnect_id='GHI'),
CircuitTermination(circuit=circuits[1], termination=sites[2], term_side='Z', port_speed=2000, upstream_speed=2000, xconnect_id='JKL'),
CircuitTermination(circuit=circuits[2], termination=sites[2], term_side='A', port_speed=3000, upstream_speed=3000, xconnect_id='MNO'),
CircuitTermination(circuit=circuits[2], termination=sites[0], term_side='Z', port_speed=3000, upstream_speed=3000, xconnect_id='PQR'),
CircuitTermination(circuit=circuits[3], termination=provider_networks[0], term_side='A'),
CircuitTermination(circuit=circuits[4], termination=provider_networks[1], term_side='A'),
CircuitTermination(circuit=circuits[5], termination=provider_networks[2], term_side='A'),
CircuitTermination(circuit=circuits[6], termination=provider_networks[0], term_side='A', mark_connected=True),
))
CircuitTermination.objects.bulk_create(circuit_terminations)
for ct in circuit_terminations:
ct.save()
Cable(a_terminations=[circuit_terminations[0]], b_terminations=[circuit_terminations[1]]).save()

46
netbox/circuits/tests/test_views.py
View File

@ -1,5 +1,6 @@
import datetime
from django.contrib.contenttypes.models import ContentType
from django.test import override_settings
from django.urls import reverse
@ -190,27 +191,31 @@ class CircuitTestCase(ViewTestCases.PrimaryObjectViewTestCase):
@override_settings(EXEMPT_VIEW_PERMISSIONS=['*'], EXEMPT_EXCLUDE_MODELS=[])
def test_bulk_import_objects_with_terminations(self):
json_data = """
site = Site.objects.first()
json_data = f"""
[
{
{{
"cid": "Circuit 7",
"provider": "Provider 1",
"type": "Circuit Type 1",
"status": "active",
"description": "Testing Import",
"terminations": [
{
{{
"term_side": "A",
"site": "Site 1"
},
{
"termination_type": "dcim.site",
"termination_id": "{site.pk}"
}},
{{
"term_side": "Z",
"site": "Site 1"
}
"termination_type": "dcim.site",
"termination_id": "{site.pk}"
}}
]
}
}}
]
"""
initial_count = self._get_queryset().count()
data = {
'data': json_data,
@ -336,7 +341,7 @@ class ProviderNetworkTestCase(ViewTestCases.PrimaryObjectViewTestCase):
}
class CircuitTerminationTestCase(ViewTestCases.PrimaryObjectViewTestCase):
class TestCase(ViewTestCases.PrimaryObjectViewTestCase):
model = CircuitTermination
@classmethod
@ -359,24 +364,27 @@ class CircuitTerminationTestCase(ViewTestCases.PrimaryObjectViewTestCase):
Circuit.objects.bulk_create(circuits)
circuit_terminations = (
CircuitTermination(circuit=circuits[0], term_side='A', site=sites[0]),
CircuitTermination(circuit=circuits[0], term_side='Z', site=sites[1]),
CircuitTermination(circuit=circuits[1], term_side='A', site=sites[0]),
CircuitTermination(circuit=circuits[1], term_side='Z', site=sites[1]),
CircuitTermination(circuit=circuits[0], term_side='A', termination=sites[0]),
CircuitTermination(circuit=circuits[0], term_side='Z', termination=sites[1]),
CircuitTermination(circuit=circuits[1], term_side='A', termination=sites[0]),
CircuitTermination(circuit=circuits[1], term_side='Z', termination=sites[1]),
)
CircuitTermination.objects.bulk_create(circuit_terminations)
for ct in circuit_terminations:
ct.save()
cls.form_data = {
'circuit': circuits[2].pk,
'term_side': 'A',
'site': sites[2].pk,
'termination_type': ContentType.objects.get_for_model(Site).pk,
'termination': sites[2].pk,
'description': 'New description',
}
site = sites[0].pk
cls.csv_data = (
"circuit,term_side,site,description",
"Circuit 3,A,Site 1,Foo",
"Circuit 3,Z,Site 1,Bar",
"circuit,term_side,termination_type,termination_id,description",
f"Circuit 3,A,dcim.site,{site},Foo",
f"Circuit 3,Z,dcim.site,{site},Bar",
)
cls.csv_update_data = (

11
netbox/circuits/views.py
View File

@ -158,7 +158,7 @@ class ProviderNetworkView(GetRelatedModelsMixin, generic.ObjectView):
instance,
extra=(
(
Circuit.objects.restrict(request.user, 'view').filter(terminations__provider_network=instance),
Circuit.objects.restrict(request.user, 'view').filter(terminations___provider_network=instance),
'provider_network_id',
),
),
@ -257,8 +257,7 @@ class CircuitTypeBulkDeleteView(generic.BulkDeleteView):
class CircuitListView(generic.ObjectListView):
queryset = Circuit.objects.prefetch_related(
'tenant__group', 'termination_a__site', 'termination_z__site',
'termination_a__provider_network', 'termination_z__provider_network',
'tenant__group', 'termination_a__termination', 'termination_z__termination',
)
filterset = filtersets.CircuitFilterSet
filterset_form = forms.CircuitFilterForm
@ -298,8 +297,7 @@ class CircuitBulkImportView(generic.BulkImportView):
class CircuitBulkEditView(generic.BulkEditView):
queryset = Circuit.objects.prefetch_related(
'termination_a__site', 'termination_z__site',
'termination_a__provider_network', 'termination_z__provider_network',
'tenant__group', 'termination_a__termination', 'termination_z__termination',
)
filterset = filtersets.CircuitFilterSet
table = tables.CircuitTable
@ -308,8 +306,7 @@ class CircuitBulkEditView(generic.BulkEditView):
class CircuitBulkDeleteView(generic.BulkDeleteView):
queryset = Circuit.objects.prefetch_related(
'termination_a__site', 'termination_z__site',
'termination_a__provider_network', 'termination_z__provider_network',
'tenant__group', 'termination_a__termination', 'termination_z__termination',
)
filterset = filtersets.CircuitFilterSet
table = tables.CircuitTable

9
netbox/dcim/api/serializers_/device_components.py
View File

@ -205,6 +205,7 @@ class InterfaceSerializer(NetBoxModelSerializer, CabledObjectSerializer, Connect
required=False,
many=True
)
qinq_svlan = VLANSerializer(nested=True, required=False, allow_null=True)
vlan_translation_policy = VLANTranslationPolicySerializer(nested=True, required=False, allow_null=True)
vrf = VRFSerializer(nested=True, required=False, allow_null=True)
l2vpn_termination = L2VPNTerminationSerializer(nested=True, read_only=True, allow_null=True)
@ -233,10 +234,10 @@ class InterfaceSerializer(NetBoxModelSerializer, CabledObjectSerializer, Connect
'id', 'url', 'display_url', 'display', 'device', 'vdcs', 'module', 'name', 'label', 'type', 'enabled',
'parent', 'bridge', 'lag', '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', 'cable_end', 'wireless_link',
'link_peers', 'link_peers_type', 'wireless_lans', 'vrf', 'l2vpn_termination', 'connected_endpoints',
'connected_endpoints_type', 'connected_endpoints_reachable', 'tags', 'custom_fields', 'created',
'last_updated', 'count_ipaddresses', 'count_fhrp_groups', '_occupied', 'vlan_translation_policy'
'tx_power', 'untagged_vlan', 'tagged_vlans', 'qinq_svlan', 'vlan_translation_policy', 'mark_connected',
'cable', 'cable_end', 'wireless_link', 'link_peers', 'link_peers_type', 'wireless_lans', 'vrf',
'l2vpn_termination', 'connected_endpoints', 'connected_endpoints_type', 'connected_endpoints_reachable',
'tags', 'custom_fields', 'created', 'last_updated', 'count_ipaddresses', 'count_fhrp_groups', '_occupied',
]
brief_fields = ('id', 'url', 'display', 'device', 'name', 'description', 'cable', '_occupied')

2
netbox/dcim/choices.py
View File

@ -1258,11 +1258,13 @@ class InterfaceModeChoices(ChoiceSet):
MODE_ACCESS = 'access'
MODE_TAGGED = 'tagged'
MODE_TAGGED_ALL = 'tagged-all'
MODE_Q_IN_Q = 'q-in-q'
CHOICES = (
(MODE_ACCESS, _('Access')),
(MODE_TAGGED, _('Tagged')),
(MODE_TAGGED_ALL, _('Tagged (All)')),
(MODE_Q_IN_Q, _('Q-in-Q (802.1ad)')),
)

6
netbox/dcim/filtersets.py
View File

@ -1796,7 +1796,8 @@ class CommonInterfaceFilterSet(django_filters.FilterSet):
return queryset
return queryset.filter(
Q(untagged_vlan_id=value) |
Q(tagged_vlans=value)
Q(tagged_vlans=value) |
Q(qinq_svlan=value)
)
def filter_vlan(self, queryset, name, value):
@ -1805,7 +1806,8 @@ class CommonInterfaceFilterSet(django_filters.FilterSet):
return queryset
return queryset.filter(
Q(untagged_vlan_id__vid=value) |
Q(tagged_vlans__vid=value)
Q(tagged_vlans__vid=value) |
Q(qinq_svlan__vid=value)
)

2
netbox/dcim/forms/common.py
View File

@ -37,6 +37,8 @@ class InterfaceCommonForm(forms.Form):
del self.fields['vlan_group']
del self.fields['untagged_vlan']
del self.fields['tagged_vlans']
if interface_mode != InterfaceModeChoices.MODE_Q_IN_Q:
del self.fields['qinq_svlan']
def clean(self):
super().clean()

18
netbox/dcim/forms/model_forms.py
View File

@ -8,6 +8,7 @@ from dcim.choices import *
from dcim.constants import *
from dcim.models import *
from extras.models import ConfigTemplate
from ipam.choices import VLANQinQRoleChoices
from ipam.models import ASN, IPAddress, VLAN, VLANGroup, VLANTranslationPolicy, VRF
from netbox.forms import NetBoxModelForm
from tenancy.forms import TenancyForm
@ -1489,6 +1490,16 @@ class InterfaceForm(InterfaceCommonForm, ModularDeviceComponentForm):
'available_on_device': '$device',
}
)
qinq_svlan = DynamicModelMultipleChoiceField(
queryset=VLAN.objects.all(),
required=False,
label=_('Q-in-Q Service VLAN'),
query_params={
'group_id': '$vlan_group',
'available_on_device': '$device',
'qinq_role': VLANQinQRoleChoices.ROLE_SERVICE,
}
)
vrf = DynamicModelChoiceField(
queryset=VRF.objects.all(),
required=False,
@ -1513,7 +1524,10 @@ class InterfaceForm(InterfaceCommonForm, ModularDeviceComponentForm):
FieldSet('vdcs', 'mtu', 'tx_power', 'enabled', 'mgmt_only', 'mark_connected', name=_('Operation')),
FieldSet('parent', 'bridge', 'lag', name=_('Related Interfaces')),
FieldSet('poe_mode', 'poe_type', name=_('PoE')),
FieldSet('mode', 'vlan_group', 'untagged_vlan', 'tagged_vlans', 'vlan_translation_policy', name=_('802.1Q Switching')),
FieldSet(
'mode', 'vlan_group', 'untagged_vlan', 'tagged_vlans', 'qinq_svlan', 'vlan_translation_policy',
name=_('802.1Q Switching')
),
FieldSet(
'rf_role', 'rf_channel', 'rf_channel_frequency', 'rf_channel_width', 'wireless_lan_group', 'wireless_lans',
name=_('Wireless')
@ -1526,7 +1540,7 @@ class InterfaceForm(InterfaceCommonForm, ModularDeviceComponentForm):
'device', 'module', 'vdcs', 'name', 'label', 'type', 'speed', 'duplex', 'enabled', 'parent', 'bridge', 'lag',
'mac_address', 'wwn', 'mtu', 'mgmt_only', 'mark_connected', 'description', 'poe_mode', 'poe_type', 'mode',
'rf_role', 'rf_channel', 'rf_channel_frequency', 'rf_channel_width', 'tx_power', 'wireless_lans',
'untagged_vlan', 'tagged_vlans', 'vrf', 'tags', 'vlan_translation_policy',
'untagged_vlan', 'tagged_vlans', 'qinq_svlan', 'vlan_translation_policy', 'vrf', 'tags',
]
widgets = {
'speed': NumberWithOptions(

18
netbox/dcim/graphql/types.py
View File

@ -394,6 +394,7 @@ class InterfaceType(IPAddressesMixin, ModularComponentType, CabledObjectMixin, P
wireless_link: Annotated["WirelessLinkType", strawberry.lazy('wireless.graphql.types')] | None
untagged_vlan: Annotated["VLANType", strawberry.lazy('ipam.graphql.types')] | None
vrf: Annotated["VRFType", strawberry.lazy('ipam.graphql.types')] | None
qinq_svlan: Annotated["VLANType", strawberry.lazy('ipam.graphql.types')] | None
vlan_translation_policy: Annotated["VLANTranslationPolicyType", strawberry.lazy('ipam.graphql.types')] | None
vdcs: List[Annotated["VirtualDeviceContextType", strawberry.lazy('dcim.graphql.types')]]
@ -472,6 +473,10 @@ class LocationType(VLANGroupsMixin, ImageAttachmentsMixin, ContactsMixin, Organi
devices: List[Annotated["DeviceType", strawberry.lazy('dcim.graphql.types')]]
children: List[Annotated["LocationType", strawberry.lazy('dcim.graphql.types')]]
@strawberry_django.field
def circuit_terminations(self) -> List[Annotated["CircuitTerminationType", strawberry.lazy('circuits.graphql.types')]]:
return self.circuit_terminations.all()
@strawberry_django.type(
models.Manufacturer,
@ -715,6 +720,10 @@ class RegionType(VLANGroupsMixin, ContactsMixin, OrganizationalObjectType):
def parent(self) -> Annotated["RegionType", strawberry.lazy('dcim.graphql.types')] | None:
return self.parent
@strawberry_django.field
def circuit_terminations(self) -> List[Annotated["CircuitTerminationType", strawberry.lazy('circuits.graphql.types')]]:
return self.circuit_terminations.all()
@strawberry_django.type(
models.Site,
@ -736,10 +745,13 @@ class SiteType(VLANGroupsMixin, ImageAttachmentsMixin, ContactsMixin, NetBoxObje
devices: List[Annotated["DeviceType", strawberry.lazy('dcim.graphql.types')]]
locations: List[Annotated["LocationType", strawberry.lazy('dcim.graphql.types')]]
asns: List[Annotated["ASNType", strawberry.lazy('ipam.graphql.types')]]
circuit_terminations: List[Annotated["CircuitTerminationType", strawberry.lazy('circuits.graphql.types')]]
clusters: List[Annotated["ClusterType", strawberry.lazy('virtualization.graphql.types')]]
vlans: List[Annotated["VLANType", strawberry.lazy('ipam.graphql.types')]]
@strawberry_django.field
def circuit_terminations(self) -> List[Annotated["CircuitTerminationType", strawberry.lazy('circuits.graphql.types')]]:
return self.circuit_terminations.all()
@strawberry_django.type(
models.SiteGroup,
@ -756,6 +768,10 @@ class SiteGroupType(VLANGroupsMixin, ContactsMixin, OrganizationalObjectType):
def parent(self) -> Annotated["SiteGroupType", strawberry.lazy('dcim.graphql.types')] | None:
return self.parent
@strawberry_django.field
def circuit_terminations(self) -> List[Annotated["CircuitTerminationType", strawberry.lazy('circuits.graphql.types')]]:
return self.circuit_terminations.all()
@strawberry_django.type(
models.VirtualChassis,

28
netbox/dcim/migrations/0196_qinq_svlan.py Normal file
View File

@ -0,0 +1,28 @@
import django.db.models.deletion
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('dcim', '0195_interface_vlan_translation_policy'),
('ipam', '0075_vlan_qinq'),
]
operations = [
migrations.AddField(
model_name='interface',
name='qinq_svlan',
field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='%(class)ss_svlan', to='ipam.vlan'),
),
migrations.AlterField(
model_name='interface',
name='tagged_vlans',
field=models.ManyToManyField(blank=True, related_name='%(class)ss_as_tagged', to='ipam.vlan'),
),
migrations.AlterField(
model_name='interface',
name='untagged_vlan',
field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='%(class)ss_as_untagged', to='ipam.vlan'),
),
]

6
netbox/dcim/migrations/0196_rename_mac_address_interface__mac_address_macaddress.py → netbox/dcim/migrations/0197_rename_mac_address_interface__mac_address_macaddress.py
View File

@ -1,5 +1,3 @@
# Generated by Django 5.1.2 on 2024-10-29 15:05
import dcim.fields
import django.db.models.deletion
import taggit.managers
@ -42,9 +40,9 @@ def populate_macaddress_objects(apps, schema_editor):
class Migration(migrations.Migration):
dependencies = [
('dcim', '0195_interface_vlan_translation_policy'),
('dcim', '0196_qinq_svlan'),
('extras', '0122_charfield_null_choices'),
('virtualization', '0043_rename_mac_address_vminterface__mac_address'),
('virtualization', '0044_rename_mac_address_vminterface__mac_address'),
]
operations = [

4
netbox/dcim/migrations/0197_remove_interface__mac_address.py → netbox/dcim/migrations/0198_remove_interface__mac_address.py
View File

@ -1,12 +1,10 @@
# Generated by Django 5.1.2 on 2024-10-29 15:41
from django.db import migrations
class Migration(migrations.Migration):
dependencies = [
('dcim', '0196_rename_mac_address_interface__mac_address_macaddress'),
('dcim', '0197_rename_mac_address_interface__mac_address_macaddress'),
]
operations = [

12
netbox/dcim/models/cables.py
View File

@ -344,7 +344,7 @@ class CableTermination(ChangeLoggedModel):
)
# A CircuitTermination attached to a ProviderNetwork cannot have a Cable
if self.termination_type.model == 'circuittermination' and self.termination.provider_network is not None:
if self.termination_type.model == 'circuittermination' and self.termination._provider_network is not None:
raise ValidationError(_("Circuit terminations attached to a provider network may not be cabled."))
def save(self, *args, **kwargs):
@ -690,19 +690,19 @@ class CablePath(models.Model):
).first()
if circuit_termination is None:
break
elif circuit_termination.provider_network:
elif circuit_termination._provider_network:
# Circuit terminates to a ProviderNetwork
path.extend([
[object_to_path_node(circuit_termination)],
[object_to_path_node(circuit_termination.provider_network)],
[object_to_path_node(circuit_termination._provider_network)],
])
is_complete = True
break
elif circuit_termination.site and not circuit_termination.cable:
# Circuit terminates to a Site
elif circuit_termination.termination and not circuit_termination.cable:
# Circuit terminates to a Region/Site/etc.
path.extend([
[object_to_path_node(circuit_termination)],
[object_to_path_node(circuit_termination.site)],
[object_to_path_node(circuit_termination.termination)],
])
break

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

@ -543,17 +543,48 @@ class BaseInterface(models.Model):
blank=True,
verbose_name=_('bridge interface')
)
untagged_vlan = models.ForeignKey(
to='ipam.VLAN',
on_delete=models.SET_NULL,
related_name='%(class)ss_as_untagged',
null=True,
blank=True,
verbose_name=_('untagged VLAN')
)
tagged_vlans = models.ManyToManyField(
to='ipam.VLAN',
related_name='%(class)ss_as_tagged',
blank=True,
verbose_name=_('tagged VLANs')
)
qinq_svlan = models.ForeignKey(
to='ipam.VLAN',
on_delete=models.SET_NULL,
related_name='%(class)ss_svlan',
null=True,
blank=True,
verbose_name=_('Q-in-Q SVLAN')
)
vlan_translation_policy = models.ForeignKey(
to='ipam.VLANTranslationPolicy',
on_delete=models.PROTECT,
null=True,
blank=True,
verbose_name=_('VLAN Translation Policy'),
verbose_name=_('VLAN Translation Policy')
)
class Meta:
abstract = True
def clean(self):
super().clean()
# SVLAN can be defined only for Q-in-Q interfaces
if self.qinq_svlan and self.mode != InterfaceModeChoices.MODE_Q_IN_Q:
raise ValidationError({
'qinq_svlan': _("Only Q-in-Q interfaces may specify a service VLAN.")
})
def save(self, *args, **kwargs):
# Remove untagged VLAN assignment for non-802.1Q interfaces
@ -699,20 +730,6 @@ class Interface(ModularComponentModel, BaseInterface, CabledObjectModel, PathEnd
blank=True,
verbose_name=_('wireless LANs')
)
untagged_vlan = models.ForeignKey(
to='ipam.VLAN',
on_delete=models.SET_NULL,
related_name='interfaces_as_untagged',
null=True,
blank=True,
verbose_name=_('untagged VLAN')
)
tagged_vlans = models.ManyToManyField(
to='ipam.VLAN',
related_name='interfaces_as_tagged',
blank=True,
verbose_name=_('tagged VLANs')
)
vrf = models.ForeignKey(
to='ipam.VRF',
on_delete=models.SET_NULL,

16
netbox/dcim/tables/devices.py
View File

@ -586,6 +586,10 @@ class BaseInterfaceTable(NetBoxTable):
orderable=False,
verbose_name=_('Tagged VLANs')
)
qinq_svlan = tables.Column(
verbose_name=_('Q-in-Q SVLAN'),
linkify=True
)
def value_ip_addresses(self, value):
return ",".join([str(obj.address) for obj in value.all()])
@ -673,11 +677,11 @@ class InterfaceTable(ModularDeviceComponentTable, BaseInterfaceTable, PathEndpoi
model = models.Interface
fields = (
'pk', 'id', 'name', 'device', 'module_bay', 'module', 'label', 'enabled', 'type', 'mgmt_only', 'mtu',
'speed', 'speed_formatted', 'duplex', 'mode', 'mac_address', 'wwn', 'poe_mode', 'poe_type', 'rf_role', 'rf_channel',
'rf_channel_frequency', 'rf_channel_width', 'tx_power', 'description', 'mark_connected', 'cable',
'cable_color', 'wireless_link', 'wireless_lans', 'link_peer', 'connection', 'tags', 'vdcs', 'vrf', 'l2vpn',
'tunnel', 'ip_addresses', 'fhrp_groups', 'untagged_vlan', 'tagged_vlans', 'inventory_items', 'created',
'last_updated',
'speed', 'speed_formatted', 'duplex', 'mode', 'mac_address', 'wwn', 'poe_mode', 'poe_type', 'rf_role',
'rf_channel', 'rf_channel_frequency', 'rf_channel_width', 'tx_power', 'description', 'mark_connected',
'cable', 'cable_color', 'wireless_link', 'wireless_lans', 'link_peer', 'connection', 'tags', 'vdcs', 'vrf',
'l2vpn', 'tunnel', 'ip_addresses', 'fhrp_groups', 'untagged_vlan', 'tagged_vlans', 'qinq_svlan',
'inventory_items', 'created', 'last_updated',
)
default_columns = ('pk', 'name', 'device', 'label', 'enabled', 'type', 'description')
@ -714,7 +718,7 @@ class DeviceInterfaceTable(InterfaceTable):
'mgmt_only', 'mtu', 'mode', 'mac_address', 'wwn', 'rf_role', 'rf_channel', 'rf_channel_frequency',
'rf_channel_width', 'tx_power', 'description', 'mark_connected', 'cable', 'cable_color', 'wireless_link',
'wireless_lans', 'link_peer', 'connection', 'tags', 'vdcs', 'vrf', 'l2vpn', 'tunnel', 'ip_addresses',
'fhrp_groups', 'untagged_vlan', 'tagged_vlans', 'actions',
'fhrp_groups', 'untagged_vlan', 'tagged_vlans', 'qinq_svlan', 'actions',
)
default_columns = (
'pk', 'name', 'label', 'enabled', 'type', 'parent', 'lag', 'mtu', 'mode', 'description', 'ip_addresses',

6
netbox/dcim/tests/test_api.py
View File

@ -7,6 +7,7 @@ from dcim.choices import *
from dcim.constants import *
from dcim.models import *
from extras.models import ConfigTemplate
from ipam.choices import VLANQinQRoleChoices
from ipam.models import ASN, RIR, VLAN, VRF
from netbox.api.serializers import GenericObjectSerializer
from tenancy.models import Tenant
@ -1618,6 +1619,7 @@ class InterfaceTest(Mixins.ComponentTraceMixin, APIViewTestCases.APIViewTestCase
VLAN(name='VLAN 1', vid=1),
VLAN(name='VLAN 2', vid=2),
VLAN(name='VLAN 3', vid=3),
VLAN(name='SVLAN 1', vid=1001, qinq_role=VLANQinQRoleChoices.ROLE_SERVICE),
)
VLAN.objects.bulk_create(vlans)
@ -1676,18 +1678,22 @@ class InterfaceTest(Mixins.ComponentTraceMixin, APIViewTestCases.APIViewTestCase
'vdcs': [vdcs[1].pk],
'name': 'Interface 7',
'type': InterfaceTypeChoices.TYPE_80211A,
'mode': InterfaceModeChoices.MODE_Q_IN_Q,
'tx_power': 10,
'wireless_lans': [wireless_lans[0].pk, wireless_lans[1].pk],
'rf_channel': WirelessChannelChoices.CHANNEL_5G_32,
'qinq_svlan': vlans[3].pk,
},
{
'device': device.pk,
'vdcs': [vdcs[1].pk],
'name': 'Interface 8',
'type': InterfaceTypeChoices.TYPE_80211A,
'mode': InterfaceModeChoices.MODE_Q_IN_Q,
'tx_power': 10,
'wireless_lans': [wireless_lans[0].pk, wireless_lans[1].pk],
'rf_channel': "",
'qinq_svlan': vlans[3].pk,
},
]

30
netbox/dcim/tests/test_cablepaths.py
View File

@ -1167,7 +1167,7 @@ class CablePathTestCase(TestCase):
[IF1] --C1-- [CT1]
"""
interface1 = Interface.objects.create(device=self.device, name='Interface 1')
circuittermination1 = CircuitTermination.objects.create(circuit=self.circuit, site=self.site, term_side='A')
circuittermination1 = CircuitTermination.objects.create(circuit=self.circuit, termination=self.site, term_side='A')
# Create cable 1
cable1 = Cable(
@ -1198,7 +1198,7 @@ class CablePathTestCase(TestCase):
"""
interface1 = Interface.objects.create(device=self.device, name='Interface 1')
interface2 = Interface.objects.create(device=self.device, name='Interface 2')
circuittermination1 = CircuitTermination.objects.create(circuit=self.circuit, site=self.site, term_side='A')
circuittermination1 = CircuitTermination.objects.create(circuit=self.circuit, termination=self.site, term_side='A')
# Create cable 1
cable1 = Cable(
@ -1214,7 +1214,7 @@ class CablePathTestCase(TestCase):
)
# Create CT2
circuittermination2 = CircuitTermination.objects.create(circuit=self.circuit, site=self.site, term_side='Z')
circuittermination2 = CircuitTermination.objects.create(circuit=self.circuit, termination=self.site, term_side='Z')
# Check for partial path to site
self.assertPathExists(
@ -1266,7 +1266,7 @@ class CablePathTestCase(TestCase):
interface2 = Interface.objects.create(device=self.device, name='Interface 2')
interface3 = Interface.objects.create(device=self.device, name='Interface 3')
interface4 = Interface.objects.create(device=self.device, name='Interface 4')
circuittermination1 = CircuitTermination.objects.create(circuit=self.circuit, site=self.site, term_side='A')
circuittermination1 = CircuitTermination.objects.create(circuit=self.circuit, termination=self.site, term_side='A')
# Create cable 1
cable1 = Cable(
@ -1282,7 +1282,7 @@ class CablePathTestCase(TestCase):
)
# Create CT2
circuittermination2 = CircuitTermination.objects.create(circuit=self.circuit, site=self.site, term_side='Z')
circuittermination2 = CircuitTermination.objects.create(circuit=self.circuit, termination=self.site, term_side='Z')
# Check for partial path to site
self.assertPathExists(
@ -1335,8 +1335,8 @@ class CablePathTestCase(TestCase):
"""
interface1 = Interface.objects.create(device=self.device, name='Interface 1')
site2 = Site.objects.create(name='Site 2', slug='site-2')
circuittermination1 = CircuitTermination.objects.create(circuit=self.circuit, site=self.site, term_side='A')
circuittermination2 = CircuitTermination.objects.create(circuit=self.circuit, site=site2, term_side='Z')
circuittermination1 = CircuitTermination.objects.create(circuit=self.circuit, termination=self.site, term_side='A')
circuittermination2 = CircuitTermination.objects.create(circuit=self.circuit, termination=site2, term_side='Z')
# Create cable 1
cable1 = Cable(
@ -1365,8 +1365,8 @@ class CablePathTestCase(TestCase):
"""
interface1 = Interface.objects.create(device=self.device, name='Interface 1')
providernetwork = ProviderNetwork.objects.create(name='Provider Network 1', provider=self.circuit.provider)
circuittermination1 = CircuitTermination.objects.create(circuit=self.circuit, site=self.site, term_side='A')
circuittermination2 = CircuitTermination.objects.create(circuit=self.circuit, provider_network=providernetwork, term_side='Z')
circuittermination1 = CircuitTermination.objects.create(circuit=self.circuit, termination=self.site, term_side='A')
circuittermination2 = CircuitTermination.objects.create(circuit=self.circuit, termination=providernetwork, term_side='Z')
# Create cable 1
cable1 = Cable(
@ -1413,8 +1413,8 @@ class CablePathTestCase(TestCase):
frontport2_2 = FrontPort.objects.create(
device=self.device, name='Front Port 2:2', rear_port=rearport2, rear_port_position=2
)
circuittermination1 = CircuitTermination.objects.create(circuit=self.circuit, site=self.site, term_side='A')
circuittermination2 = CircuitTermination.objects.create(circuit=self.circuit, site=self.site, term_side='Z')
circuittermination1 = CircuitTermination.objects.create(circuit=self.circuit, termination=self.site, term_side='A')
circuittermination2 = CircuitTermination.objects.create(circuit=self.circuit, termination=self.site, term_side='Z')
# Create cables
cable1 = Cable(
@ -1499,10 +1499,10 @@ class CablePathTestCase(TestCase):
interface1 = Interface.objects.create(device=self.device, name='Interface 1')
interface2 = Interface.objects.create(device=self.device, name='Interface 2')
circuit2 = Circuit.objects.create(provider=self.circuit.provider, type=self.circuit.type, cid='Circuit 2')
circuittermination1 = CircuitTermination.objects.create(circuit=self.circuit, site=self.site, term_side='A')
circuittermination2 = CircuitTermination.objects.create(circuit=self.circuit, site=self.site, term_side='Z')
circuittermination3 = CircuitTermination.objects.create(circuit=circuit2, site=self.site, term_side='A')
circuittermination4 = CircuitTermination.objects.create(circuit=circuit2, site=self.site, term_side='Z')
circuittermination1 = CircuitTermination.objects.create(circuit=self.circuit, termination=self.site, term_side='A')
circuittermination2 = CircuitTermination.objects.create(circuit=self.circuit, termination=self.site, term_side='Z')
circuittermination3 = CircuitTermination.objects.create(circuit=circuit2, termination=self.site, term_side='A')
circuittermination4 = CircuitTermination.objects.create(circuit=circuit2, termination=self.site, term_side='Z')
# Create cables
cable1 = Cable(

35
netbox/dcim/tests/test_filtersets.py
View File

@ -4,7 +4,8 @@ from circuits.models import Circuit, CircuitTermination, CircuitType, Provider
from dcim.choices import *
from dcim.filtersets import *
from dcim.models import *
from ipam.models import ASN, IPAddress, RIR, VLANTranslationPolicy, VRF
from ipam.choices import VLANQinQRoleChoices
from ipam.models import ASN, IPAddress, RIR, VLAN, VLANTranslationPolicy, VRF
from netbox.choices import ColorChoices, WeightUnitChoices
from tenancy.models import Tenant, TenantGroup
from users.models import User
@ -3527,7 +3528,7 @@ class PowerOutletTestCase(TestCase, DeviceComponentFilterSetTests, ChangeLoggedF
class InterfaceTestCase(TestCase, DeviceComponentFilterSetTests, ChangeLoggedFilterSetTests):
queryset = Interface.objects.all()
filterset = InterfaceFilterSet
ignore_fields = ('tagged_vlans', 'untagged_vlan', 'vdcs')
ignore_fields = ('tagged_vlans', 'untagged_vlan', 'qinq_svlan', 'vdcs')
@classmethod
def setUpTestData(cls):
@ -3683,6 +3684,13 @@ class InterfaceTestCase(TestCase, DeviceComponentFilterSetTests, ChangeLoggedFil
)
MACAddress.objects.bulk_create(mac_addresses)
vlans = (
VLAN(name='SVLAN 1', vid=1001, qinq_role=VLANQinQRoleChoices.ROLE_SERVICE),
VLAN(name='SVLAN 2', vid=1002, qinq_role=VLANQinQRoleChoices.ROLE_SERVICE),
VLAN(name='SVLAN 3', vid=1003, qinq_role=VLANQinQRoleChoices.ROLE_SERVICE),
)
VLAN.objects.bulk_create(vlans)
vlan_translation_policies = (
VLANTranslationPolicy(name='Policy 1'),
VLANTranslationPolicy(name='Policy 2'),
@ -3764,6 +3772,8 @@ class InterfaceTestCase(TestCase, DeviceComponentFilterSetTests, ChangeLoggedFil
duplex='full',
poe_mode=InterfacePoEModeChoices.MODE_PD,
poe_type=InterfacePoETypeChoices.TYPE_2_8023AT,
mode=InterfaceModeChoices.MODE_Q_IN_Q,
qinq_svlan=vlans[0],
vlan_translation_policy=vlan_translation_policies[1],
),
Interface(
@ -3773,7 +3783,9 @@ class InterfaceTestCase(TestCase, DeviceComponentFilterSetTests, ChangeLoggedFil
type=InterfaceTypeChoices.TYPE_OTHER,
enabled=True,
mgmt_only=True,
tx_power=40
tx_power=40,
mode=InterfaceModeChoices.MODE_Q_IN_Q,
qinq_svlan=vlans[1]
),
Interface(
device=devices[4],
@ -3782,7 +3794,9 @@ class InterfaceTestCase(TestCase, DeviceComponentFilterSetTests, ChangeLoggedFil
type=InterfaceTypeChoices.TYPE_OTHER,
enabled=False,
mgmt_only=False,
tx_power=40
tx_power=40,
mode=InterfaceModeChoices.MODE_Q_IN_Q,
qinq_svlan=vlans[2]
),
Interface(
device=devices[4],
@ -4042,6 +4056,13 @@ class InterfaceTestCase(TestCase, DeviceComponentFilterSetTests, ChangeLoggedFil
params = {'vdc_identifier': vdc.values_list('identifier', flat=True)}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 3)
def test_vlan(self):
vlan = VLAN.objects.filter(qinq_role=VLANQinQRoleChoices.ROLE_SERVICE).first()
params = {'vlan_id': vlan.pk}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1)
params = {'vlan': vlan.vid}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1)
def test_vlan_translation_policy(self):
vlan_translation_policies = VLANTranslationPolicy.objects.all()[:2]
params = {'vlan_translation_policy_id': [vlan_translation_policies[0].pk, vlan_translation_policies[1].pk]}
@ -5150,7 +5171,7 @@ class CableTestCase(TestCase, ChangeLoggedFilterSetTests):
provider = Provider.objects.create(name='Provider 1', slug='provider-1')
circuit_type = CircuitType.objects.create(name='Circuit Type 1', slug='circuit-type-1')
circuit = Circuit.objects.create(cid='Circuit 1', provider=provider, type=circuit_type)
circuit_termination = CircuitTermination.objects.create(circuit=circuit, term_side='A', site=sites[0])
circuit_termination = CircuitTermination.objects.create(circuit=circuit, term_side='A', termination=sites[0])
# Cables
cables = (
@ -5323,9 +5344,9 @@ class CableTestCase(TestCase, ChangeLoggedFilterSetTests):
def test_site(self):
site = Site.objects.all()[:2]
params = {'site_id': [site[0].pk, site[1].pk]}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 12)
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 11)
params = {'site': [site[0].slug, site[1].slug]}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 12)
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 11)
def test_tenant(self):
tenant = Tenant.objects.all()[:2]

6
netbox/dcim/tests/test_models.py
View File

@ -762,9 +762,9 @@ class CableTestCase(TestCase):
circuittype = CircuitType.objects.create(name='Circuit Type 1', slug='circuit-type-1')
circuit1 = Circuit.objects.create(provider=provider, type=circuittype, cid='1')
circuit2 = Circuit.objects.create(provider=provider, type=circuittype, cid='2')
CircuitTermination.objects.create(circuit=circuit1, site=site, term_side='A')
CircuitTermination.objects.create(circuit=circuit1, site=site, term_side='Z')
CircuitTermination.objects.create(circuit=circuit2, provider_network=provider_network, term_side='A')
CircuitTermination.objects.create(circuit=circuit1, termination=site, term_side='A')
CircuitTermination.objects.create(circuit=circuit1, termination=site, term_side='Z')
CircuitTermination.objects.create(circuit=circuit2, termination=provider_network, term_side='A')
def test_cable_creation(self):
"""

26
netbox/dcim/views.py
View File

@ -242,6 +242,10 @@ class RegionView(GetRelatedModelsMixin, generic.ObjectView):
extra=(
(Location.objects.restrict(request.user, 'view').filter(site__region__in=regions), 'region_id'),
(Rack.objects.restrict(request.user, 'view').filter(site__region__in=regions), 'region_id'),
(
Circuit.objects.restrict(request.user, 'view').filter(terminations___region=instance).distinct(),
'region_id'
),
),
),
}
@ -324,6 +328,10 @@ class SiteGroupView(GetRelatedModelsMixin, generic.ObjectView):
extra=(
(Location.objects.restrict(request.user, 'view').filter(site__group__in=groups), 'site_group_id'),
(Rack.objects.restrict(request.user, 'view').filter(site__group__in=groups), 'site_group_id'),
(
Circuit.objects.restrict(request.user, 'view').filter(terminations___site_group=instance).distinct(),
'site_group_id'
),
),
),
}
@ -404,8 +412,10 @@ class SiteView(GetRelatedModelsMixin, generic.ObjectView):
scope_id=instance.pk
), 'site'),
(ASN.objects.restrict(request.user, 'view').filter(sites=instance), 'site_id'),
(Circuit.objects.restrict(request.user, 'view').filter(terminations__site=instance).distinct(),
'site_id'),
(
Circuit.objects.restrict(request.user, 'view').filter(terminations___site=instance).distinct(),
'site_id'
),
),
),
}
@ -475,7 +485,17 @@ class LocationView(GetRelatedModelsMixin, generic.ObjectView):
def get_extra_context(self, request, instance):
locations = instance.get_descendants(include_self=True)
return {
'related_models': self.get_related_models(request, locations, [CableTermination]),
'related_models': self.get_related_models(
request,
locations,
[CableTermination],
(
(
Circuit.objects.restrict(request.user, 'view').filter(terminations___location=instance).distinct(),
'location_id'
),
),
),
}

8
netbox/ipam/api/serializers_/nested.py
View File

@ -6,6 +6,7 @@ from ..field_serializers import IPAddressField
__all__ = (
'NestedIPAddressSerializer',
'NestedVLANSerializer',
)
@ -16,3 +17,10 @@ class NestedIPAddressSerializer(WritableNestedSerializer):
class Meta:
model = models.IPAddress
fields = ['id', 'url', 'display_url', 'display', 'family', 'address']
class NestedVLANSerializer(WritableNestedSerializer):
class Meta:
model = models.VLAN
fields = ['id', 'url', 'display', 'vid', 'name', 'description']

7
netbox/ipam/api/serializers_/vlans.py
View File

@ -11,6 +11,7 @@ from netbox.api.serializers import NetBoxModelSerializer
from tenancy.api.serializers_.tenants import TenantSerializer
from utilities.api import get_serializer_for_model
from vpn.api.serializers_.l2vpn import L2VPNTerminationSerializer
from .nested import NestedVLANSerializer
from .roles import RoleSerializer
__all__ = (
@ -64,6 +65,8 @@ class VLANSerializer(NetBoxModelSerializer):
tenant = TenantSerializer(nested=True, required=False, allow_null=True)
status = ChoiceField(choices=VLANStatusChoices, required=False)
role = RoleSerializer(nested=True, required=False, allow_null=True)
qinq_role = ChoiceField(choices=VLANQinQRoleChoices, required=False)
qinq_svlan = NestedVLANSerializer(required=False, allow_null=True, default=None)
l2vpn_termination = L2VPNTerminationSerializer(nested=True, read_only=True, allow_null=True)
# Related object counts
@ -73,8 +76,8 @@ class VLANSerializer(NetBoxModelSerializer):
model = VLAN
fields = [
'id', 'url', 'display_url', 'display', 'site', 'group', 'vid', 'name', 'tenant', 'status', 'role',
'description', 'comments', 'l2vpn_termination', 'tags', 'custom_fields', 'created', 'last_updated',
'prefix_count',
'description', 'qinq_role', 'qinq_svlan', 'comments', 'l2vpn_termination', 'tags', 'custom_fields',
'created', 'last_updated', 'prefix_count',
]
brief_fields = ('id', 'url', 'display', 'vid', 'name', 'description')

11
netbox/ipam/choices.py
View File

@ -157,6 +157,17 @@ class VLANStatusChoices(ChoiceSet):
]
class VLANQinQRoleChoices(ChoiceSet):
ROLE_SERVICE = 's-vlan'
ROLE_CUSTOMER = 'c-vlan'
CHOICES = [
(ROLE_SERVICE, _('Service'), 'blue'),
(ROLE_CUSTOMER, _('Customer'), 'orange'),
]
#
# Services
#

11
netbox/ipam/filtersets.py
View File

@ -1041,6 +1041,17 @@ class VLANFilterSet(NetBoxModelFilterSet, TenancyFilterSet):
queryset=VirtualMachine.objects.all(),
method='get_for_virtualmachine'
)
qinq_role = django_filters.MultipleChoiceFilter(
choices=VLANQinQRoleChoices
)
qinq_svlan_id = django_filters.ModelMultipleChoiceFilter(
queryset=VLAN.objects.all(),
label=_('Q-in-Q SVLAN (ID)'),
)
qinq_svlan_vid = MultiValueNumberFilter(
field_name='qinq_svlan__vid',
label=_('Q-in-Q SVLAN number (1-4094)'),
)
l2vpn_id = django_filters.ModelMultipleChoiceFilter(
field_name='l2vpn_terminations__l2vpn',
queryset=L2VPN.objects.all(),

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

@ -527,15 +527,29 @@ class VLANBulkEditForm(NetBoxModelBulkEditForm):
max_length=200,
required=False
)
qinq_role = forms.ChoiceField(
label=_('Q-in-Q role'),
choices=add_blank_choice(VLANQinQRoleChoices),
required=False
)
qinq_svlan = DynamicModelChoiceField(
label=_('Q-in-Q SVLAN'),
queryset=VLAN.objects.all(),
required=False,
query_params={
'qinq_role': VLANQinQRoleChoices.ROLE_SERVICE,
}
)
comments = CommentField()
model = VLAN
fieldsets = (
FieldSet('status', 'role', 'tenant', 'description'),
FieldSet('qinq_role', 'qinq_svlan', name=_('Q-in-Q')),
FieldSet('region', 'site_group', 'site', 'group', name=_('Site & Group')),
)
nullable_fields = (
'site', 'group', 'tenant', 'role', 'description', 'comments',
'site', 'group', 'tenant', 'role', 'description', 'qinq_role', 'qinq_svlan', 'comments',
)

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

@ -461,10 +461,26 @@ class VLANImportForm(NetBoxModelImportForm):
to_field_name='name',
help_text=_('Functional role')
)
qinq_role = CSVChoiceField(
label=_('Q-in-Q role'),
choices=VLANStatusChoices,
required=False,
help_text=_('Operational status')
)
qinq_svlan = CSVModelChoiceField(
label=_('Q-in-Q SVLAN'),
queryset=VLAN.objects.all(),
required=False,
to_field_name='vid',
help_text=_("Service VLAN (for Q-in-Q/802.1ad customer VLANs)")
)
class Meta:
model = VLAN
fields = ('site', 'group', 'vid', 'name', 'tenant', 'status', 'role', 'description', 'comments', 'tags')
fields = (
'site', 'group', 'vid', 'name', 'tenant', 'status', 'role', 'description', 'qinq_role', 'qinq_svlan',
'comments', 'tags',
)
class VLANTranslationPolicyImportForm(NetBoxModelImportForm):

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

@ -506,6 +506,7 @@ class VLANFilterForm(TenancyFilterForm, NetBoxModelFilterSetForm):
FieldSet('q', 'filter_id', 'tag'),
FieldSet('region_id', 'site_group_id', 'site_id', name=_('Location')),
FieldSet('group_id', 'status', 'role_id', 'vid', 'l2vpn_id', name=_('Attributes')),
FieldSet('qinq_role', 'qinq_svlan_id', name=_('Q-in-Q/802.1ad')),
FieldSet('tenant_group_id', 'tenant_id', name=_('Tenant')),
)
selector_fields = ('filter_id', 'q', 'site_id')
@ -552,6 +553,17 @@ class VLANFilterForm(TenancyFilterForm, NetBoxModelFilterSetForm):
required=False,
label=_('VLAN ID')
)
qinq_role = forms.MultipleChoiceField(
label=_('Q-in-Q role'),
choices=VLANQinQRoleChoices,
required=False
)
qinq_svlan_id = DynamicModelMultipleChoiceField(
queryset=VLAN.objects.all(),
required=False,
null_option='None',
label=_('Q-in-Q SVLAN')
)
l2vpn_id = DynamicModelMultipleChoiceField(
queryset=L2VPN.objects.all(),
required=False,

12
netbox/ipam/forms/model_forms.py
View File

@ -683,13 +683,21 @@ class VLANForm(TenancyForm, NetBoxModelForm):
queryset=Role.objects.all(),
required=False
)
qinq_svlan = DynamicModelChoiceField(
label=_('Q-in-Q SVLAN'),
queryset=VLAN.objects.all(),
required=False,
query_params={
'qinq_role': VLANQinQRoleChoices.ROLE_SERVICE,
}
)
comments = CommentField()
class Meta:
model = VLAN
fields = [
'site', 'group', 'vid', 'name', 'status', 'role', 'tenant_group', 'tenant', 'description', 'comments',
'tags',
'site', 'group', 'vid', 'name', 'status', 'role', 'tenant_group', 'tenant', 'qinq_role', 'qinq_svlan',
'description', 'comments', 'tags',
]

6
netbox/ipam/graphql/types.py
View File

@ -236,7 +236,7 @@ class ServiceTemplateType(NetBoxObjectType):
@strawberry_django.type(
models.VLAN,
fields='__all__',
exclude=('qinq_svlan',),
filters=VLANFilter
)
class VLANType(NetBoxObjectType):
@ -252,6 +252,10 @@ class VLANType(NetBoxObjectType):
interfaces_as_tagged: List[Annotated["InterfaceType", strawberry.lazy('dcim.graphql.types')]]
vminterfaces_as_tagged: List[Annotated["VMInterfaceType", strawberry.lazy('virtualization.graphql.types')]]
@strawberry_django.field
def qinq_svlan(self) -> Annotated["VLANType", strawberry.lazy('ipam.graphql.types')] | None:
return self.qinq_svlan
@strawberry_django.type(
models.VLANGroup,

30
netbox/ipam/migrations/0075_vlan_qinq.py Normal file
View File

@ -0,0 +1,30 @@
import django.db.models.deletion
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('ipam', '0074_vlantranslationpolicy_vlantranslationrule'),
]
operations = [
migrations.AddField(
model_name='vlan',
name='qinq_role',
field=models.CharField(blank=True, max_length=50, null=True),
),
migrations.AddField(
model_name='vlan',
name='qinq_svlan',
field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.PROTECT, related_name='qinq_cvlans', to='ipam.vlan'),
),
migrations.AddConstraint(
model_name='vlan',
constraint=models.UniqueConstraint(fields=('qinq_svlan', 'vid'), name='ipam_vlan_unique_qinq_svlan_vid'),
),
migrations.AddConstraint(
model_name='vlan',
constraint=models.UniqueConstraint(fields=('qinq_svlan', 'name'), name='ipam_vlan_unique_qinq_svlan_name'),
),
]

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

@ -204,6 +204,21 @@ class VLAN(PrimaryModel):
null=True,
help_text=_("The primary function of this VLAN")
)
qinq_svlan = models.ForeignKey(
to='self',
on_delete=models.PROTECT,
related_name='qinq_cvlans',
blank=True,
null=True
)
qinq_role = models.CharField(
verbose_name=_('Q-in-Q role'),
max_length=50,
choices=VLANQinQRoleChoices,
blank=True,
null=True,
help_text=_("Customer/service VLAN designation (for Q-in-Q/IEEE 802.1ad)")
)
l2vpn_terminations = GenericRelation(
to='vpn.L2VPNTermination',
content_type_field='assigned_object_type',
@ -214,7 +229,7 @@ class VLAN(PrimaryModel):
objects = VLANQuerySet.as_manager()
clone_fields = [
'site', 'group', 'tenant', 'status', 'role', 'description',
'site', 'group', 'tenant', 'status', 'role', 'description', 'qinq_role', 'qinq_svlan',
]
class Meta:
@ -228,6 +243,14 @@ class VLAN(PrimaryModel):
fields=('group', 'name'),
name='%(app_label)s_%(class)s_unique_group_name'
),
models.UniqueConstraint(
fields=('qinq_svlan', 'vid'),
name='%(app_label)s_%(class)s_unique_qinq_svlan_vid'
),
models.UniqueConstraint(
fields=('qinq_svlan', 'name'),
name='%(app_label)s_%(class)s_unique_qinq_svlan_name'
),
)
verbose_name = _('VLAN')
verbose_name_plural = _('VLANs')
@ -255,9 +278,24 @@ class VLAN(PrimaryModel):
).format(ranges=ranges_to_string(self.group.vid_ranges), group=self.group)
})
# Only Q-in-Q customer VLANs may be assigned to a service VLAN
if self.qinq_svlan and self.qinq_role != VLANQinQRoleChoices.ROLE_CUSTOMER:
raise ValidationError({
'qinq_svlan': _("Only Q-in-Q customer VLANs maybe assigned to a service VLAN.")
})
# A Q-in-Q customer VLAN must be assigned to a service VLAN
if self.qinq_role == VLANQinQRoleChoices.ROLE_CUSTOMER and not self.qinq_svlan:
raise ValidationError({
'qinq_role': _("A Q-in-Q customer VLAN must be assigned to a service VLAN.")
})
def get_status_color(self):
return VLANStatusChoices.colors.get(self.status)
def get_qinq_role_color(self):
return VLANQinQRoleChoices.colors.get(self.qinq_role)
def get_interfaces(self):
# Return all device interfaces assigned to this VLAN
return Interface.objects.filter(

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

@ -132,6 +132,13 @@ class VLANTable(TenancyColumnsMixin, NetBoxTable):
verbose_name=_('Role'),
linkify=True
)
qinq_role = columns.ChoiceFieldColumn(
verbose_name=_('Q-in-Q role')
)
qinq_svlan = tables.Column(
verbose_name=_('Q-in-Q SVLAN'),
linkify=True
)
l2vpn = tables.Column(
accessor=tables.A('l2vpn_termination__l2vpn'),
linkify=True,
@ -154,7 +161,7 @@ class VLANTable(TenancyColumnsMixin, NetBoxTable):
model = VLAN
fields = (
'pk', 'id', 'vid', 'name', 'site', 'group', 'prefixes', 'tenant', 'tenant_group', 'status', 'role',
'description', 'comments', 'tags', 'l2vpn', 'created', 'last_updated',
'qinq_role', 'qinq_svlan', 'description', 'comments', 'tags', 'l2vpn', 'created', 'last_updated',
)
default_columns = ('pk', 'vid', 'name', 'site', 'group', 'prefixes', 'tenant', 'status', 'role', 'description')
row_attrs = {

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

@ -980,6 +980,7 @@ class VLANTest(APIViewTestCases.APIViewTestCase):
VLAN(name='VLAN 1', vid=1, group=vlan_groups[0]),
VLAN(name='VLAN 2', vid=2, group=vlan_groups[0]),
VLAN(name='VLAN 3', vid=3, group=vlan_groups[0]),
VLAN(name='SVLAN 1', vid=1001, qinq_role=VLANQinQRoleChoices.ROLE_SERVICE),
)
VLAN.objects.bulk_create(vlans)
@ -999,6 +1000,12 @@ class VLANTest(APIViewTestCases.APIViewTestCase):
'name': 'VLAN 6',
'group': vlan_groups[1].pk,
},
{
'vid': 2001,
'name': 'CVLAN 1',
'qinq_role': VLANQinQRoleChoices.ROLE_CUSTOMER,
'qinq_svlan': vlans[3].pk,
},
]
def test_delete_vlan_with_prefix(self):

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

@ -1630,6 +1630,7 @@ class VLANTestCase(TestCase, ChangeLoggedFilterSetTests):
Site(name='Site 4', slug='site-4', region=regions[0], group=site_groups[0]),
Site(name='Site 5', slug='site-5', region=regions[1], group=site_groups[1]),
Site(name='Site 6', slug='site-6', region=regions[2], group=site_groups[2]),
Site(name='Site 7', slug='site-7'),
)
Site.objects.bulk_create(sites)
@ -1784,9 +1785,21 @@ class VLANTestCase(TestCase, ChangeLoggedFilterSetTests):
# Create one globally available VLAN
VLAN(vid=1000, name='Global VLAN'),
# Create some Q-in-Q service VLANs
VLAN(vid=2001, name='SVLAN 1', site=sites[6], qinq_role=VLANQinQRoleChoices.ROLE_SERVICE),
VLAN(vid=2002, name='SVLAN 2', site=sites[6], qinq_role=VLANQinQRoleChoices.ROLE_SERVICE),
VLAN(vid=2003, name='SVLAN 3', site=sites[6], qinq_role=VLANQinQRoleChoices.ROLE_SERVICE),
)
VLAN.objects.bulk_create(vlans)
# Create Q-in-Q customer VLANs
VLAN.objects.bulk_create([
VLAN(vid=3001, name='CVLAN 1', site=sites[6], qinq_svlan=vlans[29], qinq_role=VLANQinQRoleChoices.ROLE_CUSTOMER),
VLAN(vid=3002, name='CVLAN 2', site=sites[6], qinq_svlan=vlans[30], qinq_role=VLANQinQRoleChoices.ROLE_CUSTOMER),
VLAN(vid=3003, name='CVLAN 3', site=sites[6], qinq_svlan=vlans[31], qinq_role=VLANQinQRoleChoices.ROLE_CUSTOMER),
])
# Assign VLANs to device interfaces
interfaces[0].untagged_vlan = vlans[0]
interfaces[0].tagged_vlans.add(vlans[1])
@ -1897,6 +1910,17 @@ class VLANTestCase(TestCase, ChangeLoggedFilterSetTests):
params = {'vminterface_id': vminterface_id}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1)
def test_qinq_role(self):
params = {'qinq_role': [VLANQinQRoleChoices.ROLE_SERVICE, VLANQinQRoleChoices.ROLE_CUSTOMER]}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 6)
def test_qinq_svlan(self):
vlans = VLAN.objects.filter(qinq_role=VLANQinQRoleChoices.ROLE_SERVICE)[:2]
params = {'qinq_svlan_id': [vlans[0].pk, vlans[1].pk]}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
params = {'qinq_svlan_vid': [vlans[0].vid, vlans[1].vid]}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
class VLANTranslationPolicyTestCase(TestCase, ChangeLoggedFilterSetTests):
queryset = VLANTranslationPolicy.objects.all()

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

@ -586,3 +586,24 @@ class TestVLANGroup(TestCase):
vlangroup.vid_ranges = string_to_ranges('2-2')
vlangroup.full_clean()
vlangroup.save()
class TestVLAN(TestCase):
@classmethod
def setUpTestData(cls):
VLAN.objects.bulk_create((
VLAN(name='VLAN 1', vid=1, qinq_role=VLANQinQRoleChoices.ROLE_SERVICE),
))
def test_qinq_role(self):
svlan = VLAN.objects.filter(qinq_role=VLANQinQRoleChoices.ROLE_SERVICE).first()
vlan = VLAN(
name='VLAN X',
vid=999,
qinq_role=VLANQinQRoleChoices.ROLE_SERVICE,
qinq_svlan=svlan
)
with self.assertRaises(ValidationError):
vlan.full_clean()

25
netbox/templates/circuits/inc/circuit_termination_fields.html
View File

@ -1,18 +1,19 @@
{% load helpers %}
{% load i18n %}
{% if termination.site %}
<tr>
<th scope="row">{% trans "Site" %}</th>
<td>
{% if termination.site.region %}
{{ termination.site.region|linkify }} /
{% endif %}
{{ termination.site|linkify }}
</td>
<th scope="row">{% trans "Termination point" %}</th>
{% if termination.termination %}
<td>
{{ termination.termination|linkify }}
<div class="fs-5 text-muted">{% trans termination.termination_type.name|bettertitle %}</div>
</td>
{% else %}
<td>{{ ''|placeholder }}</td>
{% endif %}
</tr>
<tr>
<th scope="row">{% trans "Termination" %}</th>
<th scope="row">{% trans "Connection" %}</th>
<td>
{% if termination.mark_connected %}
<span class="text-success"><i class="mdi mdi-check-bold"></i></span>
@ -57,12 +58,6 @@
{% endif %}
</td>
</tr>
{% else %}
<tr>
<th scope="row">{% trans "Provider Network" %}</th>
<td>{{ termination.provider_network.provider|linkify }} / {{ termination.provider_network|linkify }}</td>
</tr>
{% endif %}
<tr>
<th scope="row">{% trans "Speed" %}</th>
<td>

31
netbox/templates/ipam/vlan.html
View File

@ -62,6 +62,22 @@
<th scope="row">{% trans "Description" %}</th>
<td>{{ object.description|placeholder }}</td>
</tr>
<tr>
<th scope="row">{% trans "Q-in-Q Role" %}</th>
<td>
{% if object.qinq_role %}
{% badge object.get_qinq_role_display bg_color=object.get_qinq_role_color %}
{% else %}
{{ ''|placeholder }}
{% endif %}
</td>
</tr>
{% if object.qinq_role == 'c-vlan' %}
<tr>
<th scope="row">{% trans "Q-in-Q SVLAN" %}</th>
<td>{{ object.qinq_svlan|linkify|placeholder }}</td>
</tr>
{% endif %}
<tr>
<th scope="row">{% trans "L2VPN" %}</th>
<td>{{ object.l2vpn_termination.l2vpn|linkify|placeholder }}</td>
@ -92,6 +108,21 @@
</h2>
{% htmx_table 'ipam:prefix_list' vlan_id=object.pk %}
</div>
{% if object.qinq_role == 's-vlan' %}
<div class="card">
<h2 class="card-header">
{% trans "Customer VLANs" %}
{% if perms.ipam.add_vlan %}
<div class="card-actions">
<a href="{% url 'ipam:vlan_add' %}?qinq_role=c-vlan&qinq_svlan={{ object.pk }}" class="btn btn-ghost-primary btn-sm">
<i class="mdi mdi-plus-thick" aria-hidden="true"></i> {% trans "Add a VLAN" %}
</a>
</div>
{% endif %}
</h2>
{% htmx_table 'ipam:vlan_list' qinq_svlan_id=object.pk %}
</div>
{% endif %}
{% plugin_full_width_page object %}
</div>
</div>

8
netbox/templates/ipam/vlan_edit.html
View File

@ -17,6 +17,14 @@
{% render_field form.tags %}
</div>
<div class="field-group my-5">
<div class="row">
<h2 class="col-9 offset-3">{% trans "Q-in-Q (802.1ad)" %}</h2>
</div>
{% render_field form.qinq_role %}
{% render_field form.qinq_svlan %}
</div>
<div class="field-group my-5">
<div class="row">
<h2 class="col-9 offset-3">{% trans "Tenancy" %}</h2>

7
netbox/virtualization/api/serializers_/virtualmachines.py
View File

@ -89,6 +89,7 @@ class VMInterfaceSerializer(NetBoxModelSerializer):
required=False,
many=True
)
qinq_svlan = VLANSerializer(nested=True, required=False, allow_null=True)
vlan_translation_policy = VLANTranslationPolicySerializer(nested=True, required=False, allow_null=True)
vrf = VRFSerializer(nested=True, required=False, allow_null=True)
l2vpn_termination = L2VPNTerminationSerializer(nested=True, read_only=True, allow_null=True)
@ -105,9 +106,9 @@ class VMInterfaceSerializer(NetBoxModelSerializer):
model = VMInterface
fields = [
'id', 'url', 'display_url', 'display', 'virtual_machine', 'name', 'enabled', 'parent', 'bridge', 'mtu',
'mac_address', 'description', 'mode', 'untagged_vlan', 'tagged_vlans', 'vrf', 'l2vpn_termination',
'tags', 'custom_fields', 'created', 'last_updated', 'count_ipaddresses', 'count_fhrp_groups',
'vlan_translation_policy',
'mac_address', 'description', 'mode', 'untagged_vlan', 'tagged_vlans', 'qinq_svlan',
'vlan_translation_policy', 'vrf', 'l2vpn_termination', 'tags', 'custom_fields', 'created', 'last_updated',
'count_ipaddresses', 'count_fhrp_groups',
]
brief_fields = ('id', 'url', 'display', 'virtual_machine', 'name', 'description')

20
netbox/virtualization/forms/model_forms.py
View File

@ -6,6 +6,7 @@ from django.utils.translation import gettext_lazy as _
from dcim.forms.common import InterfaceCommonForm
from dcim.models import Device, DeviceRole, MACAddress, Platform, Rack, Region, Site, SiteGroup
from extras.models import ConfigTemplate
from ipam.choices import VLANQinQRoleChoices
from ipam.models import IPAddress, VLAN, VLANGroup, VLANTranslationPolicy, VRF
from netbox.forms import NetBoxModelForm
from tenancy.forms import TenancyForm
@ -338,6 +339,16 @@ class VMInterfaceForm(InterfaceCommonForm, VMComponentForm):
'available_on_virtualmachine': '$virtual_machine',
}
)
qinq_svlan = DynamicModelChoiceField(
queryset=VLAN.objects.all(),
required=False,
label=_('Q-in-Q Service VLAN'),
query_params={
'group_id': '$vlan_group',
'available_on_virtualmachine': '$virtual_machine',
'qinq_role': VLANQinQRoleChoices.ROLE_SERVICE,
}
)
vrf = DynamicModelChoiceField(
queryset=VRF.objects.all(),
required=False,
@ -354,17 +365,20 @@ class VMInterfaceForm(InterfaceCommonForm, VMComponentForm):
FieldSet('vrf', 'mac_address', name=_('Addressing')),
FieldSet('mtu', 'enabled', name=_('Operation')),
FieldSet('parent', 'bridge', name=_('Related Interfaces')),
FieldSet('mode', 'vlan_group', 'untagged_vlan', 'tagged_vlans', 'vlan_translation_policy', name=_('802.1Q Switching')),
FieldSet(
'mode', 'vlan_group', 'untagged_vlan', 'tagged_vlans', 'qinq_svlan', 'vlan_translation_policy',
name=_('802.1Q Switching')
),
)
class Meta:
model = VMInterface
fields = [
'virtual_machine', 'name', 'parent', 'bridge', 'enabled', 'mac_address', 'mtu', 'description', 'mode',
'vlan_group', 'untagged_vlan', 'tagged_vlans', 'vrf', 'tags', 'vlan_translation_policy',
'vlan_group', 'untagged_vlan', 'tagged_vlans', 'qinq_svlan', 'vlan_translation_policy', 'vrf', 'tags',
]
labels = {
'mode': '802.1Q Mode',
'mode': _('802.1Q Mode'),
}
widgets = {
'mode': HTMXSelect(),

1
netbox/virtualization/graphql/types.py
View File

@ -99,6 +99,7 @@ class VMInterfaceType(IPAddressesMixin, ComponentType):
bridge: Annotated["VMInterfaceType", strawberry.lazy('virtualization.graphql.types')] | None
untagged_vlan: Annotated["VLANType", strawberry.lazy('ipam.graphql.types')] | None
vrf: Annotated["VRFType", strawberry.lazy('ipam.graphql.types')] | None
qinq_svlan: Annotated["VLANType", strawberry.lazy('ipam.graphql.types')] | None
vlan_translation_policy: Annotated["VLANTranslationPolicyType", strawberry.lazy('ipam.graphql.types')] | None
tagged_vlans: List[Annotated["VLANType", strawberry.lazy('ipam.graphql.types')]]

2
netbox/virtualization/migrations/0042_vminterface_vlan_translation_policy.py
View File

@ -1,5 +1,3 @@
# Generated by Django 5.0.9 on 2024-10-11 19:45
import django.db.models.deletion
from django.db import migrations, models

28
netbox/virtualization/migrations/0043_qinq_svlan.py Normal file
View File

@ -0,0 +1,28 @@
import django.db.models.deletion
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('ipam', '0075_vlan_qinq'),
('virtualization', '0042_vminterface_vlan_translation_policy'),
]
operations = [
migrations.AddField(
model_name='vminterface',
name='qinq_svlan',
field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='%(class)ss_svlan', to='ipam.vlan'),
),
migrations.AlterField(
model_name='vminterface',
name='tagged_vlans',
field=models.ManyToManyField(blank=True, related_name='%(class)ss_as_tagged', to='ipam.vlan'),
),
migrations.AlterField(
model_name='vminterface',
name='untagged_vlan',
field=models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='%(class)ss_as_untagged', to='ipam.vlan'),
),
]

4
netbox/virtualization/migrations/0043_rename_mac_address_vminterface__mac_address.py → netbox/virtualization/migrations/0044_rename_mac_address_vminterface__mac_address.py
View File

@ -1,12 +1,10 @@
# Generated by Django 5.1.2 on 2024-10-29 15:05
from django.db import migrations
class Migration(migrations.Migration):
dependencies = [
('virtualization', '0042_vminterface_vlan_translation_policy'),
('virtualization', '0043_qinq_svlan'),
]
operations = [

4
netbox/virtualization/migrations/0044_remove_vminterface__mac_address.py → netbox/virtualization/migrations/0045_remove_vminterface__mac_address.py
View File

@ -1,12 +1,10 @@
# Generated by Django 5.1.2 on 2024-10-29 15:41
from django.db import migrations
class Migration(migrations.Migration):
dependencies = [
('virtualization', '0043_rename_mac_address_vminterface__mac_address'),
('virtualization', '0044_rename_mac_address_vminterface__mac_address'),
]
operations = [

14
netbox/virtualization/models/virtualmachines.py
View File

@ -322,20 +322,6 @@ class VMInterface(ComponentModel, BaseInterface, TrackingModelMixin):
max_length=100,
blank=True
)
untagged_vlan = models.ForeignKey(
to='ipam.VLAN',
on_delete=models.SET_NULL,
related_name='vminterfaces_as_untagged',
null=True,
blank=True,
verbose_name=_('untagged VLAN')
)
tagged_vlans = models.ManyToManyField(
to='ipam.VLAN',
related_name='vminterfaces_as_tagged',
blank=True,
verbose_name=_('tagged VLANs')
)
ip_addresses = GenericRelation(
to='ipam.IPAddress',
content_type_field='assigned_object_type',

7
netbox/virtualization/tables/virtualmachines.py
View File

@ -151,8 +151,8 @@ class VMInterfaceTable(BaseInterfaceTable):
model = VMInterface
fields = (
'pk', 'id', 'name', 'virtual_machine', 'enabled', 'mac_address', 'mtu', 'mode', 'description', 'tags',
'vrf', 'l2vpn', 'tunnel', 'ip_addresses', 'fhrp_groups', 'untagged_vlan', 'tagged_vlans', 'created',
'last_updated',
'vrf', 'l2vpn', 'tunnel', 'ip_addresses', 'fhrp_groups', 'untagged_vlan', 'tagged_vlans', 'qinq_svlan',
'created', 'last_updated',
)
default_columns = ('pk', 'name', 'virtual_machine', 'enabled', 'description')
@ -175,7 +175,8 @@ class VirtualMachineVMInterfaceTable(VMInterfaceTable):
model = VMInterface
fields = (
'pk', 'id', 'name', 'enabled', 'parent', 'bridge', 'mac_address', 'mtu', 'mode', 'description', 'tags',
'vrf', 'l2vpn', 'tunnel', 'ip_addresses', 'fhrp_groups', 'untagged_vlan', 'tagged_vlans', 'actions',
'vrf', 'l2vpn', 'tunnel', 'ip_addresses', 'fhrp_groups', 'untagged_vlan', 'tagged_vlans', 'qinq_svlan',
'actions',
)
default_columns = ('pk', 'name', 'enabled', 'mac_address', 'mtu', 'mode', 'description', 'ip_addresses')
row_attrs = {

8
netbox/virtualization/tests/test_api.py
View File

@ -4,6 +4,7 @@ from rest_framework import status
from dcim.choices import InterfaceModeChoices
from dcim.models import Site
from extras.models import ConfigTemplate
from ipam.choices import VLANQinQRoleChoices
from ipam.models import VLAN, VRF
from utilities.testing import APITestCase, APIViewTestCases, create_test_device, create_test_virtualmachine
from virtualization.choices import *
@ -270,6 +271,7 @@ class VMInterfaceTest(APIViewTestCases.APIViewTestCase):
VLAN(name='VLAN 1', vid=1),
VLAN(name='VLAN 2', vid=2),
VLAN(name='VLAN 3', vid=3),
VLAN(name='SVLAN 1', vid=1001, qinq_role=VLANQinQRoleChoices.ROLE_SERVICE),
)
VLAN.objects.bulk_create(vlans)
@ -307,6 +309,12 @@ class VMInterfaceTest(APIViewTestCases.APIViewTestCase):
'untagged_vlan': vlans[2].pk,
'vrf': vrfs[2].pk,
},
{
'virtual_machine': virtualmachine.pk,
'name': 'Interface 7',
'mode': InterfaceModeChoices.MODE_Q_IN_Q,
'qinq_svlan': vlans[3].pk,
},
]
def test_bulk_delete_child_interfaces(self):

24
netbox/virtualization/tests/test_filtersets.py
View File

@ -1,7 +1,9 @@
from django.test import TestCase
from dcim.choices import InterfaceModeChoices
from dcim.models import Device, DeviceRole, MACAddress, Platform, Region, Site, SiteGroup
from ipam.models import IPAddress, VLANTranslationPolicy, VRF
from ipam.choices import VLANQinQRoleChoices
from ipam.models import IPAddress, VLAN, VLANTranslationPolicy, VRF
from tenancy.models import Tenant, TenantGroup
from utilities.testing import ChangeLoggedFilterSetTests, create_test_device
from virtualization.choices import *
@ -539,7 +541,7 @@ class VirtualMachineTestCase(TestCase, ChangeLoggedFilterSetTests):
class VMInterfaceTestCase(TestCase, ChangeLoggedFilterSetTests):
queryset = VMInterface.objects.all()
filterset = VMInterfaceFilterSet
ignore_fields = ('tagged_vlans', 'untagged_vlan',)
ignore_fields = ('tagged_vlans', 'untagged_vlan', 'qinq_svlan')
@classmethod
def setUpTestData(cls):
@ -565,6 +567,13 @@ class VMInterfaceTestCase(TestCase, ChangeLoggedFilterSetTests):
)
VRF.objects.bulk_create(vrfs)
vlans = (
VLAN(name='SVLAN 1', vid=1001, qinq_role=VLANQinQRoleChoices.ROLE_SERVICE),
VLAN(name='SVLAN 2', vid=1002, qinq_role=VLANQinQRoleChoices.ROLE_SERVICE),
VLAN(name='SVLAN 3', vid=1003, qinq_role=VLANQinQRoleChoices.ROLE_SERVICE),
)
VLAN.objects.bulk_create(vlans)
vms = (
VirtualMachine(name='Virtual Machine 1', cluster=clusters[0]),
VirtualMachine(name='Virtual Machine 2', cluster=clusters[1]),
@ -611,7 +620,9 @@ class VMInterfaceTestCase(TestCase, ChangeLoggedFilterSetTests):
enabled=False,
mtu=300,
vrf=vrfs[2],
description='foobar3'
description='foobar3',
mode=InterfaceModeChoices.MODE_Q_IN_Q,
qinq_svlan=vlans[0]
),
)
VMInterface.objects.bulk_create(interfaces)
@ -686,6 +697,13 @@ class VMInterfaceTestCase(TestCase, ChangeLoggedFilterSetTests):
params = {'description': ['foobar1', 'foobar2']}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
def test_vlan(self):
vlan = VLAN.objects.filter(qinq_role=VLANQinQRoleChoices.ROLE_SERVICE).first()
params = {'vlan_id': vlan.pk}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1)
params = {'vlan': vlan.vid}
self.assertEqual(self.filterset(params, self.queryset).qs.count(), 1)
def test_vlan_translation_policy(self):
vlan_translation_policies = VLANTranslationPolicy.objects.all()[:2]
params = {'vlan_translation_policy_id': [vlan_translation_policies[0].pk, vlan_translation_policies[1].pk]}