RomanNum3ral
/
netbox
mirror of https://github.com/netbox-community/netbox.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix #23160 - Change-log the disconnect on terminating objects when a Cable is deleted

This commit is contained in:
Arthur 2026-09-10 10:41:19 -07:00
parent 4564f591ba
commit ab55a76566
3 changed files with 118 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -385,20 +385,32 @@ class Cable(PrimaryModel):
self._terminations_modified = False
def delete(self, *args, **kwargs):
# Track this Cable as being deleted so the post_delete signal handler
# for cascaded CableTerminations can skip redundant path retracing;
# retrace_cable_paths() will retrace each affected path once after the
# Cable itself is deleted. Cache the PK locally because super().delete()
# clears self.pk before the finally block runs. The tracking set lives
# on a threading.local() to isolate concurrent deletions across threads.
if not hasattr(Cable._deletion_tracking, 'pks'):
Cable._deletion_tracking.pks = set()
# Cache the PK locally because super().delete() clears self.pk before the finally block runs. The
# tracking itself is done by the pre_delete/post_delete receivers in dcim.signals, which also cover a
# queryset delete; this pairing just guarantees the PK is discarded if the delete raises.
pk = self.pk
Cable._deletion_tracking.pks.add(pk)
Cable._track_deletion(pk)
try:
return super().delete(*args, **kwargs)
finally:
Cable._deletion_tracking.pks.discard(pk)
Cable._untrack_deletion(pk)
@classmethod
def _track_deletion(cls, pk):
"""
Track a Cable as being deleted, so that the post_delete handler for its cascaded CableTerminations can
record the disconnect on each terminating object and skip redundant path retracing (retrace_cable_paths()
retraces each affected path once, after the Cable itself is deleted). The tracking set lives on a
threading.local() to isolate concurrent deletions across threads.
"""
if not hasattr(cls._deletion_tracking, 'pks'):
cls._deletion_tracking.pks = set()
cls._deletion_tracking.pks.add(pk)
@classmethod
def _untrack_deletion(cls, pk):
if hasattr(cls._deletion_tracking, 'pks'):
cls._deletion_tracking.pks.discard(pk)
@classmethod
def _is_being_deleted(cls, pk):

42
netbox/dcim/signals.py
View File

