Coverage for circuits/models/circuits.py: 53%
164 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 18:35 +0000
1from django.apps import apps
2from django.contrib.contenttypes.fields import GenericForeignKey, GenericRelation
3from django.core.exceptions import ValidationError
4from django.db import models
5from django.urls import reverse
6from django.utils.translation import gettext_lazy as _
8from circuits.choices import *
9from dcim.models import CabledObjectModel
10from netbox.models import ChangeLoggedModel, OrganizationalModel, PrimaryModel
11from netbox.models.features import (
12 ContactsMixin,
13 CustomFieldsMixin,
14 CustomLinksMixin,
15 ExportTemplatesMixin,
16 ImageAttachmentsMixin,
17 TagsMixin,
18)
19from netbox.models.mixins import DistanceMixin
20from utilities.data import normalize_update_fields
21from utilities.string import title
23from .base import BaseCircuitType
25__all__ = (
26 'Circuit',
27 'CircuitGroup',
28 'CircuitGroupAssignment',
29 'CircuitTermination',
30 'CircuitType',
31)
34class CircuitType(BaseCircuitType):
35 """
36 Circuits can be organized by their functional role. For example, a user might wish to define CircuitTypes named
37 "Long Haul," "Metro," or "Out-of-Band".
38 """
39 class Meta:
40 ordering = ('name',)
41 verbose_name = _('circuit type')
42 verbose_name_plural = _('circuit types')
45class Circuit(ContactsMixin, ImageAttachmentsMixin, DistanceMixin, PrimaryModel):
46 """
47 A communications circuit connects two points. Each Circuit belongs to a Provider; Providers may have multiple
48 circuits. Each circuit is also assigned a CircuitType and a Site, and may optionally be assigned to a particular
49 ProviderAccount. Circuit port speed and commit rate are measured in Kbps.
50 """
51 cid = models.CharField(
52 max_length=100,
53 verbose_name=_('circuit ID'),
54 help_text=_('Unique circuit ID')
55 )
56 provider = models.ForeignKey(
57 to='circuits.Provider',
58 on_delete=models.PROTECT,
59 related_name='circuits'
60 )
61 provider_account = models.ForeignKey(
62 to='circuits.ProviderAccount',
63 on_delete=models.PROTECT,
64 related_name='circuits',
65 blank=True,
66 null=True
67 )
68 type = models.ForeignKey(
69 to='circuits.CircuitType',
70 on_delete=models.PROTECT,
71 related_name='circuits'
72 )
73 status = models.CharField(
74 verbose_name=_('status'),
75 max_length=50,
76 choices=CircuitStatusChoices,
77 default=CircuitStatusChoices.STATUS_ACTIVE
78 )
79 tenant = models.ForeignKey(
80 to='tenancy.Tenant',
81 on_delete=models.PROTECT,
82 related_name='circuits',
83 blank=True,
84 null=True
85 )
86 install_date = models.DateField(
87 blank=True,
88 null=True,
89 verbose_name=_('installed')
90 )
91 termination_date = models.DateField(
92 blank=True,
93 null=True,
94 verbose_name=_('terminates')
95 )
96 commit_rate = models.PositiveIntegerField(
97 blank=True,
98 null=True,
99 verbose_name=_('commit rate (Kbps)'),
100 help_text=_("Committed rate")
101 )
103 # Cache associated CircuitTerminations
104 termination_a = models.ForeignKey(
105 to='circuits.CircuitTermination',
106 on_delete=models.SET_NULL,
107 related_name='+',
108 editable=False,
109 blank=True,
110 null=True
111 )
112 termination_z = models.ForeignKey(
113 to='circuits.CircuitTermination',
114 on_delete=models.SET_NULL,
115 related_name='+',
116 editable=False,
117 blank=True,
118 null=True
119 )
121 group_assignments = GenericRelation(
122 to='circuits.CircuitGroupAssignment',
123 content_type_field='member_type',
124 object_id_field='member_id',
125 related_query_name='circuit'
126 )
128 clone_fields = (
129 'provider', 'provider_account', 'type', 'status', 'tenant', 'install_date', 'termination_date', 'commit_rate',
130 'description',
131 )
132 prerequisite_models = (
133 'circuits.CircuitType',
134 'circuits.Provider',
135 )
137 class Meta:
138 ordering = ['provider', 'provider_account', 'cid']
139 constraints = (
140 models.UniqueConstraint(
141 fields=('provider', 'cid'),
142 name='%(app_label)s_%(class)s_unique_provider_cid'
143 ),
144 models.UniqueConstraint(
145 fields=('provider_account', 'cid'),
146 name='%(app_label)s_%(class)s_unique_provideraccount_cid'
147 ),
148 )
149 indexes = (
150 models.Index(fields=('provider', 'provider_account', 'cid')), # Default ordering
151 )
152 verbose_name = _('circuit')
153 verbose_name_plural = _('circuits')
155 def __str__(self):
156 return self.cid
158 def get_status_color(self):
159 return CircuitStatusChoices.colors.get(self.status)
161 def clean(self):
162 super().clean()
164 if self.provider_account and self.provider != self.provider_account.provider:
165 raise ValidationError({'provider_account': "The assigned account must belong to the assigned provider."})
168class CircuitGroup(OrganizationalModel):
169 """
170 An administrative grouping of Circuits.
171 """
172 tenant = models.ForeignKey(
173 to='tenancy.Tenant',
174 on_delete=models.PROTECT,
175 related_name='circuit_groups',
176 blank=True,
177 null=True
178 )
180 class Meta:
181 ordering = ('name',)
182 verbose_name = _('circuit group')
183 verbose_name_plural = _('circuit groups')
185 def __str__(self):
186 return self.name
189class CircuitGroupAssignment(CustomFieldsMixin, ExportTemplatesMixin, TagsMixin, ChangeLoggedModel):
190 """
191 Assignment of a physical or virtual circuit to a CircuitGroup with an optional priority.
192 """
193 member_type = models.ForeignKey(
194 to='contenttypes.ContentType',
195 on_delete=models.PROTECT,
196 related_name='+'
197 )
198 member_id = models.PositiveBigIntegerField(
199 verbose_name=_('member ID')
200 )
201 member = GenericForeignKey(
202 ct_field='member_type',
203 fk_field='member_id'
204 )
205 group = models.ForeignKey(
206 to='circuits.CircuitGroup',
207 on_delete=models.CASCADE,
208 related_name='assignments'
209 )
210 priority = models.CharField(
211 verbose_name=_('priority'),
212 max_length=50,
213 choices=CircuitPriorityChoices,
214 blank=True,
215 null=True
216 )
217 prerequisite_models = (
218 'circuits.CircuitGroup',
219 )
221 class Meta:
222 ordering = ('group', 'member_type', 'member_id', 'priority', 'pk')
223 constraints = (
224 models.UniqueConstraint(
225 fields=('member_type', 'member_id', 'group'),
226 name='%(app_label)s_%(class)s_unique_member_group'
227 ),
228 )
229 indexes = (
230 models.Index(fields=('group', 'member_type', 'member_id', 'priority', 'id')), # Default ordering
231 )
232 verbose_name = _('Circuit group assignment')
233 verbose_name_plural = _('Circuit group assignments')
235 def __str__(self):
236 if self.priority:
237 return f"{self.group} ({self.get_priority_display()})"
238 return str(self.group)
240 def get_absolute_url(self):
241 return reverse('circuits:circuitgroupassignment', args=[self.pk])
244class CircuitTermination(
245 CustomFieldsMixin,
246 CustomLinksMixin,
247 ExportTemplatesMixin,
248 TagsMixin,
249 ChangeLoggedModel,
250 CabledObjectModel
251):
252 circuit = models.ForeignKey(
253 to='circuits.Circuit',
254 on_delete=models.CASCADE,
255 related_name='terminations'
256 )
257 term_side = models.CharField(
258 max_length=1,
259 choices=CircuitTerminationSideChoices,
260 verbose_name=_('termination side')
261 )
262 termination_type = models.ForeignKey(
263 to='contenttypes.ContentType',
264 on_delete=models.PROTECT,
265 related_name='+',
266 blank=True,
267 null=True
268 )
269 termination_id = models.PositiveBigIntegerField(
270 blank=True,
271 null=True
272 )
273 termination = GenericForeignKey(
274 ct_field='termination_type',
275 fk_field='termination_id'
276 )
277 port_speed = models.PositiveIntegerField(
278 verbose_name=_('port speed (Kbps)'),
279 blank=True,
280 null=True,
281 help_text=_('Physical circuit speed')
282 )
283 upstream_speed = models.PositiveIntegerField(
284 blank=True,
285 null=True,
286 verbose_name=_('upstream speed (Kbps)'),
287 help_text=_('Upstream speed, if different from port speed')
288 )
289 xconnect_id = models.CharField(
290 max_length=50,
291 blank=True,
292 verbose_name=_('cross-connect ID'),
293 help_text=_('ID of the local cross-connect')
294 )
295 pp_info = models.CharField(
296 max_length=100,
297 blank=True,
298 verbose_name=_('patch panel/port(s)'),
299 help_text=_('Patch panel ID and port number(s)')
300 )
301 description = models.CharField(
302 verbose_name=_('description'),
303 max_length=200,
304 blank=True
305 )
307 # Cached associations to enable efficient filtering
308 _provider_network = models.ForeignKey(
309 to='circuits.ProviderNetwork',
310 on_delete=models.PROTECT,
311 related_name='circuit_terminations',
312 blank=True,
313 null=True
314 )
315 _location = models.ForeignKey(
316 to='dcim.Location',
317 on_delete=models.CASCADE,
318 related_name='circuit_terminations',
319 blank=True,
320 null=True
321 )
322 _site = models.ForeignKey(
323 to='dcim.Site',
324 on_delete=models.CASCADE,
325 related_name='circuit_terminations',
326 blank=True,
327 null=True
328 )
329 _region = models.ForeignKey(
330 to='dcim.Region',
331 on_delete=models.CASCADE,
332 related_name='circuit_terminations',
333 blank=True,
334 null=True
335 )
336 _site_group = models.ForeignKey(
337 to='dcim.SiteGroup',
338 on_delete=models.CASCADE,
339 related_name='circuit_terminations',
340 blank=True,
341 null=True
342 )
344 class Meta:
345 ordering = ['circuit', 'term_side']
346 constraints = (
347 models.UniqueConstraint(
348 fields=('circuit', 'term_side'),
349 name='%(app_label)s_%(class)s_unique_circuit_term_side'
350 ),
351 )
352 indexes = (
353 models.Index(fields=('termination_type', 'termination_id')),
354 )
355 verbose_name = _('circuit termination')
356 verbose_name_plural = _('circuit terminations')
358 def __init__(self, *args, **kwargs):
359 super().__init__(*args, **kwargs)
361 # Cache original values to detect changes
362 self._orig_circuit_id = self.__dict__.get('circuit_id')
363 self._orig_term_side = self.__dict__.get('term_side')
365 def __str__(self):
366 return f'{self.circuit}: Termination {self.term_side}'
368 def get_absolute_url(self):
369 return reverse('circuits:circuittermination', args=[self.pk])
371 def clean(self):
372 if self.termination_type and not (self.termination or self.termination_id):
373 termination_type = self.termination_type.model_class()
374 raise ValidationError(
375 _("Please select a {termination_type}.").format(
376 termination_type=_(title(termination_type._meta.verbose_name))
377 )
378 )
379 super().clean()
381 if self.termination is None:
382 raise ValidationError(_("A circuit termination must attach to a terminating object."))
384 def save(self, *args, **kwargs):
385 is_new = self._state.adding
386 update_fields = normalize_update_fields(kwargs)
388 # Only consider circuit/term_side changes if those fields
389 # are actually being persisted
390 if update_fields is not None:
391 tracking_relevant = 'circuit' in update_fields or 'term_side' in update_fields
392 else:
393 tracking_relevant = True
395 circuit_changed = tracking_relevant and self._orig_circuit_id and self._orig_circuit_id != self.circuit_id
396 term_side_changed = tracking_relevant and self._orig_term_side and self._orig_term_side != self.term_side
398 # Cache objects associated with the terminating object (for filtering)
399 self.cache_related_objects()
401 super().save(*args, **kwargs)
403 # Clear the old termination reference if circuit or term_side changed
404 if circuit_changed or term_side_changed:
405 old_termination_name = f'termination_{self._orig_term_side.lower()}'
406 Circuit.objects.filter(pk=self._orig_circuit_id).update(**{old_termination_name: None})
408 # Update the cache if this is a new termination or circuit/term_side changed
409 if is_new or circuit_changed or term_side_changed:
410 # Update the new circuit's termination reference
411 termination_name = f'termination_{self.term_side.lower()}'
412 Circuit.objects.filter(pk=self.circuit_id).update(**{termination_name: self.pk})
414 # Update cached values for subsequent saves
415 self._orig_circuit_id = self.circuit_id
416 self._orig_term_side = self.term_side
418 def cache_related_objects(self):
419 self._provider_network = self._region = self._site_group = self._site = self._location = None
420 if self.termination_type:
421 termination_type = self.termination_type.model_class()
422 if termination_type == apps.get_model('dcim', 'region'):
423 self._region = self.termination
424 elif termination_type == apps.get_model('dcim', 'sitegroup'):
425 self._site_group = self.termination
426 elif termination_type == apps.get_model('dcim', 'site'):
427 self._region = self.termination.region
428 self._site_group = self.termination.group
429 self._site = self.termination
430 elif termination_type == apps.get_model('dcim', 'location'):
431 self._region = self.termination.site.region
432 self._site_group = self.termination.site.group
433 self._site = self.termination.site
434 self._location = self.termination
435 elif termination_type == apps.get_model('circuits', 'providernetwork'):
436 self._provider_network = self.termination
437 cache_related_objects.alters_data = True
439 def to_objectchange(self, action):
440 objectchange = super().to_objectchange(action)
441 objectchange.related_object = self.circuit
442 return objectchange
444 @property
445 def parent_object(self):
446 return self.circuit
448 def get_peer_termination(self):
449 peer_side = 'Z' if self.term_side == 'A' else 'A'
450 try:
451 return CircuitTermination.objects.prefetch_related('termination').get(
452 circuit=self.circuit,
453 term_side=peer_side
454 )
455 except CircuitTermination.DoesNotExist:
456 return None