Просмотр исходного кода

fix(dcim): Resolve link peers for cable profile connectors

Add `_get_profile_link_peers()` method to handle connector-to-connector
mappings when cables use profiles. Includes regression test for
TRUNK_4C1P profile ensuring correct peer resolution between interfaces
and rear ports.

Fixes #21917
Martin Hauser 1 месяц назад
Родитель
Сommit
aa14e1d322
2 измененных файлов с 99 добавлено и 7 удалено
  1. 31 7
      netbox/dcim/models/device_components.py
  2. 68 0
      netbox/dcim/tests/test_cable_profiles.py

+ 31 - 7
netbox/dcim/models/device_components.py

@@ -254,13 +254,37 @@ class CabledObjectModel(models.Model):
 
 
     @cached_property
     @cached_property
     def link_peers(self):
     def link_peers(self):
-        if self.cable:
-            return [
-                peer.termination
-                for peer in self.cable.terminations.all()
-                if peer.cable_end != self.cable_end
-            ]
-        return []
+        if not self.cable:
+            return []
+
+        if self.cable.profile:
+            return self._get_profile_link_peers()
+
+        return [peer.termination for peer in self.cable.terminations.all() if peer.cable_end != self.cable_end]
+
+    def _get_profile_link_peers(self):
+        if self.cable_end is None or self.cable_connector is None or not self.cable_positions:
+            return []
+
+        profile = self.cable.profile_class()
+        peer_terminations = {
+            (peer.connector, position): peer.termination
+            for peer in self.cable.terminations.all()
+            if peer.cable_end == self.opposite_cable_end and peer.connector is not None
+            for position in peer.positions or []
+        }
+        link_peers = []
+
+        for position in self.cable_positions:
+            mapped_position = profile.get_mapped_position(self.cable_end, self.cable_connector, position)
+            if mapped_position is None:
+                continue
+
+            peer = peer_terminations.get(mapped_position)
+            if peer is not None and peer not in link_peers:
+                link_peers.append(peer)
+
+        return link_peers
 
 
     @property
     @property
     def _occupied(self):
     def _occupied(self):

+ 68 - 0
netbox/dcim/tests/test_cable_profiles.py

@@ -0,0 +1,68 @@
+from django.test import tag
+
+from dcim.choices import CableProfileChoices
+from dcim.models import Cable, Interface, RearPort
+from dcim.tests.utils import CablePathTestCase
+
+
+class CableProfileLinkPeerTests(CablePathTestCase):
+    """
+    Tests for link peer resolution with cable profiles.
+    """
+
+    @tag('regression')  # #21917
+    def test_trunk_4c1p_link_peers(self):
+        """
+        Link peers for trunk profile cables should honor connector mappings.
+        """
+        interfaces = [Interface.objects.create(device=self.device, name=f'Interface {i}') for i in range(1, 5)]
+        rear_ports = [
+            RearPort.objects.create(device=self.device, name=f'Rear Port {i}', positions=1) for i in range(1, 5)
+        ]
+
+        cable = Cable(
+            profile=CableProfileChoices.TRUNK_4C1P,
+            a_terminations=interfaces,
+            b_terminations=rear_ports,
+        )
+        cable.clean()
+        cable.save()
+
+        for interface, rear_port in zip(interfaces, rear_ports):
+            interface.refresh_from_db()
+            rear_port.refresh_from_db()
+
+            self.assertEqual(interface.link_peers, [rear_port])
+            self.assertEqual(rear_port.link_peers, [interface])
+
+    @tag('regression')  # #21917
+    def test_breakout_shuffle_link_peers(self):
+        """
+        Link peers for asymmetric breakout profiles should honor mapped connectors.
+        """
+        rear_ports = [
+            RearPort.objects.create(device=self.device, name=f'Rear Port {i}', positions=4) for i in range(1, 3)
+        ]
+        interfaces = [Interface.objects.create(device=self.device, name=f'Interface {i}') for i in range(1, 9)]
+
+        cable = Cable(
+            profile=CableProfileChoices.BREAKOUT_2C4P_8C1P_SHUFFLE,
+            a_terminations=rear_ports,
+            b_terminations=interfaces,
+        )
+        cable.clean()
+        cable.save()
+
+        for rear_port in rear_ports:
+            rear_port.refresh_from_db()
+        for interface in interfaces:
+            interface.refresh_from_db()
+
+        self.assertEqual(rear_ports[0].link_peers, [interfaces[0], interfaces[1], interfaces[4], interfaces[5]])
+        self.assertEqual(rear_ports[1].link_peers, [interfaces[2], interfaces[3], interfaces[6], interfaces[7]])
+
+        for interface in interfaces[0:2] + interfaces[4:6]:
+            self.assertEqual(interface.link_peers, [rear_ports[0]])
+
+        for interface in interfaces[2:4] + interfaces[6:8]:
+            self.assertEqual(interface.link_peers, [rear_ports[1]])