Arthur 1 день назад
Родитель
Сommit
84ada0bbb4
2 измененных файлов с 171 добавлено и 178 удалено
  1. 1 2
      netbox/dcim/models/cables.py
  2. 170 176
      netbox/dcim/tests/test_cablepaths.py

+ 1 - 2
netbox/dcim/models/cables.py

@@ -514,8 +514,7 @@ class Cable(PrimaryModel):
         """
         with transaction.atomic(using=router.db_for_write(CablePath)):
 
-            # A channelized parent mirrors its cable attributes onto its channel subinterfaces with a bulk
-            # write, which emits no change record: remirror them, or the retrace expands it to nothing
+            # Restore channel cable attributes omitted by bulk-update change logging.
             for ct in CableTermination.objects.filter(cable=self).prefetch_related('termination'):
                 if isinstance(ct.termination, Interface) and ct.termination.channels:
                     ct.termination.propagate_channel_cables()

+ 170 - 176
netbox/dcim/tests/test_cablepaths.py

@@ -16,6 +16,27 @@ class LegacyCablePathTestCase(BaseCablePathTestCase):
         3XX: Test responses to changes in existing objects
         4XX: Test to exclude specific cable topologies
     """
+    def _create_cable_raw(self, termination_a, termination_b, status=LinkStatusChoices.STATUS_CONNECTED):
+        """
+        Write a Cable and its terminations directly to the database, bypassing Cable.save(). Unprofiled
+        cables only: the connector & positions a profile assigns are not replicated here.
+        """
+        cable = Cable(status=status)
+        cable.save_base(raw=True)
+
+        for termination, cable_end in (
+            (termination_a, CableEndChoices.SIDE_A),
+            (termination_b, CableEndChoices.SIDE_B),
+        ):
+            ct = CableTermination(cable=cable, cable_end=cable_end, termination=termination)
+            ct.cache_related_objects()
+            ct.save_base(raw=True)
+            termination.cable = cable
+            termination.cable_end = cable_end
+            termination.save()
+
+        return cable
+
     def test_101_interface_to_interface(self):
         """
         [IF1] --C1-- [IF2]
@@ -2892,176 +2913,7 @@ class LegacyCablePathTestCase(BaseCablePathTestCase):
         interface3.refresh_from_db()
         self.assertPathIsNotSet(interface3)
 
-    def test_401_exclude_midspan_devices(self):
-        """
-        [IF1] --C1-- [FP1][Test Device][RP1] --C2-- [RP2][Test Device][FP2] --C3-- [IF2]
-                     [FP3][Test mid-span Device][RP3] --C4-- [RP4][Test mid-span Device][FP4] /
-        """
-        device = Device.objects.create(
-            site=self.site,
-            device_type=self.device.device_type,
-            role=self.device.role,
-            name='Test mid-span Device'
-        )
-        interface1 = Interface.objects.create(device=self.device, name='Interface 1')
-        interface2 = Interface.objects.create(device=self.device, name='Interface 2')
-        rearport1 = RearPort.objects.create(device=self.device, name='Rear Port 1')
-        rearport2 = RearPort.objects.create(device=self.device, name='Rear Port 2')
-        rearport3 = RearPort.objects.create(device=device, name='Rear Port 3')
-        rearport4 = RearPort.objects.create(device=device, name='Rear Port 4')
-        frontport1 = FrontPort.objects.create(device=self.device, name='Front Port 1')
-        frontport2 = FrontPort.objects.create(device=self.device, name='Front Port 2')
-        frontport3 = FrontPort.objects.create(device=self.device, name='Front Port 3')
-        frontport4 = FrontPort.objects.create(device=self.device, name='Front Port 4')
-        PortMapping.objects.bulk_create([
-            PortMapping(
-                device=self.device,
-                front_port=frontport1,
-                front_port_position=1,
-                rear_port=rearport1,
-                rear_port_position=1,
-            ),
-            PortMapping(
-                device=self.device,
-                front_port=frontport2,
-                front_port_position=1,
-                rear_port=rearport2,
-                rear_port_position=1,
-            ),
-            PortMapping(
-                device=self.device,
-                front_port=frontport3,
-                front_port_position=1,
-                rear_port=rearport3,
-                rear_port_position=1,
-            ),
-            PortMapping(
-                device=self.device,
-                front_port=frontport4,
-                front_port_position=1,
-                rear_port=rearport4,
-                rear_port_position=1,
-            ),
-        ])
-
-        cable2 = Cable(
-            a_terminations=[rearport1],
-            b_terminations=[rearport2],
-            label='C2'
-        )
-        cable2.save()
-        cable4 = Cable(
-            a_terminations=[rearport3],
-            b_terminations=[rearport4],
-            label='C4'
-        )
-        cable4.save()
-        self.assertEqual(CablePath.objects.count(), 0)
-
-        # Create cable1
-        cable1 = Cable(
-            a_terminations=[interface1],
-            b_terminations=[frontport1, frontport3],
-            label='C1'
-        )
-        with self.assertRaises(AbortRequest):
-            cable1.save()
-
-        self.assertPathDoesNotExist(
-            (
-                interface1, cable1, (frontport1, frontport3), (rearport1, rearport3), (cable2, cable4),
-                (rearport2, rearport4), (frontport2, frontport4)
-            ),
-            is_complete=False
-        )
-        self.assertEqual(CablePath.objects.count(), 0)
-
-        # Create cable 3
-        cable3 = Cable(
-            a_terminations=[frontport2, frontport4],
-            b_terminations=[interface2],
-            label='C3'
-        )
-
-        with self.assertRaises(AbortRequest):
-            cable3.save()
-
-        self.assertPathDoesNotExist(
-            (
-                interface2, cable3, (frontport2, frontport4), (rearport2, rearport4), (cable2, cable4),
-                (rearport1, rearport3), (frontport1, frontport2), cable1, interface1
-            ),
-            is_complete=True,
-            is_active=True
-        )
-        self.assertPathDoesNotExist(
-            (
-                interface1, cable1, (frontport1, frontport3), (rearport1, rearport3), (cable2, cable4),
-                (rearport2, rearport4), (frontport2, frontport4), cable3, interface2
-            ),
-            is_complete=True,
-            is_active=True
-        )
-        self.assertEqual(CablePath.objects.count(), 0)
-
-    def test_402_exclude_circuit_loopback(self):
-        interface = Interface.objects.create(device=self.device, name='Interface 1')
-        circuittermination1 = CircuitTermination.objects.create(
-            circuit=self.circuit,
-            termination=self.site,
-            term_side='A'
-        )
-        circuittermination2 = CircuitTermination.objects.create(
-            circuit=self.circuit,
-            termination=self.site,
-            term_side='Z'
-        )
-
-        # Create cables
-        cable = Cable(
-            a_terminations=[interface],
-            b_terminations=[circuittermination1, circuittermination2]
-        )
-        cable.save()
-
-        path = self.assertPathExists(
-            (interface, cable, (circuittermination1, circuittermination2)),
-            is_active=True,
-            is_complete=False,
-            is_split=True
-        )
-        self.assertEqual(CablePath.objects.count(), 1)
-        interface.refresh_from_db()
-        self.assertPathIsSet(interface, path)
-
-
-class CableDependentObjectsTestCase(BaseCablePathTestCase):
-    """
-    Test Cable.update_dependent_objects(), which retraces the paths of a Cable written to the database
-    by a process that bypasses save() (e.g. a tool replaying serialized changes).
-    """
-    def _create_cable_raw(self, termination_a, termination_b, status=LinkStatusChoices.STATUS_CONNECTED):
-        """
-        Write a Cable and its terminations directly to the database, bypassing Cable.save(). Unprofiled
-        cables only: the connector & positions a profile assigns are not replicated here.
-        """
-        cable = Cable(status=status)
-        cable.save_base(raw=True)
-
-        for termination, cable_end in (
-            (termination_a, CableEndChoices.SIDE_A),
-            (termination_b, CableEndChoices.SIDE_B),
-        ):
-            ct = CableTermination(cable=cable, cable_end=cable_end, termination=termination)
-            ct.cache_related_objects()
-            ct.save_base(raw=True)
-            termination.cable = cable
-            termination.cable_end = cable_end
-            termination.save()
-
-        return cable
-
-    def test_retrace_after_raw_create(self):
+    def test_304_retrace_cable_created_without_save(self):
         interface1 = Interface.objects.create(device=self.device, name='Interface 1')
         interface2 = Interface.objects.create(device=self.device, name='Interface 2')
 
@@ -3074,7 +2926,7 @@ class CableDependentObjectsTestCase(BaseCablePathTestCase):
         self.assertPathExists((interface2, cable, interface1), is_complete=True, is_active=True)
         self.assertEqual(CablePath.objects.count(), 2)
 
-    def test_retrace_extends_path_via_pass_through(self):
+    def test_305_retrace_cable_extends_incomplete_path(self):
         """
         [IF1] --C1-- [FP1] [RP1] --C2-- [IF2], with C2 written raw. Retracing from a termination which is not
         itself a path endpoint must extend the existing incomplete path.
