feat: automate patchpanel mappings and cable tenancy

netbox_utilities/__init__.py

@@ -1,12 +1,12 @@
 from netbox.plugins import PluginConfig, get_plugin_config
 
-__version__ = "0.6.1"
+__version__ = "0.7.0"
 
 
 class NetBoxUtilitiesConfig(PluginConfig):
     name = "netbox_utilities"
     verbose_name = "NetBox Utilities"
-    description = "Navigation, tenant utilities, bulk image/module upload, and atomic rack reordering"
+    description = "Navigation, tenant utilities, patchpanel mapping, bulk uploads, and atomic rack reordering"
     version = __version__
     author = "LKE"
     base_url = "utilities"
@@ -24,11 +24,13 @@ class NetBoxUtilitiesConfig(PluginConfig):
 
     def ready(self):
         super().ready()
+        from .patchpanel import install_patchpanel_automation
         from .tenant_scope import install_search_filter
         from .tenant_validation import install_tenant_validation
 
         install_search_filter()
         install_tenant_validation()
+        install_patchpanel_automation()
         if get_plugin_config("netbox_utilities", "reorder_rack_bulk_save_enabled"):
             from .reorder_rack import install_reorder_rack_bulk_save
 

netbox_utilities/migrations/0004_patchpanel_port_mappings.py

@@ -0,0 +1,55 @@
+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),
+    ]

netbox_utilities/patchpanel.py

@@ -0,0 +1,118 @@
+import re
+from contextvars import ContextVar
+from functools import wraps
+
+from django.db import transaction
+from django.db.models.signals import post_save
+
+PATCHPANEL_ROLE_NAME = "Patchpanel"
+
+_component_instantiation_active = ContextVar(
+    "netbox_utilities_patchpanel_component_instantiation_active",
+    default=False,
+)
+
+
+def natural_port_key(port):
+    """Return a stable natural-sort key for a front or rear port."""
+    name = str(getattr(port, "name", ""))
+    parts = tuple(int(part) if part.isdigit() else part.casefold() for part in re.split(r"(\d+)", name))
+    return parts, getattr(port, "pk", 0) or 0
+
+
+def is_patchpanel(device):
+    role = getattr(device, "role", None)
+    return bool(role and str(getattr(role, "name", "")).casefold() == PATCHPANEL_ROLE_NAME.casefold())
+
+
+def pair_ports(front_ports, rear_ports):
+    """Pair naturally sorted front and rear ports one-to-one."""
+    return list(zip(sorted(front_ports, key=natural_port_key), sorted(rear_ports, key=natural_port_key)))
+
+
+def synchronize_patchpanel(device_id):
+    """Enforce front position 1 -> rear position 1 for a Patchpanel device."""
+    if not device_id:
+        return 0
+
+    from dcim.models import Device, FrontPort, PortMapping, RearPort
+
+    with transaction.atomic():
+        device = Device.objects.select_for_update().select_related("role").filter(pk=device_id).first()
+        if device is None or not is_patchpanel(device):
+            return 0
+
+        front_ports = list(FrontPort.objects.filter(device_id=device_id).only("pk", "name"))
+        rear_ports = list(RearPort.objects.filter(device_id=device_id).only("pk", "name"))
+        pairs = pair_ports(front_ports, rear_ports)
+        expected = {(front.pk, rear.pk, 1, 1) for front, rear in pairs}
+        current = set(
+            PortMapping.objects.filter(device_id=device_id).values_list(
+                "front_port_id",
+                "rear_port_id",
+                "front_port_position",
+                "rear_port_position",
+            )
+        )
+        if current == expected:
+            return len(pairs)
+
+        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
+            ]
+        )
+        return len(pairs)
+
+
+def _component_saved(sender, instance, raw=False, **kwargs):
+    if not raw and not _component_instantiation_active.get():
+        synchronize_patchpanel(getattr(instance, "device_id", None))
+
+
+def _wrap_component_container_save(model, device_id_getter):
+    marker = "_netbox_utilities_patchpanel_save"
+    if getattr(model, marker, False):
+        return
+
+    original_save = model.save
+
+    @wraps(original_save)
+    def patchpanel_aware_save(instance, *args, **kwargs):
+        token = _component_instantiation_active.set(True)
+        try:
+            result = original_save(instance, *args, **kwargs)
+        finally:
+            _component_instantiation_active.reset(token)
+        synchronize_patchpanel(device_id_getter(instance))
+        return result
+
+    model.save = patchpanel_aware_save
+    setattr(model, marker, True)
+
+
+def install_patchpanel_automation():
+    from dcim.models import Device, FrontPort, Module, RearPort
+
+    _wrap_component_container_save(Device, lambda device: device.pk)
+    _wrap_component_container_save(Module, lambda module: module.device_id)
+    post_save.connect(
+        _component_saved,
+        sender=FrontPort,
+        dispatch_uid="netbox_utilities.synchronize_patchpanel_front_ports",
+        weak=False,
+    )
+    post_save.connect(
+        _component_saved,
+        sender=RearPort,
+        dispatch_uid="netbox_utilities.synchronize_patchpanel_rear_ports",
+        weak=False,
+    )

