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

netbox/dcim/signals.py

@@ -323,12 +323,34 @@ def nullify_connected_endpoints(instance, **kwargs):
     Disassociate the Cable from the termination object, and retrace any affected CablePaths.
     """
     model = instance.termination_type.model_class()
-    model.objects.filter(pk=instance.termination_id).update(
-        cable=None,
-        cable_end=None,
-        cable_connector=None,
-        cable_positions=None,
-    )
+
+    # 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.
+    termination = None
+    if Cable._is_being_deleted(instance.cable_id):
+        termination = model.objects.filter(pk=instance.termination_id).first()
+
+    if termination is not None:
+        termination.cable_id = instance.cable_id
+        termination.cable_end = instance.cable_end
+        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()
+    else:
+        # Already recorded by CableTermination.delete(), or the terminating object is going away too.
+        model.objects.filter(pk=instance.termination_id).update(
+            cable=None,
+            cable_end=None,
+            cable_connector=None,
+            cable_positions=None,
+        )
 
     # If the removed termination was a channelized interface, also clear the cable attributes mirrored onto its channel
     # subinterfaces. This must happen before the retrace below so that each channel's (now dead) path is torn down

netbox/dcim/tests/test_models.py

@@ -1,21 +1,25 @@
+import uuid
 from decimal import Decimal
 
 from django.core.exceptions import ValidationError
 from django.db import connection
 from django.db.models import ProtectedError
 from django.db.models.signals import post_save
-from django.test import TestCase, tag
+from django.test import RequestFactory, TestCase, tag
 from django.test.utils import CaptureQueriesContext
 
 from circuits.models import *
-from core.models import ObjectType
+from core.choices import ObjectChangeActionChoices
+from core.models import ObjectChange, ObjectType
 from dcim.choices import *
 from dcim.models import *
 from extras.events import serialize_for_event
 from extras.models import CustomField
 from ipam.models import Prefix
 from netbox.choices import DiameterUnitChoices, FlowRateUnitChoices, WeightUnitChoices
+from netbox.context_managers import event_tracking
 from tenancy.models import Tenant
+from users.models import User
 from utilities.data import drange
 from virtualization.models import Cluster, ClusterType
 
@@ -2922,6 +2926,103 @@ class CableTerminationTestCase(TestCase):
             cable_termination.cache_related_objects()
 
 
+class CableDisconnectChangeLoggingTestCase(TestCase):
+    """
+    Deleting a Cable must change-log the disconnect on each terminating object. Its CableTerminations are
+    deleted by the cascade, so CableTermination.delete() -- which records the disconnect when a termination
+    is removed from a Cable -- never runs.
+    """
+
+    @classmethod
+    def setUpTestData(cls):
+        site = Site.objects.create(name='Test Site 1', slug='test-site-1')
+        manufacturer = Manufacturer.objects.create(name='Test Manufacturer 1', slug='test-manufacturer-1')
+        device_type = DeviceType.objects.create(
+            manufacturer=manufacturer, model='Test Device Type 1', slug='test-device-type-1'
+        )
+        role = DeviceRole.objects.create(name='Test Device Role 1', slug='test-device-role-1')
+        cls.device1 = Device.objects.create(
+            name='Device 1', site=site, device_type=device_type, role=role
+        )
+        cls.device2 = Device.objects.create(
+            name='Device 2', site=site, device_type=device_type, role=role
+        )
+        cls.interface1 = Interface.objects.create(
+            device=cls.device1, name='eth0', type=InterfaceTypeChoices.TYPE_1GE_FIXED
+        )
+        cls.interface2 = Interface.objects.create(
+            device=cls.device2, name='eth0', type=InterfaceTypeChoices.TYPE_1GE_FIXED
+        )
+        cls.interface3 = Interface.objects.create(
+            device=cls.device2, name='eth1', type=InterfaceTypeChoices.TYPE_1GE_FIXED
+        )
+        cls.user = User.objects.create_user(username='testuser')
+
+    def _request(self):
+        request = RequestFactory().get('/')
+        request.id = uuid.uuid4()
+        request.user = self.user
+        return request
+
+    def _connect(self, termination_a, termination_b):
+        with event_tracking(self._request()):
+            cable = Cable(a_terminations=[termination_a], b_terminations=[termination_b])
+            cable.save()
+        return cable
+
+    def _updates(self, interface):
+        return ObjectChange.objects.filter(
+            changed_object_type=ObjectType.objects.get_for_model(Interface),
+            changed_object_id=interface.pk,
+            action=ObjectChangeActionChoices.ACTION_UPDATE,
+        ).order_by('time')
+
+    @tag('regression')  # Ref: netbox-branching#631
+    def test_cable_deletion_records_disconnect(self):
+        cable = self._connect(self.interface1, self.interface2)
+        cable_pk = cable.pk
+
+        with event_tracking(self._request()):
+            Cable.objects.get(pk=cable_pk).delete()
+
+        for interface in (self.interface1, self.interface2):
+            # 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()
+            self.assertEqual(change.prechange_data['cable'], cable_pk)
+            self.assertIsNone(change.postchange_data['cable'])
+            self.assertIsNone(change.postchange_data['cable_end'])
+            self.assertIsNone(Interface.objects.get(pk=interface.pk).cable_id)
+
+    def test_termination_removal_records_disconnect(self):
+        # Removing a termination from a Cable (rather than deleting the Cable) is recorded by
+        # CableTermination.delete(); the disconnect must not be logged twice or logged as a no-op.
+        cable = self._connect(self.interface1, self.interface2)
+        cable_pk = cable.pk
+
+        request = self._request()
+        with event_tracking(request):
+            cable = Cable.objects.get(pk=cable_pk)
+            cable.b_terminations = [self.interface3]
+            cable.save()
+
+        changes = self._updates(self.interface2).filter(request_id=request.id)
+        self.assertEqual(changes.count(), 1)
+        self.assertEqual(changes[0].prechange_data['cable'], cable_pk)
+        self.assertIsNone(changes[0].postchange_data['cable'])
+
+    def test_terminating_object_deletion_records_no_update(self):
+        # The cascade from deleting the terminating object itself reaches the same handler, but the object
+        # is on its way out: it must be left alone rather than resurrected as an update.
+        self._connect(self.interface1, self.interface2)
+
+        request = self._request()
+        with event_tracking(request):
+            Interface.objects.get(pk=self.interface1.pk).delete()
+
+        self.assertFalse(self._updates(self.interface1).filter(request_id=request.id).exists())
+
+
 class VirtualDeviceContextTestCase(TestCase):
 
     @classmethod