fix: skip patchpanel automation on edits

README.md

@@ -35,7 +35,7 @@ Release-Tag oder ein bestimmter Commit verwendet werden:
 
 ```bash
 /opt/netbox/venv/bin/pip install --upgrade --force-reinstall \
-  "git+https://git.mrblake.cc/MrBlake/Netbox-Utilities.git@v0.7.2"
+  "git+https://git.mrblake.cc/MrBlake/Netbox-Utilities.git@v0.7.4"
 ```
 
 Alternativ kann hinter dem `@` die vollständige Commit-ID stehen.
@@ -142,25 +142,15 @@ weiter. Auch Kurzformen wie `F01` und `R1` werden als dieselbe Kennung erkannt.
 Fehlt die passende Gegenseite, bleibt der Port unverknüpft; die nachfolgenden
 Nummern rutschen nicht auf.
 
-Die Zuordnung wird nach Änderungen am Gerät, nach dem Einbau eines Moduls und
-nach dem Anlegen oder Umbenennen einzelner Front- beziehungsweise Rearports
-erneut hergestellt. Dabei ersetzt die Patchpanel-Automatik abweichende
-vorhandene Port-Mappings auf diesen Geräten durch die feste 1:1-Zuordnung.
+Die Zuordnung wird ausschließlich beim erstmaligen Anlegen eines Patchpanels,
+beim Einbau eines neuen Moduls sowie beim Anlegen eines neuen Front- oder
+Rearports hergestellt. Das Bearbeiten eines vorhandenen Patchpanels, Moduls
+oder Ports löst die Automatik nicht aus.
 
-Beim Update auf Version `0.7.2` korrigiert eine Plugin-Migration die Zuordnung
-automatisch für alle bereits vorhandenen Geräte mit dieser Rolle und berechnet
-die betroffenen Kabelpfade beziehungsweise Verbindungsenden neu. Vor dem
-Produktivupdate sollte daher geprüft werden, ob Geräte mit bewusst
-abweichenden oder mehrpoligen Mappings nicht die Rolle `Patchpanel` tragen
-sollen.
-
-Der Reparaturlauf kann bei Bedarf jederzeit erneut und ohne erneute Migration
-gestartet werden:
-
-```bash
-cd /opt/netbox/netbox
-/opt/netbox/venv/bin/python manage.py repair_patchpanel_mappings
-```
+Bei Installation oder Update findet kein automatischer Bestandslauf statt.
+Vorhandene Patchpanel-Geräte und deren Kabelpfade bleiben unverändert. Die
+Automatik greift erst, wenn ein betreffendes Gerät, Modul oder ein Front-/Rearport
+neu angelegt wird.
 
 ### Mehrere Geräte mit NetBox Reorder Rack verschieben
 

netbox_utilities/__init__.py

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

netbox_utilities/management/commands/repair_patchpanel_mappings.py

@@ -1,15 +0,0 @@
-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/0004_patchpanel_port_mappings.py

@@ -1,55 +1,12 @@
-import re
-
 from django.db import migrations
 
 
-def natural_port_key(port):
-    parts = tuple(
-        int(part) if part.isdigit() else part.casefold()
-        for part in re.split(r"(\d+)", str(port.name))
-    )
-    return parts, port.pk
-
-
-def synchronize_existing_patchpanels(apps, schema_editor):
-    Device = apps.get_model("dcim", "Device")
-    FrontPort = apps.get_model("dcim", "FrontPort")
-    PortMapping = apps.get_model("dcim", "PortMapping")
-    RearPort = apps.get_model("dcim", "RearPort")
-    database = schema_editor.connection.alias
-
-    device_ids = Device.objects.using(database).filter(role__name__iexact="Patchpanel").values_list("pk", flat=True)
-    for device_id in device_ids.iterator():
-        front_ports = sorted(
-            FrontPort.objects.using(database).filter(device_id=device_id).only("pk", "name"),
-            key=natural_port_key,
-        )
-        rear_ports = sorted(
-            RearPort.objects.using(database).filter(device_id=device_id).only("pk", "name"),
-            key=natural_port_key,
-        )
-        PortMapping.objects.using(database).filter(device_id=device_id).delete()
-        PortMapping.objects.using(database).bulk_create(
-            [
-                PortMapping(
-                    device_id=device_id,
-                    front_port_id=front_port.pk,
-                    rear_port_id=rear_port.pk,
-                    front_port_position=1,
-                    rear_port_position=1,
-                )
-                for front_port, rear_port in zip(front_ports, rear_ports)
-            ],
-            batch_size=1000,
-        )
-
-
 class Migration(migrations.Migration):
     dependencies = [
         ("dcim", "0237_module_remove_local_context_data"),
         ("netbox_utilities", "0003_navigationpreference_sidebar_layout"),
     ]
 
