mirror of
https://github.com/netbox-community/netbox.git
synced 2025-07-16 04:02:52 -06:00
make single front/rear port work when between panels
This commit is contained in:
parent
edf15532d2
commit
a0f4d481dc
@ -115,26 +115,32 @@ class CableTermination(models.Model):
|
|||||||
|
|
||||||
# Map a front port to its corresponding rear port
|
# Map a front port to its corresponding rear port
|
||||||
if isinstance(termination, FrontPort):
|
if isinstance(termination, FrontPort):
|
||||||
position_stack.append(termination.rear_port_position)
|
|
||||||
# Retrieve the corresponding RearPort from database to ensure we have an up-to-date instance
|
# Retrieve the corresponding RearPort from database to ensure we have an up-to-date instance
|
||||||
peer_port = RearPort.objects.get(pk=termination.rear_port.pk)
|
peer_port = RearPort.objects.get(pk=termination.rear_port.pk)
|
||||||
|
|
||||||
|
# Don't use the stack for 1-on-1 ports, they don't have to come in pairs
|
||||||
|
if peer_port.positions > 1:
|
||||||
|
position_stack.append(termination.rear_port_position)
|
||||||
|
|
||||||
return peer_port
|
return peer_port
|
||||||
|
|
||||||
# Map a rear port/position to its corresponding front port
|
# Map a rear port/position to its corresponding front port
|
||||||
elif isinstance(termination, RearPort):
|
elif isinstance(termination, RearPort):
|
||||||
|
if termination.positions > 1:
|
||||||
|
# Can't map to a FrontPort without a position if there are multiple options
|
||||||
|
if not position_stack:
|
||||||
|
raise CableTraceSplit(termination)
|
||||||
|
|
||||||
# Can't map to a FrontPort without a position if there are multiple options
|
position = position_stack.pop()
|
||||||
if termination.positions > 1 and not position_stack:
|
|
||||||
raise CableTraceSplit(termination)
|
|
||||||
|
|
||||||
# We can assume position 1 if the RearPort has only one position
|
# Validate the position
|
||||||
position = position_stack.pop() if position_stack else 1
|
if position not in range(1, termination.positions + 1):
|
||||||
|
raise Exception("Invalid position for {} ({} positions): {})".format(
|
||||||
# Validate the position
|
termination, termination.positions, position
|
||||||
if position not in range(1, termination.positions + 1):
|
))
|
||||||
raise Exception("Invalid position for {} ({} positions): {})".format(
|
else:
|
||||||
termination, termination.positions, position
|
# Don't use the stack for 1-on-1 ports, they don't have to come in pairs
|
||||||
))
|
position = 1
|
||||||
|
|
||||||
try:
|
try:
|
||||||
peer_port = FrontPort.objects.get(
|
peer_port = FrontPort.objects.get(
|
||||||
|
@ -512,6 +512,11 @@ class CablePathTestCase(TestCase):
|
|||||||
FrontPort(device=patch_panel, name='Front Port 4', rear_port=rearport, rear_port_position=4, type=PortTypeChoices.TYPE_8P8C),
|
FrontPort(device=patch_panel, name='Front Port 4', rear_port=rearport, rear_port_position=4, type=PortTypeChoices.TYPE_8P8C),
|
||||||
))
|
))
|
||||||
|
|
||||||
|
# Create a 1-on-1 patch panel
|
||||||
|
patch_panel = Device.objects.create(device_type=devicetype, device_role=devicerole, name='Panel 5', site=site)
|
||||||
|
rearport = RearPort.objects.create(device=patch_panel, name='Rear Port 1', positions=1, type=PortTypeChoices.TYPE_8P8C)
|
||||||
|
FrontPort.objects.create(device=patch_panel, name='Front Port 1', rear_port=rearport, rear_port_position=1, type=PortTypeChoices.TYPE_8P8C)
|
||||||
|
|
||||||
def test_direct_connection(self):
|
def test_direct_connection(self):
|
||||||
"""
|
"""
|
||||||
Test a direct connection between two interfaces.
|
Test a direct connection between two interfaces.
|
||||||
@ -554,17 +559,17 @@ class CablePathTestCase(TestCase):
|
|||||||
Test a connection which passes through a single front/rear port pair.
|
Test a connection which passes through a single front/rear port pair.
|
||||||
|
|
||||||
1 2
|
1 2
|
||||||
[Device 1] ----- [Panel 1] ----- [Device 2]
|
[Device 1] ----- [Panel 5] ----- [Device 2]
|
||||||
Iface1 FP1 RP1 Iface1
|
Iface1 FP1 RP1 Iface1
|
||||||
"""
|
"""
|
||||||
# Create cables
|
# Create cables (FP first, RP second)
|
||||||
cable1 = Cable(
|
cable1 = Cable(
|
||||||
termination_a=Interface.objects.get(device__name='Device 1', name='Interface 1'),
|
termination_a=Interface.objects.get(device__name='Device 1', name='Interface 1'),
|
||||||
termination_b=FrontPort.objects.get(device__name='Panel 1', name='Front Port 1')
|
termination_b=FrontPort.objects.get(device__name='Panel 5', name='Front Port 1')
|
||||||
)
|
)
|
||||||
cable1.save()
|
cable1.save()
|
||||||
cable2 = Cable(
|
cable2 = Cable(
|
||||||
termination_b=RearPort.objects.get(device__name='Panel 1', name='Rear Port 1'),
|
termination_b=RearPort.objects.get(device__name='Panel 5', name='Rear Port 1'),
|
||||||
termination_a=Interface.objects.get(device__name='Device 2', name='Interface 1')
|
termination_a=Interface.objects.get(device__name='Device 2', name='Interface 1')
|
||||||
)
|
)
|
||||||
cable2.save()
|
cable2.save()
|
||||||
@ -592,6 +597,155 @@ class CablePathTestCase(TestCase):
|
|||||||
self.assertIsNone(endpoint_a.connection_status)
|
self.assertIsNone(endpoint_a.connection_status)
|
||||||
self.assertIsNone(endpoint_b.connection_status)
|
self.assertIsNone(endpoint_b.connection_status)
|
||||||
|
|
||||||
|
# Recreate cable 1 to test creating the cables in reverse order (RP first, FP second)
|
||||||
|
cable1.save()
|
||||||
|
|
||||||
|
# Refresh endpoints
|
||||||
|
endpoint_a.refresh_from_db()
|
||||||
|
endpoint_b.refresh_from_db()
|
||||||
|
|
||||||
|
# Validate connections
|
||||||
|
self.assertEqual(endpoint_a.connected_endpoint, endpoint_b)
|
||||||
|
self.assertEqual(endpoint_b.connected_endpoint, endpoint_a)
|
||||||
|
self.assertTrue(endpoint_a.connection_status)
|
||||||
|
self.assertTrue(endpoint_b.connection_status)
|
||||||
|
|
||||||
|
# Delete cable 2
|
||||||
|
cable2.delete()
|
||||||
|
|
||||||
|
# Refresh endpoints
|
||||||
|
endpoint_a.refresh_from_db()
|
||||||
|
endpoint_b.refresh_from_db()
|
||||||
|
|
||||||
|
# Check that connections have been nullified
|
||||||
|
self.assertIsNone(endpoint_a.connected_endpoint)
|
||||||
|
self.assertIsNone(endpoint_b.connected_endpoint)
|
||||||
|
self.assertIsNone(endpoint_a.connection_status)
|
||||||
|
self.assertIsNone(endpoint_b.connection_status)
|
||||||
|
|
||||||
|
def test_connection_via_nested_single_rear_port(self):
|
||||||
|
"""
|
||||||
|
Test a connection which passes through a single front/rear port pair between two multi-port MUXes.
|
||||||
|
|
||||||
|
Test two connections via patched rear ports:
|
||||||
|
Device 1 <---> Device 2
|
||||||
|
Device 3 <---> Device 4
|
||||||
|
|
||||||
|
1 2
|
||||||
|
[Device 1] -----------+ +----------- [Device 2]
|
||||||
|
Iface1 | | Iface1
|
||||||
|
FP1 | 3 4 | FP1
|
||||||
|
[Panel 1] ----- [Panel 5] ----- [Panel 2]
|
||||||
|
FP2 | RP1 RP1 FP1 RP1 | FP2
|
||||||
|
Iface1 | | Iface1
|
||||||
|
[Device 3] -----------+ +----------- [Device 4]
|
||||||
|
5 6
|
||||||
|
"""
|
||||||
|
# Create cables (Panel 5 RP first, FP second)
|
||||||
|
cable1 = Cable(
|
||||||
|
termination_a=Interface.objects.get(device__name='Device 1', name='Interface 1'),
|
||||||
|
termination_b=FrontPort.objects.get(device__name='Panel 1', name='Front Port 1')
|
||||||
|
)
|
||||||
|
cable1.save()
|
||||||
|
cable2 = Cable(
|
||||||
|
termination_b=FrontPort.objects.get(device__name='Panel 2', name='Front Port 1'),
|
||||||
|
termination_a=Interface.objects.get(device__name='Device 2', name='Interface 1')
|
||||||
|
)
|
||||||
|
cable2.save()
|
||||||
|
cable3 = Cable(
|
||||||
|
termination_b=RearPort.objects.get(device__name='Panel 1', name='Rear Port 1'),
|
||||||
|
termination_a=RearPort.objects.get(device__name='Panel 5', name='Rear Port 1')
|
||||||
|
)
|
||||||
|
cable3.save()
|
||||||
|
cable4 = Cable(
|
||||||
|
termination_b=FrontPort.objects.get(device__name='Panel 5', name='Front Port 1'),
|
||||||
|
termination_a=RearPort.objects.get(device__name='Panel 2', name='Rear Port 1')
|
||||||
|
)
|
||||||
|
cable4.save()
|
||||||
|
cable5 = Cable(
|
||||||
|
termination_b=FrontPort.objects.get(device__name='Panel 1', name='Front Port 2'),
|
||||||
|
termination_a=Interface.objects.get(device__name='Device 3', name='Interface 1')
|
||||||
|
)
|
||||||
|
cable5.save()
|
||||||
|
cable6 = Cable(
|
||||||
|
termination_b=FrontPort.objects.get(device__name='Panel 2', name='Front Port 2'),
|
||||||
|
termination_a=Interface.objects.get(device__name='Device 4', name='Interface 1')
|
||||||
|
)
|
||||||
|
cable6.save()
|
||||||
|
|
||||||
|
# Retrieve endpoints
|
||||||
|
endpoint_a = Interface.objects.get(device__name='Device 1', name='Interface 1')
|
||||||
|
endpoint_b = Interface.objects.get(device__name='Device 2', name='Interface 1')
|
||||||
|
endpoint_c = Interface.objects.get(device__name='Device 3', name='Interface 1')
|
||||||
|
endpoint_d = Interface.objects.get(device__name='Device 4', name='Interface 1')
|
||||||
|
|
||||||
|
# Validate connections
|
||||||
|
self.assertEqual(endpoint_a.connected_endpoint, endpoint_b)
|
||||||
|
self.assertEqual(endpoint_b.connected_endpoint, endpoint_a)
|
||||||
|
self.assertEqual(endpoint_c.connected_endpoint, endpoint_d)
|
||||||
|
self.assertEqual(endpoint_d.connected_endpoint, endpoint_c)
|
||||||
|
self.assertTrue(endpoint_a.connection_status)
|
||||||
|
self.assertTrue(endpoint_b.connection_status)
|
||||||
|
self.assertTrue(endpoint_c.connection_status)
|
||||||
|
self.assertTrue(endpoint_d.connection_status)
|
||||||
|
|
||||||
|
# Delete cable 3
|
||||||
|
cable3.delete()
|
||||||
|
|
||||||
|
# Refresh endpoints
|
||||||
|
endpoint_a.refresh_from_db()
|
||||||
|
endpoint_b.refresh_from_db()
|
||||||
|
endpoint_c.refresh_from_db()
|
||||||
|
endpoint_d.refresh_from_db()
|
||||||
|
|
||||||
|
# Check that connections have been nullified
|
||||||
|
self.assertIsNone(endpoint_a.connected_endpoint)
|
||||||
|
self.assertIsNone(endpoint_b.connected_endpoint)
|
||||||
|
self.assertIsNone(endpoint_c.connected_endpoint)
|
||||||
|
self.assertIsNone(endpoint_d.connected_endpoint)
|
||||||
|
self.assertIsNone(endpoint_a.connection_status)
|
||||||
|
self.assertIsNone(endpoint_b.connection_status)
|
||||||
|
self.assertIsNone(endpoint_c.connection_status)
|
||||||
|
self.assertIsNone(endpoint_d.connection_status)
|
||||||
|
|
||||||
|
# Recreate cable 3 to test reverse order (Panel 5 FP first, RP second)
|
||||||
|
cable3.save()
|
||||||
|
|
||||||
|
# Refresh endpoints
|
||||||
|
endpoint_a.refresh_from_db()
|
||||||
|
endpoint_b.refresh_from_db()
|
||||||
|
endpoint_c.refresh_from_db()
|
||||||
|
endpoint_d.refresh_from_db()
|
||||||
|
|
||||||
|
# Validate connections
|
||||||
|
self.assertEqual(endpoint_a.connected_endpoint, endpoint_b)
|
||||||
|
self.assertEqual(endpoint_b.connected_endpoint, endpoint_a)
|
||||||
|
self.assertEqual(endpoint_c.connected_endpoint, endpoint_d)
|
||||||
|
self.assertEqual(endpoint_d.connected_endpoint, endpoint_c)
|
||||||
|
self.assertTrue(endpoint_a.connection_status)
|
||||||
|
self.assertTrue(endpoint_b.connection_status)
|
||||||
|
self.assertTrue(endpoint_c.connection_status)
|
||||||
|
self.assertTrue(endpoint_d.connection_status)
|
||||||
|
|
||||||
|
# Delete cable 4
|
||||||
|
cable4.delete()
|
||||||
|
|
||||||
|
# Refresh endpoints
|
||||||
|
endpoint_a.refresh_from_db()
|
||||||
|
endpoint_b.refresh_from_db()
|
||||||
|
endpoint_c.refresh_from_db()
|
||||||
|
endpoint_d.refresh_from_db()
|
||||||
|
|
||||||
|
# Check that connections have been nullified
|
||||||
|
self.assertIsNone(endpoint_a.connected_endpoint)
|
||||||
|
self.assertIsNone(endpoint_b.connected_endpoint)
|
||||||
|
self.assertIsNone(endpoint_c.connected_endpoint)
|
||||||
|
self.assertIsNone(endpoint_d.connected_endpoint)
|
||||||
|
self.assertIsNone(endpoint_a.connection_status)
|
||||||
|
self.assertIsNone(endpoint_b.connection_status)
|
||||||
|
self.assertIsNone(endpoint_c.connection_status)
|
||||||
|
self.assertIsNone(endpoint_d.connection_status)
|
||||||
|
|
||||||
def test_connections_via_patch(self):
|
def test_connections_via_patch(self):
|
||||||
"""
|
"""
|
||||||
Test two connections via patched rear ports:
|
Test two connections via patched rear ports:
|
||||||
|
Loading…
Reference in New Issue
Block a user