NeoAnd/netbox
1
0
Fork 0
mirror of https://github.com/netbox-community/netbox.git synced 2025-07-25 18:08:38 -06:00
Code Issues Actions 10 Packages Projects Releases Wiki Activity

15390 send trace_paths if cable terminations updated

Browse Source
This commit is contained in:
Arthur Hanson 2024-06-17 15:23:47 -07:00
parent 82893cab33
commit 1bd09b673b
1 changed files with 25 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -220,22 +220,24 @@ class Cable(PrimaryModel):
b_terminations = {ct.termination: ct for ct in self.terminations.filter(cable_end='B')}
# Delete stale CableTerminations
# Call CableTermination with no_traceoath_signal so signal handler isn't sent multiple
# times as we handle that below
if self._terminations_modified:
for termination, ct in a_terminations.items():
if termination.pk and termination not in self.a_terminations:
ct.delete()
ct.delete(no_tracepath_signal=True)
for termination, ct in b_terminations.items():
if termination.pk and termination not in self.b_terminations:
ct.delete()
ct.delete(no_tracepath_signal=True)
# Save new CableTerminations (if any)
if self._terminations_modified:
for termination in self.a_terminations:
if not termination.pk or termination not in a_terminations:
CableTermination(cable=self, cable_end='A', termination=termination).save()
CableTermination(cable=self, cable_end='A', termination=termination).save(no_tracepath_signal=True)
for termination in self.b_terminations:
if not termination.pk or termination not in b_terminations:
CableTermination(cable=self, cable_end='B', termination=termination).save()
CableTermination(cable=self, cable_end='B', termination=termination).save(no_tracepath_signal=True)
trace_paths.send(Cable, instance=self, created=_created)
@ -355,6 +357,7 @@ class CableTermination(ChangeLoggedModel):
raise ValidationError(_("Circuit terminations attached to a provider network may not be cabled."))
def save(self, *args, **kwargs):
no_tracepath_signal = kwargs.pop('no_tracepath_signal', False)
# Cache objects associated with the terminating object (for filtering)
self.cache_related_objects()
@ -369,22 +372,27 @@ class CableTermination(ChangeLoggedModel):
termination.cable_end = self.cable_end
termination.save()
if not created:
if self._orig_cable != self.cable:
# Need to send signal to both old and new cable
self._orig_cable._terminations_modified = True
trace_paths.send(Cable, instance=self._orig_cable, created=False)
self._orig_cable = self.cable
if not no_tracepath_signal:
if created:
self.cable._terminations_modified = True
trace_paths.send(Cable, instance=self.cable, created=False)
else:
if self._orig_cable and self._orig_cable != self.cable:
# Need to send signal to both old and new cable
self._orig_cable._terminations_modified = True
trace_paths.send(Cable, instance=self._orig_cable, created=False)
elif self._orig_cable_end != self.cable_end:
self._orig_cable_end = self.cable_end
self.cable._terminations_modified = True
trace_paths.send(Cable, instance=self.cable, created=False)
self._orig_cable = self.cable
self.cable._terminations_modified = True
trace_paths.send(Cable, instance=self.cable, created=False)
elif self._orig_cable_end and self._orig_cable_end != self.cable_end:
self._orig_cable_end = self.cable_end
self.cable._terminations_modified = True
trace_paths.send(Cable, instance=self.cable, created=False)
def delete(self, *args, **kwargs):
no_tracepath_signal = kwargs.pop('no_tracepath_signal', False)
# Delete the cable association on the terminating object
termination_model = self.termination._meta.model
@ -394,7 +402,8 @@ class CableTermination(ChangeLoggedModel):
)
super().delete(*args, **kwargs)
trace_paths.send(Cable, instance=self.cable, created=False)
if not no_tracepath_signal:
trace_paths.send(Cable, instance=self.cable, created=False)
def cache_related_objects(self):
"""