Arthur 1 dag sedan
förälder
incheckning
1f6c5f305c
2 ändrade filer med 120 tillägg och 4 borttagningar
  1. 24 3
      netbox/dcim/models/device_components.py
  2. 96 1
      netbox/dcim/tests/test_channelization.py

+ 24 - 3
netbox/dcim/models/device_components.py

@@ -1,6 +1,7 @@
 from functools import cached_property
 from functools import cached_property
 
 
 from django.contrib.contenttypes.fields import GenericForeignKey, GenericRelation
 from django.contrib.contenttypes.fields import GenericForeignKey, GenericRelation
+from django.contrib.contenttypes.models import ContentType
 from django.contrib.postgres.fields import ArrayField
 from django.contrib.postgres.fields import ArrayField
 from django.contrib.postgres.indexes import GistIndex
 from django.contrib.postgres.indexes import GistIndex
 from django.core.exceptions import ObjectDoesNotExist, ValidationError
 from django.core.exceptions import ObjectDoesNotExist, ValidationError
@@ -19,6 +20,7 @@ from dcim.models.mixins import (
     InterfaceValidationMixin,
     InterfaceValidationMixin,
     MaxFlowMixin,
     MaxFlowMixin,
 )
 )
+from dcim.utils import rebuild_paths
 from netbox.choices import ColorChoices
 from netbox.choices import ColorChoices
 from netbox.models import NetBoxModel, OrganizationalModel
 from netbox.models import NetBoxModel, OrganizationalModel
 from netbox.models.features import ChangeLoggingMixin
 from netbox.models.features import ChangeLoggingMixin