@ -2,7 +2,7 @@ import logging
from django.db import transaction
from django.db.models import Q
from django.db.models.signals import post_delete, post_save, pre_save
from django.db.models.signals import post_delete, post_save, pre_delete, pre_save
from django.dispatch import receiver
from dcim.choices import CableEndChoices, LinkStatusChoices
@ -306,6 +306,24 @@ def retrace_cable_paths(instance, **kwargs):
cablepath.retrace()
@receiver(pre_delete, sender=Cable)
def track_cable_deletion(instance, **kwargs):
"""
Flag the Cable as being deleted for nullify_connected_endpoints() below, which runs for each of its
cascaded CableTerminations. Tracking here rather than only in Cable.delete() covers a queryset delete,
which never calls the model's delete() method.
"""
Cable._track_deletion(instance.pk)
@receiver(post_delete, sender=Cable)
def untrack_cable_deletion(instance, **kwargs):
# Registered after retrace_cable_paths() so that the flag is still set while it runs. A delete that raises
# between the two signals leaves the PK tracked; Cable.delete() clears it in a finally, and for a queryset
# delete the transaction rolls back with only this stale entry left behind.
Cable._untrack_deletion(instance.pk)
@receiver((post_delete, post_save), sender=PortMapping)
def update_passthrough_port_paths(instance, **kwargs):
"""
@ -326,8 +344,10 @@ def nullify_connected_endpoints(instance, **kwargs):
# Deleting a Cable deletes its terminations in bulk, bypassing CableTermination.delete() and the
# change-logged clear it performs on the terminating object; do the same here so the disconnect is
# recorded. `cable` is a SET_NULL FK which the deletion collector has already nulled by now, so restore
# the pre-delete values before snapshotting or the record shows no change.
# recorded. `cable` is a SET_NULL FK which the deletion collector has already nulled by now, so the
# pre-delete values are restored before snapshotting, or the record would show no change. They are
# restored field by field rather than via set_cable_termination(), whose Interface override would
# propagate the cable back onto the channel subinterfaces we are about to clear.
termination = None
if Cable._is_being_deleted(instance.cable_id):
termination = model.objects.filter(pk=instance.termination_id).first()
@ -338,11 +358,17 @@ def nullify_connected_endpoints(instance, **kwargs):
termination.cable_connector = instance.connector
termination.cable_positions = instance.positions
termination.snapshot()
termination.cable = None
termination.cable_end = None
termination.cable_connector = None
termination.cable_positions = None
termination.save()
termination.clear_cable_termination(instance)
update_fields = ['cable', 'cable_end', 'cable_connector', 'cable_positions', 'last_updated']
# retrace_cable_paths() tears down the originating path once the Cable itself is deleted, clearing
# _path outside the changelog. Clear it here so the recorded state doesn't outlive the path.
if isinstance(termination, PathEndpoint):
termination._path = None
update_fields.append('_path')
# A narrow write: this row was read mid-cascade, and its full save() would pull in unrelated work
termination.save(update_fields=update_fields)
else:
# Already recorded by CableTermination.delete(), or the terminating object is going away too.
model.objects.filter(pk=instance.termination_id).update(

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

@ -2956,6 +2956,12 @@ class CableDisconnectChangeLoggingTestCase(TestCase):
cls.interface3 = Interface.objects.create(
device=cls.device2, name='eth1', type=InterfaceTypeChoices.TYPE_1GE_FIXED
)
cls.rear_port = RearPort.objects.create(
device=cls.device2, name='Rear Port 1', type=PortTypeChoices.TYPE_8P8C, positions=4
)
cls.front_port = FrontPort.objects.create(
device=cls.device2, name='Front Port 1', type=PortTypeChoices.TYPE_8P8C, positions=4
)
cls.user = User.objects.create_user(username='testuser')
def _request(self):
@ -2964,16 +2970,16 @@ class CableDisconnectChangeLoggingTestCase(TestCase):
request.user = self.user
return request
def _connect(self, termination_a, termination_b):
def _connect(self, termination_a, termination_b, **kwargs):
with event_tracking(self._request()):
cable = Cable(a_terminations=[termination_a], b_terminations=[termination_b])
cable = Cable(a_terminations=[termination_a], b_terminations=[termination_b], **kwargs)
cable.save()
return cable
def _updates(self, interface):
def _updates(self, obj):
return ObjectChange.objects.filter(
changed_object_type=ObjectType.objects.get_for_model(Interface),
changed_object_id=interface.pk,
changed_object_type=ObjectType.objects.get_for_model(obj),
changed_object_id=obj.pk,
action=ObjectChangeActionChoices.ACTION_UPDATE,
).order_by('time')
@ -2989,10 +2995,61 @@ class CableDisconnectChangeLoggingTestCase(TestCase):
# One update for the connect, one for the disconnect
self.assertEqual(self._updates(interface).count(), 2, f'No disconnect recorded for {interface}')
change = self._updates(interface).last()
# The cable attributes are cleared in the database before this is recorded, so the pre-change
# data is only correct if they were restored before the snapshot was taken
self.assertEqual(change.prechange_data['cable'], cable_pk)
self.assertIn(change.prechange_data['cable_end'], (CableEndChoices.SIDE_A, CableEndChoices.SIDE_B))
self.assertIsNone(change.postchange_data['cable'])
self.assertIsNone(change.postchange_data['cable_end'])
# The originating path is deleted along with the Cable, outside the changelog
self.assertIsNone(change.postchange_data['_path'])
self.assertIsNone(Interface.objects.get(pk=interface.pk).cable_id)
self.assertFalse(CablePath.objects.exists())
def test_profiled_cable_deletion_records_connector_and_positions(self):
cable = self._connect(self.interface1, self.rear_port, profile=CableProfileChoices.SINGLE_1C4P)
cable_pk = cable.pk
connector = Interface.objects.get(pk=self.interface1.pk).cable_connector
positions = Interface.objects.get(pk=self.interface1.pk).cable_positions
self.assertIsNotNone(connector)
self.assertTrue(positions)
with event_tracking(self._request()):
Cable.objects.get(pk=cable_pk).delete()
change = self._updates(self.interface1).last()
self.assertEqual(change.prechange_data['cable_connector'], connector)
self.assertEqual(change.prechange_data['cable_positions'], positions)
self.assertIsNone(change.postchange_data['cable_connector'])
self.assertIsNone(change.postchange_data['cable_positions'])
def test_queryset_deletion_records_disconnect(self):
# A queryset delete never calls Cable.delete(), so the disconnect is tracked by a pre_delete receiver
cable = self._connect(self.interface1, self.interface2)
cable_pk = cable.pk
with event_tracking(self._request()):
Cable.objects.filter(pk=cable_pk).delete()
change = self._updates(self.interface1).last()
self.assertEqual(change.prechange_data['cable'], cable_pk)
self.assertIsNone(change.postchange_data['cable'])
def test_non_path_endpoint_termination(self):
# A front port carries no _path of its own; the disconnect is recorded the same way
cable = self._connect(self.interface1, self.front_port)
cable_pk = cable.pk
with event_tracking(self._request()):
Cable.objects.get(pk=cable_pk).delete()
change = self._updates(self.front_port).last()
self.assertEqual(change.prechange_data['cable'], cable_pk)
self.assertIsNone(change.postchange_data['cable'])
self.assertIsNone(FrontPort.objects.get(pk=self.front_port.pk).cable_id)
def test_termination_removal_records_disconnect(self):
# Removing a termination from a Cable (rather than deleting the Cable) is recorded by