Coverage for ipam/models/vlans.py: 74%
155 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 18:35 +0000
1from django.contrib.contenttypes.fields import GenericForeignKey, GenericRelation
2from django.contrib.contenttypes.models import ContentType
3from django.contrib.postgres.fields import ArrayField, IntegerRangeField
4from django.core.exceptions import ValidationError
5from django.core.validators import MaxValueValidator, MinValueValidator
6from django.db import models
7from django.db.backends.postgresql.psycopg_any import NumericRange
8from django.utils.translation import gettext_lazy as _
10from dcim.models import Interface, Site, SiteGroup
11from ipam.choices import *
12from ipam.constants import *
13from ipam.querysets import VLANGroupQuerySet, VLANQuerySet
14from netbox.models import NetBoxModel, OrganizationalModel, PrimaryModel
15from utilities.data import (
16 check_ranges_overlap,
17 get_inclusive_integer_range_bounds,
18 normalize_integer_range,
19 normalize_update_fields,
20 ranges_to_string,
21 ranges_to_string_list,
22)
23from virtualization.models import VMInterface
25__all__ = (
26 'VLAN',
27 'VLANGroup',
28 'VLANTranslationPolicy',
29 'VLANTranslationRule',
30)
33def default_vid_ranges():
34 return [NumericRange(VLAN_VID_MIN, VLAN_VID_MAX + 1)]
37class VLANGroup(OrganizationalModel):
38 """
39 A VLAN group is an arbitrary collection of VLANs within which VLAN IDs and names must be unique. Each group must
40 define one or more ranges of valid VLAN IDs, and may be assigned a specific scope.
41 """
42 name = models.CharField(
43 verbose_name=_('name'),
44 max_length=100,
45 db_collation="natural_sort"
46 )
47 slug = models.SlugField(
48 verbose_name=_('slug'),
49 max_length=100
50 )
51 scope_type = models.ForeignKey(
52 to='contenttypes.ContentType',
53 on_delete=models.CASCADE,
54 blank=True,
55 null=True
56 )
57 scope_id = models.PositiveBigIntegerField(
58 blank=True,
59 null=True
60 )
61 scope = GenericForeignKey(
62 ct_field='scope_type',
63 fk_field='scope_id'
64 )
65 # Normalized to canonical '[)' bounds on save() to match PostgreSQL storage.
66 vid_ranges = ArrayField(
67 IntegerRangeField(),
68 verbose_name=_('VLAN ID ranges'),
69 default=default_vid_ranges
70 )
71 total_vlan_ids = models.PositiveBigIntegerField(
72 default=VLAN_VID_MAX - VLAN_VID_MIN + 1,
73 )
74 tenant = models.ForeignKey(
75 to='tenancy.Tenant',
76 on_delete=models.PROTECT,
77 related_name='vlan_groups',
78 blank=True,
79 null=True
80 )
82 objects = VLANGroupQuerySet.as_manager()
84 class Meta:
85 ordering = ('name', 'pk') # Name may be non-unique
86 indexes = (
87 models.Index(fields=('name', 'id')), # Default ordering
88 models.Index(fields=('scope_type', 'scope_id')),
89 )
90 constraints = (
91 models.UniqueConstraint(
92 fields=('scope_type', 'scope_id', 'name'),
93 name='%(app_label)s_%(class)s_unique_scope_name'
94 ),
95 models.UniqueConstraint(
96 fields=('scope_type', 'scope_id', 'slug'),
97 name='%(app_label)s_%(class)s_unique_scope_slug'
98 ),
99 )
100 verbose_name = _('VLAN group')
101 verbose_name_plural = _('VLAN groups')
103 def clean(self):
104 # Validate the scope pair first, since BaseModel.clean() keys its errors to scope_id, which forms omit
105 if self.scope_type and not self.scope_id: 105 ↛ 106line 105 didn't jump to line 106 because the condition on line 105 was never true
106 scope_type = self.scope_type.model_class()
107 raise ValidationError(
108 _("Please select a {scope_type}.").format(scope_type=scope_type._meta.model_name)
109 )
110 if self.scope_id and not self.scope_type: 110 ↛ 111line 110 didn't jump to line 111 because the condition on line 110 was never true
111 raise ValidationError({'scope_type': _("Please select a scope type.")})
113 super().clean()
115 # Validate VID ranges
116 for vid_range in self.vid_ranges:
117 lower_vid, upper_vid = get_inclusive_integer_range_bounds(vid_range)
118 if lower_vid < VLAN_VID_MIN: 118 ↛ 119line 118 didn't jump to line 119 because the condition on line 118 was never true
119 raise ValidationError({
120 'vid_ranges': _("Starting VLAN ID in range ({value}) cannot be less than {minimum}").format(
121 value=lower_vid, minimum=VLAN_VID_MIN
122 )
123 })
124 if upper_vid > VLAN_VID_MAX: 124 ↛ 125line 124 didn't jump to line 125 because the condition on line 124 was never true
125 raise ValidationError({
126 'vid_ranges': _("Ending VLAN ID in range ({value}) cannot exceed {maximum}").format(
127 value=upper_vid, maximum=VLAN_VID_MAX
128 )
129 })
130 if lower_vid > upper_vid: 130 ↛ 131line 130 didn't jump to line 131 because the condition on line 130 was never true
131 raise ValidationError({
132 'vid_ranges': _(
133 "Ending VLAN ID in range must be greater than or equal to the starting VLAN ID ({range})"
134 ).format(range=f'{lower_vid}-{upper_vid}')
135 })
137 # Check for overlapping VID ranges
138 if self.vid_ranges and check_ranges_overlap(self.vid_ranges): 138 ↛ 139line 138 didn't jump to line 139 because the condition on line 138 was never true
139 raise ValidationError({'vid_ranges': _("Ranges cannot overlap.")})
141 def save(self, *args, **kwargs):
142 # Canonicalize vid_ranges on the instance so callers (e.g. Custom Scripts)
143 # see the same value in memory that PostgreSQL stores.
144 self.total_vlan_ids = 0
145 vid_ranges = []
146 for vid_range in self.vid_ranges:
147 vid_range = normalize_integer_range(vid_range)
148 vid_ranges.append(vid_range)
149 self.total_vlan_ids += vid_range.upper - vid_range.lower
150 self.vid_ranges = vid_ranges
152 update_fields = normalize_update_fields(kwargs)
153 if update_fields is not None and 'vid_ranges' in update_fields: 153 ↛ 155line 153 didn't jump to line 155 because the condition on line 153 was never true
154 # total_vlan_ids is a denormalized cache of vid_ranges; persist them together.
155 kwargs['update_fields'] = update_fields | {'total_vlan_ids'}
157 super().save(*args, **kwargs)
159 def get_available_vids(self):
160 """
161 Return all available VLANs within this group.
162 """
163 available_vlans = set()
164 for vlan_range in self.vid_ranges:
165 lower_vid, upper_vid = get_inclusive_integer_range_bounds(vlan_range)
166 available_vlans.update(range(lower_vid, upper_vid + 1))
167 available_vlans -= set(VLAN.objects.filter(group=self).values_list('vid', flat=True))
169 return sorted(available_vlans)
171 def get_next_available_vid(self):
172 """
173 Return the first available VLAN ID (1-4094) in the group.
174 """
175 available_vids = self.get_available_vids()
176 if available_vids:
177 return available_vids[0]
178 return None
180 def get_child_vlans(self):
181 """
182 Return all VLANs within this group.
183 """
184 return VLAN.objects.filter(group=self).order_by('vid')
186 @property
187 def vid_ranges_items(self):
188 """
189 Property that converts VID ranges to a list of string representations.
190 """
191 return ranges_to_string_list(self.vid_ranges)
193 @property
194 def vid_ranges_list(self):
195 """
196 Property that converts VID ranges into a string representation.
197 """
198 return ranges_to_string(self.vid_ranges)
201class VLAN(PrimaryModel):
202 """
203 A VLAN is a distinct layer two forwarding domain identified by a 12-bit integer (1-4094). Each VLAN must be assigned
204 to a Site, however VLAN IDs need not be unique within a Site. A VLAN may optionally be assigned to a VLANGroup,
205 within which all VLAN IDs and names but be unique.
207 Like Prefixes, each VLAN is assigned an operational status and optionally a user-defined Role. A VLAN can have zero
208 or more Prefixes assigned to it.
209 """
210 site = models.ForeignKey(
211 to='dcim.Site',
212 on_delete=models.PROTECT,
213 related_name='vlans',
214 blank=True,
215 null=True,
216 help_text=_("The specific site to which this VLAN is assigned (if any)")
217 )
218 group = models.ForeignKey(
219 to='ipam.VLANGroup',
220 on_delete=models.PROTECT,
221 related_name='vlans',
222 blank=True,
223 null=True,
224 help_text=_("VLAN group (optional)")
225 )
226 vid = models.PositiveSmallIntegerField(
227 verbose_name=_('VLAN ID'),
228 validators=(
229 MinValueValidator(VLAN_VID_MIN),
230 MaxValueValidator(VLAN_VID_MAX)
231 ),
232 help_text=_("Numeric VLAN ID (1-4094)")
233 )
234 name = models.CharField(
235 verbose_name=_('name'),
236 max_length=64
237 )
238 tenant = models.ForeignKey(
239 to='tenancy.Tenant',
240 on_delete=models.PROTECT,
241 related_name='vlans',
242 blank=True,
243 null=True
244 )
245 status = models.CharField(
246 verbose_name=_('status'),
247 max_length=50,
248 choices=VLANStatusChoices,
249 default=VLANStatusChoices.STATUS_ACTIVE,
250 help_text=_("Operational status of this VLAN")
251 )
252 role = models.ForeignKey(
253 to='ipam.Role',
254 on_delete=models.SET_NULL,
255 related_name='vlans',
256 blank=True,
257 null=True,
258 help_text=_("The primary function of this VLAN")
259 )
260 qinq_svlan = models.ForeignKey(
261 to='self',
262 on_delete=models.PROTECT,
263 related_name='qinq_cvlans',
264 blank=True,
265 null=True
266 )
267 qinq_role = models.CharField(
268 verbose_name=_('Q-in-Q role'),
269 max_length=50,
270 choices=VLANQinQRoleChoices,
271 blank=True,
272 null=True,
273 help_text=_("Customer/service VLAN designation (for Q-in-Q/IEEE 802.1ad)")
274 )
275 l2vpn_terminations = GenericRelation(
276 to='vpn.L2VPNTermination',
277 content_type_field='assigned_object_type',
278 object_id_field='assigned_object_id',
279 related_query_name='vlan'
280 )
282 objects = VLANQuerySet.as_manager()
284 clone_fields = [
285 'site', 'group', 'tenant', 'status', 'role', 'description', 'qinq_role', 'qinq_svlan',
286 ]
288 class Meta:
289 ordering = ('site', 'group', 'vid', 'pk') # (site, group, vid) may be non-unique
290 indexes = (
291 models.Index(fields=('site', 'group', 'vid', 'id')), # Default ordering
292 )
293 constraints = (
294 models.UniqueConstraint(
295 fields=('group', 'vid'),
296 name='%(app_label)s_%(class)s_unique_group_vid'
297 ),
298 models.UniqueConstraint(
299 fields=('group', 'name'),
300 name='%(app_label)s_%(class)s_unique_group_name'
301 ),
302 models.UniqueConstraint(
303 fields=('qinq_svlan', 'vid'),
304 name='%(app_label)s_%(class)s_unique_qinq_svlan_vid'
305 ),
306 models.UniqueConstraint(
307 fields=('qinq_svlan', 'name'),
308 name='%(app_label)s_%(class)s_unique_qinq_svlan_name'
309 ),
310 )
311 verbose_name = _('VLAN')
312 verbose_name_plural = _('VLANs')
314 def __str__(self):
315 return f'{self.name} ({self.vid})'
317 def clean(self):
318 super().clean()
320 # Validate VLAN group (if assigned)
321 if self.group and self.site and self.group.scope_type == ContentType.objects.get_for_model(Site): 321 ↛ 322line 321 didn't jump to line 322 because the condition on line 321 was never true
322 if self.site != self.group.scope:
323 raise ValidationError(
324 _(
325 "VLAN is assigned to group {group} (scope: {scope}); cannot also assign to site {site}."
326 ).format(group=self.group, scope=self.group.scope, site=self.site)
327 )
328 if self.group and self.site and self.group.scope_type == ContentType.objects.get_for_model(SiteGroup): 328 ↛ 329line 328 didn't jump to line 329 because the condition on line 328 was never true
329 if self.site not in self.group.scope.sites.all():
330 raise ValidationError(
331 _(
332 "The assigned site {site} is not a member of the assigned group {group} (scope: {scope})."
333 ).format(group=self.group, scope=self.group.scope, site=self.site)
334 )
336 # Check that the VLAN ID is permitted in the assigned group (if any)
337 if self.group and self.vid is not None: 337 ↛ 338line 337 didn't jump to line 338 because the condition on line 337 was never true
338 if not any([self.vid in r for r in self.group.vid_ranges]):
339 raise ValidationError({
340 'vid': _(
341 "VID must be in ranges {ranges} for VLANs in group {group}"
342 ).format(ranges=ranges_to_string(self.group.vid_ranges), group=self.group)
343 })
345 # Only Q-in-Q customer VLANs may be assigned to a service VLAN
346 if self.qinq_svlan and self.qinq_role != VLANQinQRoleChoices.ROLE_CUSTOMER: 346 ↛ 347line 346 didn't jump to line 347 because the condition on line 346 was never true
347 raise ValidationError({
348 'qinq_svlan': _("Only Q-in-Q customer VLANs maybe assigned to a service VLAN.")
349 })
351 # A Q-in-Q customer VLAN must be assigned to a service VLAN
352 if self.qinq_role == VLANQinQRoleChoices.ROLE_CUSTOMER and not self.qinq_svlan: 352 ↛ 353line 352 didn't jump to line 353 because the condition on line 352 was never true
353 raise ValidationError({
354 'qinq_role': _("A Q-in-Q customer VLAN must be assigned to a service VLAN.")
355 })
357 def get_status_color(self):
358 return VLANStatusChoices.colors.get(self.status)
360 def get_qinq_role_color(self):
361 return VLANQinQRoleChoices.colors.get(self.qinq_role)
363 def get_interfaces(self):
364 # Return all device interfaces assigned to this VLAN
365 return Interface.objects.filter(
366 Q(untagged_vlan_id=self.pk) |
367 Q(tagged_vlans=self.pk)
368 ).distinct()
370 def get_vminterfaces(self):
371 # Return all VM interfaces assigned to this VLAN
372 return VMInterface.objects.filter(
373 Q(untagged_vlan_id=self.pk) |
374 Q(tagged_vlans=self.pk)
375 ).distinct()
377 @property
378 def l2vpn_termination(self):
379 return self.l2vpn_terminations.first()
382class VLANTranslationPolicy(PrimaryModel):
383 name = models.CharField(
384 verbose_name=_('name'),
385 max_length=100,
386 unique=True,
387 )
389 class Meta:
390 verbose_name = _('VLAN translation policy')
391 verbose_name_plural = _('VLAN translation policies')
392 ordering = ('name',)
394 def __str__(self):
395 return self.name
398class VLANTranslationRule(NetBoxModel):
399 policy = models.ForeignKey(
400 to=VLANTranslationPolicy,
401 related_name='rules',
402 on_delete=models.CASCADE,
403 )
404 description = models.CharField(
405 verbose_name=_('description'),
406 max_length=200,
407 blank=True
408 )
409 local_vid = models.PositiveSmallIntegerField(
410 verbose_name=_('Local VLAN ID'),
411 validators=(
412 MinValueValidator(VLAN_VID_MIN),
413 MaxValueValidator(VLAN_VID_MAX)
414 ),
415 help_text=_("Numeric VLAN ID (1-4094)")
416 )
417 remote_vid = models.PositiveSmallIntegerField(
418 verbose_name=_('Remote VLAN ID'),
419 validators=(
420 MinValueValidator(VLAN_VID_MIN),
421 MaxValueValidator(VLAN_VID_MAX)
422 ),
423 help_text=_("Numeric VLAN ID (1-4094)")
424 )
425 prerequisite_models = (
426 'ipam.VLANTranslationPolicy',
427 )
429 clone_fields = ['policy']
431 class Meta:
432 verbose_name = _('VLAN translation rule')
433 ordering = ('policy', 'local_vid',)
434 constraints = (
435 models.UniqueConstraint(
436 fields=('policy', 'local_vid'),
437 name='%(app_label)s_%(class)s_unique_policy_local_vid'
438 ),
439 models.UniqueConstraint(
440 fields=('policy', 'remote_vid'),
441 name='%(app_label)s_%(class)s_unique_policy_remote_vid'
442 ),
443 )
445 def __str__(self):
446 return f'{self.local_vid} -> {self.remote_vid} ({self.policy})'
448 def to_objectchange(self, action):
449 objectchange = super().to_objectchange(action)
450 objectchange.related_object = self.policy
451 return objectchange