fix: retrace existing patchpanel cable paths

netbox_utilities/__init__.py

@@ -1,6 +1,6 @@
 from netbox.plugins import PluginConfig, get_plugin_config
 
-__version__ = "0.7.1"
+__version__ = "0.7.2"
 
 
 class NetBoxUtilitiesConfig(PluginConfig):

netbox_utilities/management/commands/repair_patchpanel_mappings.py

@@ -0,0 +1,15 @@
+from django.core.management.base import BaseCommand
+
+from netbox_utilities.patchpanel import repair_existing_patchpanels
+
+
+class Command(BaseCommand):
+    help = "Repair front/rear mappings and cable paths for all devices with the Patchpanel role"
+
+    def handle(self, *args, **options):
+        device_count, mapping_count = repair_existing_patchpanels()
+        self.stdout.write(
+            self.style.SUCCESS(
+                f"Repaired {mapping_count} front/rear mappings on {device_count} Patchpanel devices."
+            )
+        )

netbox_utilities/migrations/0006_retrace_existing_patchpanel_paths.py

@@ -0,0 +1,19 @@
+from django.db import migrations
+
+
+def repair_mappings_and_paths(apps, schema_editor):
+    # Runtime models are intentional here: NetBox's CablePath.retrace() is
+    # required to refresh the denormalized path data after migration 0005.
+    from netbox_utilities.patchpanel import repair_existing_patchpanels
+
+    repair_existing_patchpanels()
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("netbox_utilities", "0005_fix_patchpanel_port_number_mappings"),
+    ]
+
+    operations = [
+        migrations.RunPython(repair_mappings_and_paths, migrations.RunPython.noop),
+    ]

netbox_utilities/patchpanel.py

@@ -59,7 +59,23 @@ def pair_ports(front_ports, rear_ports):
     return sorted(pairs, key=lambda pair: natural_port_key(pair[0]))
 
 
-def synchronize_patchpanel(device_id):
+def retrace_patchpanel_paths(ports):
+    """Retrace cable paths touching any of the supplied pass-through ports."""
+    from dcim.models import CablePath
+
+    path_ids = set()
+    for port in ports:
+        path_ids.update(CablePath.objects.filter(_nodes__contains=port).values_list("pk", flat=True))
+
+    retraced = 0
+    for path_id in path_ids:
+        if cable_path := CablePath.objects.filter(pk=path_id).first():
+            cable_path.retrace()
+            retraced += 1
+    return retraced
+
+
+def synchronize_patchpanel(device_id, *, force_retrace=False):
     """Enforce front position 1 -> rear position 1 for a Patchpanel device."""
     if not device_id:
         return 0
@@ -83,25 +99,40 @@ def synchronize_patchpanel(device_id):
                 "rear_port_position",
             )
         )
-        if current == expected:
-            return len(pairs)
+        if current != expected:
+            PortMapping.objects.filter(device_id=device_id).delete()
+            PortMapping.objects.bulk_create(
+                [
+                    PortMapping(
+                        device_id=device_id,
+                        front_port_id=front.pk,
+                        rear_port_id=rear.pk,
+                        front_port_position=1,
+                        rear_port_position=1,
+                    )
+                    for front, rear in pairs
+                ]
+            )
+            force_retrace = True
 
-        PortMapping.objects.filter(device_id=device_id).delete()
-        PortMapping.objects.bulk_create(
-            [
-                PortMapping(
-                    device_id=device_id,
-                    front_port_id=front.pk,
-                    rear_port_id=rear.pk,
-                    front_port_position=1,
-                    rear_port_position=1,
-                )
-                for front, rear in pairs
-            ]
-        )
+        if force_retrace:
+            retrace_patchpanel_paths([*front_ports, *rear_ports])
         return len(pairs)
 
 
+def repair_existing_patchpanels():
+    """Repair mappings and cable paths for all existing Patchpanel devices."""
+    from dcim.models import Device
+
+    device_ids = Device.objects.filter(role__name__iexact=PATCHPANEL_ROLE_NAME).values_list("pk", flat=True)
+    device_count = 0
+    mapping_count = 0
+    for device_id in device_ids:
+        mapping_count += synchronize_patchpanel(device_id, force_retrace=True)
+        device_count += 1
+    return device_count, mapping_count
+
+
 def _component_saved(sender, instance, raw=False, **kwargs):
     if not raw and not _component_instantiation_active.get():
         synchronize_patchpanel(getattr(instance, "device_id", None))

netbox_utilities/tests/test_patchpanel.py

@@ -1,8 +1,9 @@
 from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
 
 from django.test import SimpleTestCase
 
-from netbox_utilities.patchpanel import is_patchpanel, pair_ports, port_identifier
+from netbox_utilities.patchpanel import is_patchpanel, pair_ports, port_identifier, retrace_patchpanel_paths
 
 
 def port(pk, name):
@@ -37,3 +38,25 @@ class PatchpanelPairingTest(SimpleTestCase):
         pairs = pair_ports([port(1, "Front 1")], [port(102, "Rear 2")])
 
         self.assertEqual(pairs, [])
+
+    @patch("dcim.models.CablePath")
+    def test_retraces_each_affected_cable_path_once(self, cable_path_model):
+        first_path = MagicMock()
+        second_path = MagicMock()
+
+        def filter_paths(**kwargs):
+            queryset = MagicMock()
+            if "_nodes__contains" in kwargs:
+                port_id = kwargs["_nodes__contains"].pk
+                queryset.values_list.return_value = [1] if port_id == 1 else [1, 2]
+            else:
+                queryset.first.return_value = {1: first_path, 2: second_path}.get(kwargs["pk"])
+            return queryset
+
+        cable_path_model.objects.filter.side_effect = filter_paths
+
+        retraced = retrace_patchpanel_paths([port(1, "Front 1"), port(2, "Front 2")])
+
+        self.assertEqual(retraced, 2)
+        first_path.retrace.assert_called_once_with()
+        second_path.retrace.assert_called_once_with()