-    operations = [
-        migrations.RunPython(synchronize_existing_patchpanels, migrations.RunPython.noop),
-    ]
+    # Kept in the migration graph for installations which already recorded it.
+    # Existing devices must not be modified automatically.
+    operations = []

netbox_utilities/migrations/0005_fix_patchpanel_port_number_mappings.py

@@ -1,81 +1,11 @@
-import re
-from collections import defaultdict
-
 from django.db import migrations
 
 
-def natural_port_key(port):
-    parts = tuple(
-        int(part) if part.isdigit() else part.casefold()
-        for part in re.split(r"(\d+)", str(port.name))
-    )
-    return parts, port.pk
-
-
-def port_identifier(port):
-    name = str(port.name).casefold()
-    name = re.sub(r"\b(?:front|rear)(?:[\s_-]*port)?(?=\b|\d)", " ", name)
-    name = re.sub(r"^(?:fp|rp|f|r)(?=\s*[-_.:/]?\s*\d)", "", name)
-    tokens = re.findall(r"\d+|[^\W\d_]+", name)
-    identifier = tuple(int(token) if token.isdigit() else token for token in tokens)
-    return identifier or None
-
-
-def pair_ports(front_ports, rear_ports):
-    front_by_identifier = defaultdict(list)
-    rear_by_identifier = defaultdict(list)
-    for port in front_ports:
-        if identifier := port_identifier(port):
-            front_by_identifier[identifier].append(port)
-    for port in rear_ports:
-        if identifier := port_identifier(port):
-            rear_by_identifier[identifier].append(port)
-
-    pairs = []
-    for identifier, matching_front_ports in front_by_identifier.items():
-        pairs.extend(
-            zip(
-                sorted(matching_front_ports, key=natural_port_key),
-                sorted(rear_by_identifier.get(identifier, []), key=natural_port_key),
-            )
-        )
-    return pairs
-
-
-def fix_existing_patchpanel_mappings(apps, schema_editor):
-    Device = apps.get_model("dcim", "Device")
-    FrontPort = apps.get_model("dcim", "FrontPort")
-    PortMapping = apps.get_model("dcim", "PortMapping")
-    RearPort = apps.get_model("dcim", "RearPort")
-    database = schema_editor.connection.alias
-
-    device_ids = Device.objects.using(database).filter(role__name__iexact="Patchpanel").values_list("pk", flat=True)
-    for device_id in device_ids.iterator():
-        front_ports = FrontPort.objects.using(database).filter(device_id=device_id).only("pk", "name")
-        rear_ports = RearPort.objects.using(database).filter(device_id=device_id).only("pk", "name")
-        pairs = pair_ports(front_ports, rear_ports)
-
-        PortMapping.objects.using(database).filter(device_id=device_id).delete()
-        PortMapping.objects.using(database).bulk_create(
-            [
-                PortMapping(
-                    device_id=device_id,
-                    front_port_id=front_port.pk,
-                    rear_port_id=rear_port.pk,
-                    front_port_position=1,
-                    rear_port_position=1,
-                )
-                for front_port, rear_port in pairs
-            ],
-            batch_size=1000,
-        )
-
-
 class Migration(migrations.Migration):
     dependencies = [
         ("netbox_utilities", "0004_patchpanel_port_mappings"),
     ]
 
-    operations = [
-        migrations.RunPython(fix_existing_patchpanel_mappings, migrations.RunPython.noop),
-    ]
+    # Kept in the migration graph for installations which already recorded it.
+    # Existing devices must not be modified automatically.
+    operations = []

netbox_utilities/migrations/0006_retrace_existing_patchpanel_paths.py

@@ -1,19 +1,11 @@
 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),