netbox_utilities/tenant_autofill.py

@@ -42,25 +42,105 @@ def tenant_id_from_object(obj, depth=0):
 
 
 def _related_object(form, field_name):
+    objects = _related_objects(form, field_name)
+    return objects[0] if objects else None
+
+
+def _related_objects(form, field_name):
     field = form.fields.get(field_name)
     if field is None or not hasattr(field, "queryset"):
-        return None
+        return []
 
-    value = None
+    values = []
     if form.is_bound:
-        value = form.data.get(form.add_prefix(field_name))
-    if not value:
-        value = form.initial.get(field_name)
-    if hasattr(value, "_meta"):
-        return value
-    if isinstance(value, (list, tuple)):
-        value = value[0] if value else None
-    if not value:
-        return None
+        field_name = form.add_prefix(field_name)
+        if hasattr(form.data, "getlist"):
+            values = form.data.getlist(field_name)
+        else:
+            values = form.data.get(field_name)
+    if not values:
+        values = form.initial.get(field_name)
+    if not isinstance(values, (list, tuple, set)):
+        values = [values]
+
+    objects = [value for value in values if hasattr(value, "_meta")]
+    object_ids = [value for value in values if value and not hasattr(value, "_meta")]
+    if not object_ids:
+        return objects
     try:
-        return field.queryset.filter(pk=value).first()
+        objects.extend(field.queryset.filter(pk__in=object_ids))
     except (TypeError, ValueError):
+        pass
+    return objects
+
+
+def _is_cable(instance):
+    meta = getattr(instance, "_meta", None)
+    return bool(
+        meta
+        and getattr(meta, "app_label", None) == "dcim"
+        and getattr(meta, "model_name", None) == "cable"
+    )
+
+
+def _instance_cable_terminations(instance):
+    if not _is_cable(instance):
+        return []
+    terminations = []
+    for name in ("a_terminations", "b_terminations"):
+        try:
+            values = getattr(instance, name, None) or []
+        except (ObjectDoesNotExist, ValueError):
+            continue
+        terminations.extend(values)
+    return terminations
+
+
+def tenant_id_from_cable_terminations(terminations):
+    """Return the common derivable tenant, or None for no/conflicting tenants."""
+    tenant_ids = {tenant_id for obj in terminations if (tenant_id := tenant_id_from_object(obj))}
+    if len(tenant_ids) == 1:
+        return tenant_ids.pop()
+    return None
+
+
+def _cable_form_tenant_result(form):
+    instance = getattr(form, "instance", None)
+    is_cable_form = _is_cable(instance) or any(name in form.fields for name in ("a_terminations", "b_terminations"))
+    if not is_cable_form:
+        return None, False
+
+    terminations = []
+    for field_name in (
+        "a_terminations",
+        "b_terminations",
+        "termination_a_device",
+        "termination_b_device",
+        "termination_a_powerpanel",
+        "termination_b_powerpanel",
+        "termination_a_circuit",
+        "termination_b_circuit",
+    ):
+        terminations.extend(_related_objects(form, field_name))
+    if not terminations:
+        terminations = _instance_cable_terminations(instance)
+
+    tenant_ids = {tenant_id for obj in terminations if (tenant_id := tenant_id_from_object(obj))}
+    if len(tenant_ids) == 1:
+        return tenant_ids.pop(), True
+    if len(tenant_ids) > 1:
+        return None, True
+    return None, False
+
+
+def apply_cable_instance_tenant(instance):
+    """Assign a cable's tenant when all derivable terminations agree."""
+    if not _is_cable(instance) or getattr(instance, "tenant_id", None):
         return None