@@ -1427,12 +1429,31 @@ class Interface(
 
 
     def update_dependent_objects(self):
     def update_dependent_objects(self):
         """
         """
-        Mirror a channelized parent's cable onto this channel subinterface and recreate the cable's paths.
+        Reconcile a channel subinterface's mirrored cable attributes with its channelized parent, and retrace the
+        CablePaths which traverse this interface.
         """
         """
+        from dcim.models import CablePath, CableTermination
+
+        parent = None
         if self.channel_id and self.parent_id:
         if self.channel_id and self.parent_id:
             parent = Interface.objects.filter(pk=self.parent_id, cable__isnull=False).select_related('cable').first()
             parent = Interface.objects.filter(pk=self.parent_id, cable__isnull=False).select_related('cable').first()
-            if parent and parent.channels:
-                parent.cable.update_dependent_objects()
+        if parent and parent.channels:
+            parent.cable.update_dependent_objects()
+        elif self.cable_id and not CableTermination.objects.filter(
+            termination_type=ContentType.objects.get_for_model(Interface), termination_id=self.pk
+        ).exists():
+            # A cable with no termination of this interface's own was mirrored from a former channelized parent
+            Interface.objects.filter(pk=self.pk).update(
+                cable=None, cable_end='', cable_connector=None, cable_positions=None
+            )
+            for cablepath in CablePath.objects.filter(_nodes__contains=self):
+                if self in cablepath.origins:
+                    cablepath.delete()
+        else:
+            return
+
+        # Retrace paths still traversing this interface from a former parent's cable
+        rebuild_paths([self])
     update_dependent_objects.alters_data = True
     update_dependent_objects.alters_data = True
 
 
     def propagate_channel_cables(self):
     def propagate_channel_cables(self):

+ 96 - 1
netbox/dcim/tests/test_channelization.py

@@ -3,7 +3,7 @@ from unittest import mock
 
 
 from django.contrib.contenttypes.models import ContentType
 from django.contrib.contenttypes.models import ContentType
 from django.core.exceptions import ValidationError
 from django.core.exceptions import ValidationError
-from django.db import connection, router
+from django.db import connection, router, transaction
 from django.test import Client, TestCase, TransactionTestCase
 from django.test import Client, TestCase, TransactionTestCase
 from django.test.utils import CaptureQueriesContext
 from django.test.utils import CaptureQueriesContext
 from django.urls import reverse
 from django.urls import reverse
@@ -882,6 +882,101 @@ class ChannelizedCablePathTestCase(BaseCablePathTestCase):
         self.assertCurrentPathExists((far[3], cable, channel), is_complete=True, is_active=True)
         self.assertCurrentPathExists((far[3], cable, channel), is_complete=True, is_active=True)
         self.assertEqual(CablePath.objects.count(), 8)
         self.assertEqual(CablePath.objects.count(), 8)
 
 
+    def _assert_hook_matches_save(self, channel, **attrs):
+        """
+        Assert that writing attrs to channel raw and calling update_dependent_objects() leaves the same cable
+        attributes and CablePaths as a normal save().
+        """
+        def snapshot():
+            interfaces = Interface.objects.order_by('pk').values_list(
+                'pk', 'cable_id', 'cable_end', 'cable_connector', 'cable_positions', '_path__path'
+            )
+            paths = sorted((p.path, p.is_complete, p.is_active, p.is_split) for p in CablePath.objects.all())
+            return list(interfaces), paths
+
+        class Rollback(Exception):
+            pass
+
+        try:
+            with transaction.atomic():
+                instance = Interface.objects.get(pk=channel.pk)
+                for name, value in attrs.items():
+                    setattr(instance, name, value)
+                instance.full_clean()
+                instance.save()
+                expected = snapshot()
+                raise Rollback
+        except Rollback:
+            pass
+
+        instance = Interface.objects.get(pk=channel.pk)
+        for name, value in attrs.items():
+            setattr(instance, name, value)
+        instance.save_base(raw=True)
+        self.update_dependent_objects(instance)
+
+        self.assertEqual(snapshot(), expected)
+
+    def _create_breakout(self, name, far_prefix):
+        parent, channels = self._create_channelized_interface(name, 4)
+        far = [
+            Interface.objects.create(
+                device=self.device, name=f'{far_prefix}{i}', type=InterfaceTypeChoices.TYPE_10GE_SFP_PLUS
+            )
+            for i in range(4)
+        ]
+        cable = Cable(profile=CableProfileChoices.BREAKOUT_1C4P_4C1P, a_terminations=[parent], b_terminations=far)
+        cable.clean()
+        cable.save()
+        return parent, channels, far, cable
+
+    def test_125_update_dependent_objects_clears_raw_detached_channel(self):
+        """
+        A channel subinterface detached by a raw write must drop its mirrored cable attributes and paths. Refs: #23595
+        """
+        _, channels, far, _ = self._create_breakout('et0', 'xe')
+
+        self._assert_hook_matches_save(
+            channels[3], parent=None, channel_id=None, type=InterfaceTypeChoices.TYPE_10GE_SFP_PLUS
+        )
+
+        channel = Interface.objects.get(pk=channels[3].pk)
+        self.assertIsNone(channel.cable_id)
+        self.assertPathIsNotSet(channel)
+        self.assertFalse(CablePath.objects.filter(_nodes__contains=channel).exists())
+
+    def test_126_update_dependent_objects_retraces_raw_reparented_channel(self):
+        """
+        A channel subinterface moved to another channelized parent by a raw write must leave no path from its former
+        parent's far end. Refs: #23595
+        """
+        parent1, channels1, far1, cable1 = self._create_breakout('et0', 'xe')
+        parent2, channels2, far2, cable2 = self._create_breakout('et1', 'ge')
+        channels2[3].delete()
+
+        self._assert_hook_matches_save(channels1[3], parent=Interface.objects.get(pk=parent2.pk), name='et1:4')
+
+        channel = Interface.objects.get(pk=channels1[3].pk)
+        self.assertEqual(channel.cable_id, cable2.pk)
+        self.assertCurrentPathExists((channel, cable2, far2[3]), is_complete=True, is_active=True)
+        self.assertCurrentPathExists((far2[3], cable2, channel), is_complete=True, is_active=True)
+        self.assertFalse(CablePath.objects.filter(_nodes__contains=cable1).filter(_nodes__contains=channel).exists())
+
+    def test_127_update_dependent_objects_clears_channel_of_uncabled_parent(self):
+        """
+        A channel subinterface moved by a raw write to a channelized parent with no cable must drop its mirrored
+        cable attributes and paths. Refs: #23595
+        """
+        _, channels, _, _ = self._create_breakout('et0', 'xe')
+        parent2, channels2 = self._create_channelized_interface('et1', 4)
+        channels2[3].delete()
+
+        self._assert_hook_matches_save(channels[3], parent=Interface.objects.get(pk=parent2.pk), name='et1:4')
+
+        channel = Interface.objects.get(pk=channels[3].pk)
+        self.assertIsNone(channel.cable_id)
+        self.assertFalse(CablePath.objects.filter(_nodes__contains=channel).exists())
+
 
 
 class ChannelizedInterfaceTestCase(TestCase):
 class ChannelizedInterfaceTestCase(TestCase):
     """
     """