-    ]
+    # Kept in the migration graph for installations which already recorded it.
+    # Existing devices and cable paths must not be modified automatically.
+    operations = []

netbox_utilities/patchpanel.py

@@ -75,7 +75,7 @@ def retrace_patchpanel_paths(ports):
     return retraced
 
 
-def synchronize_patchpanel(device_id, *, force_retrace=False):
+def synchronize_patchpanel(device_id):
     """Enforce front position 1 -> rear position 1 for a Patchpanel device."""
     if not device_id:
         return 0
@@ -113,28 +113,12 @@ def synchronize_patchpanel(device_id, *, force_retrace=False):
                     for front, rear in pairs
                 ]
             )
-            force_retrace = True
-
-        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():
+def _component_saved(sender, instance, created=False, raw=False, **kwargs):
+    if created and not raw and not _component_instantiation_active.get():
         synchronize_patchpanel(getattr(instance, "device_id", None))
 
 
@@ -147,6 +131,10 @@ def _wrap_component_container_save(model, device_id_getter):
 
     @wraps(original_save)
     def patchpanel_aware_save(instance, *args, **kwargs):
+        created = instance._state.adding
+        if not created:
+            return original_save(instance, *args, **kwargs)
+
         token = _component_instantiation_active.set(True)
         try:
             result = original_save(instance, *args, **kwargs)

netbox_utilities/tests/test_patchpanel.py

@@ -1,9 +1,17 @@
+from importlib import import_module
 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, retrace_patchpanel_paths
+from netbox_utilities.patchpanel import (
+    _component_saved,
+    _wrap_component_container_save,
+    is_patchpanel,
+    pair_ports,
+    port_identifier,
+    retrace_patchpanel_paths,
+)
 
 
 def port(pk, name):
@@ -11,6 +19,53 @@ def port(pk, name):
 
 
 class PatchpanelPairingTest(SimpleTestCase):
+    @patch("netbox_utilities.patchpanel.synchronize_patchpanel")
+    def test_component_edits_do_not_trigger_automation(self, synchronize):
+        component = SimpleNamespace(device_id=42)
+
+        _component_saved(None, component, created=False)
+
+        synchronize.assert_not_called()
+
+    @patch("netbox_utilities.patchpanel.synchronize_patchpanel")
+    def test_new_components_trigger_automation(self, synchronize):
+        component = SimpleNamespace(device_id=42)
+
+        _component_saved(None, component, created=True)
+
+        synchronize.assert_called_once_with(42)
+
+    @patch("netbox_utilities.patchpanel.synchronize_patchpanel")
+    def test_existing_device_save_does_not_trigger_automation(self, synchronize):
+        class Container:
+            def __init__(self, adding):
+                self._state = SimpleNamespace(adding=adding)
+                self.pk = 42
+
+            def save(self):
+                self._state.adding = False
+                return "saved"
+
+        _wrap_component_container_save(Container, lambda instance: instance.pk)
+
+        self.assertEqual(Container(adding=False).save(), "saved")
+        synchronize.assert_not_called()
+
+        self.assertEqual(Container(adding=True).save(), "saved")
+        synchronize.assert_called_once_with(42)
+
+    def test_existing_device_migrations_are_noops(self):
+        migration_names = (
+            "0004_patchpanel_port_mappings",
+            "0005_fix_patchpanel_port_number_mappings",
+            "0006_retrace_existing_patchpanel_paths",
+        )
+
+        for migration_name in migration_names:
+            with self.subTest(migration=migration_name):
+                module = import_module(f"netbox_utilities.migrations.{migration_name}")
+                self.assertEqual(module.Migration.operations, [])
+
     def test_detects_patchpanel_role_case_insensitively(self):
         self.assertTrue(is_patchpanel(SimpleNamespace(role=SimpleNamespace(name="Patchpanel"))))
         self.assertTrue(is_patchpanel(SimpleNamespace(role=SimpleNamespace(name="PATCHPANEL"))))

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "netbox-utilities"
-version = "0.7.2"
+version = "0.7.4"
 description = "Navigation, tenant utilities, patchpanel mapping, bulk uploads, and atomic rack reordering for NetBox 4.6"
 readme = "README.md"
 requires-python = ">=3.12"