+    tenant_id = tenant_id_from_cable_terminations(_instance_cable_terminations(instance))
+    if tenant_id:
+        instance.tenant_id = tenant_id
+    return tenant_id
 
 
 def infer_tenant_id(form):
@@ -77,6 +157,10 @@ def infer_tenant_id(form):
         except (TypeError, ValueError):
             pass
 
+    cable_tenant_id, cable_context_found = _cable_form_tenant_result(form)
+    if cable_context_found:
+        return cable_tenant_id
+
     for relation in PARENT_RELATIONS:
         if tenant_id := tenant_id_from_object(_related_object(form, relation)):
             return tenant_id

netbox_utilities/tenant_validation.py

@@ -7,7 +7,7 @@ from django.forms.models import BaseModelForm
 from tenancy.models import Tenant
 
 from .runtime import tenant_required
-from .tenant_autofill import apply_tenant_autofill
+from .tenant_autofill import apply_cable_instance_tenant, apply_tenant_autofill
 
 VALIDATION_MESSAGE = "Für dieses Objekt muss ein Mandant angegeben werden."
 
@@ -36,6 +36,7 @@ def _install_model_clean_validation(model):
 
     @wraps(original_clean)
     def tenant_aware_clean(instance, *args, **kwargs):
+        apply_cable_instance_tenant(instance)
         result = original_clean(instance, *args, **kwargs)
         validate_tenant_assignment(instance)
         return result

netbox_utilities/tests/test_patchpanel.py

@@ -0,0 +1,29 @@
+from types import SimpleNamespace
+
+from django.test import SimpleTestCase
+
+from netbox_utilities.patchpanel import is_patchpanel, pair_ports
+
+
+def port(pk, name):
+    return SimpleNamespace(pk=pk, name=name)
+
+
+class PatchpanelPairingTest(SimpleTestCase):
+    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"))))
+        self.assertFalse(is_patchpanel(SimpleNamespace(role=SimpleNamespace(name="Switch"))))
+
+    def test_pairs_front_and_rear_ports_in_natural_order(self):
+        front_ports = [port(10, "10"), port(2, "2"), port(1, "1")]
+        rear_ports = [port(102, "Rear 2"), port(110, "Rear 10"), port(101, "Rear 1")]
+
+        pairs = pair_ports(front_ports, rear_ports)
+
+        self.assertEqual([(front.pk, rear.pk) for front, rear in pairs], [(1, 101), (2, 102), (10, 110)])
+
+    def test_leaves_unmatched_ports_without_a_mapping(self):
+        pairs = pair_ports([port(1, "1"), port(2, "2")], [port(101, "1")])
+
+        self.assertEqual([(front.pk, rear.pk) for front, rear in pairs], [(1, 101)])

netbox_utilities/tests/test_tenant_autofill.py

@@ -5,7 +5,13 @@ from django import forms
 from django.test import SimpleTestCase
 from tenancy.models import TenantGroup
 
