From 4564f591ba2e30d378176844566d551f5ddc5bd6 Mon Sep 17 00:00:00 2001 From: Arthur Date: Thu, 10 Sep 2026 09:51:10 -0700 Subject: [PATCH] Fix #23160 - Change-log the disconnect on terminating objects when a Cable is deleted --- netbox/dcim/signals.py | 34 ++++++++-- netbox/dcim/tests/test_models.py | 105 ++++++++++++++++++++++++++++++- 2 files changed, 131 insertions(+), 8 deletions(-) diff --git a/netbox/dcim/signals.py b/netbox/dcim/signals.py index f051fe146..208e9f591 100644 --- a/netbox/dcim/signals.py +++ b/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 diff --git a/netbox/dcim/tests/test_models.py b/netbox/dcim/tests/test_models.py index 036426665..da6041500 100644 --- a/netbox/dcim/tests/test_models.py +++ b/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