@@ -3112,7 +2964,7 @@ class CableDependentObjectsTestCase(BaseCablePathTestCase):
         )
         self.assertEqual(CablePath.objects.count(), 2)
 
-    def test_retrace_takes_status_from_database(self):
+    def test_306_retrace_cable_status_from_database(self):
         """
         A raw write leaves no in-memory record of the Cable's status, so path activity must come from the
         stored value.
@@ -3127,7 +2979,7 @@ class CableDependentObjectsTestCase(BaseCablePathTestCase):
         self.assertPathExists((interface2, cable, interface1), is_complete=True, is_active=False)
         self.assertEqual(CablePath.objects.count(), 2)
 
-    def test_retrace_preserves_path_through_pass_through(self):
+    def test_307_retrace_cable_preserves_path_via_pass_through(self):
         """
         [IF1] --C1-- [FP1] [RP1] --C2-- [IF2]. Retracing C1, whose B side is not a path endpoint, must
         preserve the reverse path originating at IF2.
@@ -3163,7 +3015,7 @@ class CableDependentObjectsTestCase(BaseCablePathTestCase):
         )
         self.assertEqual(CablePath.objects.count(), 2)
 
-    def test_retrace_preserves_paths_of_mid_span_cable(self):
+    def test_308_retrace_midspan_cable_preserves_paths(self):
         """
         [IF1] --C1-- [FP1] [RP1] --C2-- [RP2] [FP2] --C3-- [IF2]. Retracing C2, which originates nothing
         itself (both sides are rear ports), must preserve both paths.
@@ -3212,7 +3064,7 @@ class CableDependentObjectsTestCase(BaseCablePathTestCase):
         )
         self.assertEqual(CablePath.objects.count(), 2)
 
-    def test_retrace_preserves_path_through_circuit(self):
+    def test_309_retrace_cable_preserves_path_via_circuit(self):
         """
         [IF1] --C1-- [CT1] [CT2] --C2-- [IF2]. Retracing C1, whose B side is a circuit termination, must
         preserve the reverse path originating at IF2.