-from netbox_utilities.tenant_autofill import apply_tenant_autofill, infer_tenant_id, tenant_id_from_object
+from netbox_utilities.tenant_autofill import (
+    apply_cable_instance_tenant,
+    apply_tenant_autofill,
+    infer_tenant_id,
+    tenant_id_from_cable_terminations,
+    tenant_id_from_object,
+)
 from netbox_utilities.tenant_scope import ActiveTenantScope, active_tenant_scope
 
 
@@ -17,6 +23,11 @@ class FakeMeta:
         return [SimpleNamespace(name=name) for name in self.field_names]
 
 
+class FakeCableMeta(FakeMeta):
+    app_label = "dcim"
+    model_name = "cable"
+
+
 class FakeObject:
     def __init__(self, *, tenant_id=None, **relations):
         self.tenant_id = tenant_id
@@ -25,6 +36,14 @@ class FakeObject:
             setattr(self, name, value)
 
 
+class FakeCable(FakeObject):
+    def __init__(self, *, tenant_id=None, a_terminations=None, b_terminations=None):
+        super().__init__(tenant_id=tenant_id)
+        self._meta = FakeCableMeta("tenant")
+        self.a_terminations = a_terminations or []
+        self.b_terminations = b_terminations or []
+
+
 class FakeQuerySet:
     def __init__(self, model):
         self.model = model
@@ -50,6 +69,54 @@ class TenantAutofillTest(SimpleTestCase):
         finally:
             active_tenant_scope.reset(token)
 
+    def test_finds_common_cable_tenant_from_both_ends(self):
+        terminations = [FakeObject(device=FakeObject(tenant_id=42)), FakeObject(device=FakeObject(tenant_id=42))]
+
+        self.assertEqual(tenant_id_from_cable_terminations(terminations), 42)
+
+    def test_does_not_guess_when_cable_ends_have_different_tenants(self):
+        terminations = [FakeObject(device=FakeObject(tenant_id=42)), FakeObject(device=FakeObject(tenant_id=43))]
+
+        self.assertIsNone(tenant_id_from_cable_terminations(terminations))
+
+    def test_assigns_common_tenant_to_cable_instance(self):
+        cable = FakeCable(
+            a_terminations=[FakeObject(device=FakeObject(tenant_id=42))],
+            b_terminations=[FakeObject(device=FakeObject(tenant_id=42))],
+        )
+
+        self.assertEqual(apply_cable_instance_tenant(cable), 42)
+        self.assertEqual(cable.tenant_id, 42)
+
+    def test_prefers_cable_terminations_over_global_scope(self):
+        form = SimpleNamespace(
+            fields={"tenant": forms.IntegerField()},
+            initial={},
+            instance=FakeCable(a_terminations=[FakeObject(device=FakeObject(tenant_id=42))]),
+            is_bound=False,
+        )
+        token = active_tenant_scope.set(ActiveTenantScope("tenant", 17, frozenset({17})))
+        try:
+            self.assertEqual(infer_tenant_id(form), 42)
+        finally:
+            active_tenant_scope.reset(token)
+
+    def test_conflicting_cable_ends_do_not_fall_back_to_global_scope(self):
+        form = SimpleNamespace(
+            fields={"tenant": forms.IntegerField()},
+            initial={},
+            instance=FakeCable(
+                a_terminations=[FakeObject(device=FakeObject(tenant_id=42))],
+                b_terminations=[FakeObject(device=FakeObject(tenant_id=43))],
+            ),
+            is_bound=False,
+        )
+        token = active_tenant_scope.set(ActiveTenantScope("tenant", 17, frozenset({17})))
+        try:
+            self.assertIsNone(infer_tenant_id(form))
+        finally:
+            active_tenant_scope.reset(token)
+
     @patch("netbox_utilities.tenant_autofill._infer_group_id", return_value=None)
     @patch("netbox_utilities.tenant_autofill.infer_tenant_id", return_value=23)
     def test_prefills_and_marks_tenant_field(self, _infer_tenant_id, _infer_group_id):