From 0c4a737d41c41fe19b94c1fc70ef34470ba0d20d Mon Sep 17 00:00:00 2001 From: Louis Date: Thu, 6 Aug 2026 15:09:08 +0200 Subject: [PATCH] feat: automate patchpanel mappings and cable tenancy --- README.md | 33 ++++- netbox_utilities/__init__.py | 6 +- .../0004_patchpanel_port_mappings.py | 55 ++++++++ netbox_utilities/patchpanel.py | 118 ++++++++++++++++++ netbox_utilities/tenant_autofill.py | 108 ++++++++++++++-- netbox_utilities/tenant_validation.py | 3 +- netbox_utilities/tests/test_patchpanel.py | 29 +++++ .../tests/test_tenant_autofill.py | 69 +++++++++- pyproject.toml | 4 +- 9 files changed, 405 insertions(+), 20 deletions(-) create mode 100644 netbox_utilities/migrations/0004_patchpanel_port_mappings.py create mode 100644 netbox_utilities/patchpanel.py create mode 100644 netbox_utilities/tests/test_patchpanel.py diff --git a/README.md b/README.md index 97acf9a..ba53bc5 100644 --- a/README.md +++ b/README.md @@ -1,11 +1,12 @@ # NetBox Utilities -Plugin für **NetBox 4.6.5** mit sieben Funktionen: +Plugin für **NetBox 4.6.5** mit acht Funktionen: - Jeder Benutzer kann die Menüs der linken Navigation verschieben oder ausblenden. - Ein Dropdown in der Kopfleiste setzt einen sitzungsweiten Filter für einen Mandanten oder eine Mandantengruppe. - Optional verpflichtende Mandantenzuordnung für alle mandantenfähigen Objekte. - Automatische Vorbelegung von Mandant und Mandantengruppe aus dem Objekt- oder Filterkontext. +- Automatische 1:1-Verknüpfung von Front- und Rearports auf Geräten mit der Rolle `Patchpanel`. - Mehrere Bilder in einem Schritt im Bilder-Tab eines Objekts hochladen. - Mehrere Module desselben Typs in einem Schritt in freie Modulschächte einbauen. - Optionale Mehrfachspeicherung für verschobene Geräte aus NetBox Reorder Rack. @@ -34,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.6.1" + "git+https://git.mrblake.cc/MrBlake/Netbox-Utilities.git@v0.7.0" ``` Alternativ kann hinter dem `@` die vollständige Commit-ID stehen. @@ -131,6 +132,27 @@ empfohlen. ## Verwendung +### Front- und Rearports von Patchpaneln automatisch verknüpfen + +Geräte, deren NetBox-Geräterolle `Patchpanel` heißt, werden automatisch als +Patchfeld erkannt. Groß- und Kleinschreibung spielen bei der Rollenbezeichnung +keine Rolle. Das Plugin sortiert Front- und Rearports natürlich, sodass zum +Beispiel `1, 2, 10` in dieser Reihenfolge verarbeitet wird, und verknüpft sie +anschließend 1:1 auf Portposition 1. Der erste Frontport wird mit dem ersten +Rearport verbunden, der zweite mit dem zweiten und so weiter. Überzählige Ports +bleiben unverknüpft. + +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. + +Beim Update auf Version `0.7.0` führt die Plugin-Migration dieselbe Zuordnung +einmalig für alle bereits vorhandenen Geräte mit dieser Rolle aus. Vor dem +Produktivupdate sollte daher geprüft werden, ob Geräte mit bewusst +abweichenden oder mehrpoligen Mappings nicht die Rolle `Patchpanel` tragen +sollen. + ### Mehrere Geräte mit NetBox Reorder Rack verschieben Wenn `netbox-reorder-rack` in Version `1.1.4` installiert und aktiviert ist, @@ -270,6 +292,13 @@ erkennbar, wird der global ausgewählte Mandant verwendet. Felder für eine Mandantengruppe werden aus dem erkannten Mandanten oder der global ausgewählten Mandantengruppe vorbelegt. Bereits vorhandene Werte werden nicht überschrieben. +Bei Kabeln werden die ausgewählten Anschlüsse beider Kabelenden ausgewertet. +Gehören die Anschlüsse beziehungsweise ihre Geräte, Stromverteiler oder +Schaltkreise eindeutig zu demselben Mandanten, übernimmt das Kabel diesen +Mandanten automatisch. Ist nur an einem Ende eine Zuordnung herleitbar, wird +diese verwendet. Bei widersprüchlichen Mandanten an den beiden Enden nimmt das +Plugin keine automatische Zuordnung vor. + Bleibt ein automatisch gesetzter Mandant bis zum Speichern unverändert, fragt der Browser unmittelbar nach dem Klick auf **Speichern** noch einmal nach einer ausdrücklichen Bestätigung. Wird der Mandant manuell geändert, entfällt diese diff --git a/netbox_utilities/__init__.py b/netbox_utilities/__init__.py index 3c8002c..97b230d 100644 --- a/netbox_utilities/__init__.py +++ b/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 diff --git a/netbox_utilities/migrations/0004_patchpanel_port_mappings.py b/netbox_utilities/migrations/0004_patchpanel_port_mappings.py new file mode 100644 index 0000000..3e681df --- /dev/null +++ b/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), + ] diff --git a/netbox_utilities/patchpanel.py b/netbox_utilities/patchpanel.py new file mode 100644 index 0000000..9225902 --- /dev/null +++ b/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, + ) diff --git a/netbox_utilities/tenant_autofill.py b/netbox_utilities/tenant_autofill.py index f66a85e..a7c785d 100644 --- a/netbox_utilities/tenant_autofill.py +++ b/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 diff --git a/netbox_utilities/tenant_validation.py b/netbox_utilities/tenant_validation.py index 70d21ff..9ad1109 100644 --- a/netbox_utilities/tenant_validation.py +++ b/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 diff --git a/netbox_utilities/tests/test_patchpanel.py b/netbox_utilities/tests/test_patchpanel.py new file mode 100644 index 0000000..61efd17 --- /dev/null +++ b/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)]) diff --git a/netbox_utilities/tests/test_tenant_autofill.py b/netbox_utilities/tests/test_tenant_autofill.py index eb4d985..c436a41 100644 --- a/netbox_utilities/tests/test_tenant_autofill.py +++ b/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): diff --git a/pyproject.toml b/pyproject.toml index cbcaac6..ce87bf5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,8 +4,8 @@ build-backend = "hatchling.build" [project] name = "netbox-utilities" -version = "0.6.1" -description = "Navigation, tenant utilities, bulk image/module upload, and atomic rack reordering for NetBox 4.6" +version = "0.7.0" +description = "Navigation, tenant utilities, patchpanel mapping, bulk uploads, and atomic rack reordering for NetBox 4.6" readme = "README.md" requires-python = ">=3.12" license = { text = "MIT" }