@@ -3245,7 +3097,7 @@ class CableDependentObjectsTestCase(BaseCablePathTestCase):
         )
         self.assertEqual(CablePath.objects.count(), 2)
 
-    def test_retrace_is_idempotent(self):
+    def test_310_retrace_cable_is_idempotent(self):
         interface1 = Interface.objects.create(device=self.device, name='Interface 1')
         interface2 = Interface.objects.create(device=self.device, name='Interface 2')
 
@@ -3262,3 +3114,145 @@ class CableDependentObjectsTestCase(BaseCablePathTestCase):
         interface2.refresh_from_db()
         self.assertPathIsSet(interface1, path1)
         self.assertPathIsSet(interface2, path2)
+
+    def test_401_exclude_midspan_devices(self):
+        """
+        [IF1] --C1-- [FP1][Test Device][RP1] --C2-- [RP2][Test Device][FP2] --C3-- [IF2]
+                     [FP3][Test mid-span Device][RP3] --C4-- [RP4][Test mid-span Device][FP4] /
+        """
+        device = Device.objects.create(
+            site=self.site,
+            device_type=self.device.device_type,
+            role=self.device.role,
+            name='Test mid-span Device'
+        )
+        interface1 = Interface.objects.create(device=self.device, name='Interface 1')
+        interface2 = Interface.objects.create(device=self.device, name='Interface 2')
+        rearport1 = RearPort.objects.create(device=self.device, name='Rear Port 1')
+        rearport2 = RearPort.objects.create(device=self.device, name='Rear Port 2')
+        rearport3 = RearPort.objects.create(device=device, name='Rear Port 3')
+        rearport4 = RearPort.objects.create(device=device, name='Rear Port 4')
+        frontport1 = FrontPort.objects.create(device=self.device, name='Front Port 1')
+        frontport2 = FrontPort.objects.create(device=self.device, name='Front Port 2')
+        frontport3 = FrontPort.objects.create(device=self.device, name='Front Port 3')
+        frontport4 = FrontPort.objects.create(device=self.device, name='Front Port 4')
+        PortMapping.objects.bulk_create([
+            PortMapping(
+                device=self.device,
+                front_port=frontport1,
+                front_port_position=1,
+                rear_port=rearport1,
+                rear_port_position=1,
+            ),
+            PortMapping(
+                device=self.device,
+                front_port=frontport2,
+                front_port_position=1,
+                rear_port=rearport2,
+                rear_port_position=1,
+            ),
+            PortMapping(
+                device=self.device,
+                front_port=frontport3,
+                front_port_position=1,
+                rear_port=rearport3,
+                rear_port_position=1,
+            ),
+            PortMapping(
+                device=self.device,
+                front_port=frontport4,
+                front_port_position=1,
+                rear_port=rearport4,
+                rear_port_position=1,
+            ),
+        ])
+
+        cable2 = Cable(
+            a_terminations=[rearport1],
+            b_terminations=[rearport2],
+            label='C2'
+        )
+        cable2.save()
+        cable4 = Cable(
+            a_terminations=[rearport3],
+            b_terminations=[rearport4],
+            label='C4'
+        )
+        cable4.save()
+        self.assertEqual(CablePath.objects.count(), 0)
+
+        # Create cable1
+        cable1 = Cable(
+            a_terminations=[interface1],
+            b_terminations=[frontport1, frontport3],
+            label='C1'
+        )
+        with self.assertRaises(AbortRequest):
+            cable1.save()
+
+        self.assertPathDoesNotExist(
+            (
+                interface1, cable1, (frontport1, frontport3), (rearport1, rearport3), (cable2, cable4),
+                (rearport2, rearport4), (frontport2, frontport4)
+            ),
+            is_complete=False
+        )
+        self.assertEqual(CablePath.objects.count(), 0)
+
+        # Create cable 3
+        cable3 = Cable(
+            a_terminations=[frontport2, frontport4],
+            b_terminations=[interface2],
+            label='C3'
+        )
+
+        with self.assertRaises(AbortRequest):
+            cable3.save()
+
+        self.assertPathDoesNotExist(
+            (
+                interface2, cable3, (frontport2, frontport4), (rearport2, rearport4), (cable2, cable4),
+                (rearport1, rearport3), (frontport1, frontport2), cable1, interface1
+            ),
+            is_complete=True,
+            is_active=True
+        )
+        self.assertPathDoesNotExist(
+            (
+                interface1, cable1, (frontport1, frontport3), (rearport1, rearport3), (cable2, cable4),
+                (rearport2, rearport4), (frontport2, frontport4), cable3, interface2
+            ),
+            is_complete=True,
+            is_active=True
+        )
+        self.assertEqual(CablePath.objects.count(), 0)
+
+    def test_402_exclude_circuit_loopback(self):
+        interface = Interface.objects.create(device=self.device, name='Interface 1')
+        circuittermination1 = CircuitTermination.objects.create(
+            circuit=self.circuit,
+            termination=self.site,
+            term_side='A'
+        )
+        circuittermination2 = CircuitTermination.objects.create(
+            circuit=self.circuit,
+            termination=self.site,
+            term_side='Z'
+        )
+
+        # Create cables
+        cable = Cable(
+            a_terminations=[interface],
+            b_terminations=[circuittermination1, circuittermination2]
+        )
+        cable.save()
+
+        path = self.assertPathExists(
+            (interface, cable, (circuittermination1, circuittermination2)),
+            is_active=True,
+            is_complete=False,
+            is_split=True
+        )
+        self.assertEqual(CablePath.objects.count(), 1)
+        interface.refresh_from_db()
+        self.assertPathIsSet(interface, path)