Rewrite and extend cabling logic tests

netbox/dcim/tests/test_models.py

@@ -1,6 +1,7 @@
 from django.core.exceptions import ValidationError
 from django.test import TestCase
 
+from circuits.models import *
 from dcim.choices import *
 from dcim.models import *
 from tenancy.models import Tenant
@@ -459,95 +460,346 @@ class CableTestCase(TestCase):
 
 class CablePathTestCase(TestCase):
 
-    def setUp(self):
+    @classmethod
+    def setUpTestData(cls):
 
-        site = Site.objects.create(name='Test Site 1', slug='test-site-1')
-        manufacturer = Manufacturer.objects.create(name='Test Manufacturer 1', slug='test-manufacturer-1')
+        site = Site.objects.create(name='Site 1', slug='site-1')
+        manufacturer = Manufacturer.objects.create(name='Manufacturer 1', slug='manufacturer-1')
         devicetype = DeviceType.objects.create(
-            manufacturer=manufacturer, model='Test Device Type 1', slug='test-device-type-1'
+            manufacturer=manufacturer, model='Device Type 1', slug='device-type-1'
         )
         devicerole = DeviceRole.objects.create(
-            name='Test Device Role 1', slug='test-device-role-1', color='ff0000'
+            name='Device Role 1', slug='device-role-1', color='ff0000'
+        )
+        provider = Provider.objects.create(name='Provider 1', slug='provider-1')
+        circuittype = CircuitType.objects.create(name='Circuit Type 1', slug='circuit-type-1')
+        circuit = Circuit.objects.create(provider=provider, type=circuittype, cid='1')
+        CircuitTermination.objects.bulk_create((
+            CircuitTermination(circuit=circuit, site=site, term_side='A', port_speed=1000),
+            CircuitTermination(circuit=circuit, site=site, term_side='Z', port_speed=1000),
+        ))
+
+        # Create four network devices with four interfaces each
+        devices = (
+            Device(device_type=devicetype, device_role=devicerole, name='Device 1', site=site),
+            Device(device_type=devicetype, device_role=devicerole, name='Device 2', site=site),
+            Device(device_type=devicetype, device_role=devicerole, name='Device 3', site=site),
+            Device(device_type=devicetype, device_role=devicerole, name='Device 4', site=site),
+        )
+        Device.objects.bulk_create(devices)
+        for device in devices:
+            Interface.objects.bulk_create((
+                Interface(device=device, name='Interface 1', type=InterfaceTypeChoices.TYPE_1GE_FIXED),
+                Interface(device=device, name='Interface 2', type=InterfaceTypeChoices.TYPE_1GE_FIXED),
+                Interface(device=device, name='Interface 3', type=InterfaceTypeChoices.TYPE_1GE_FIXED),
+                Interface(device=device, name='Interface 4', type=InterfaceTypeChoices.TYPE_1GE_FIXED),
+            ))
+
+        # Create four patch panels, each with one rear port and four front ports
+        patch_panels = (
+            Device(device_type=devicetype, device_role=devicerole, name='Panel 1', site=site),
+            Device(device_type=devicetype, device_role=devicerole, name='Panel 2', site=site),
+            Device(device_type=devicetype, device_role=devicerole, name='Panel 3', site=site),
+            Device(device_type=devicetype, device_role=devicerole, name='Panel 4', site=site),
+        )
+        Device.objects.bulk_create(patch_panels)
+        for patch_panel in patch_panels:
+            rearport = RearPort.objects.create(device=patch_panel, name='Rear Port 1', positions=4, type=PortTypeChoices.TYPE_8P8C)
+            FrontPort.objects.bulk_create((
+                FrontPort(device=patch_panel, name='Front Port 1', rear_port=rearport, rear_port_position=1, type=PortTypeChoices.TYPE_8P8C),
+                FrontPort(device=patch_panel, name='Front Port 2', rear_port=rearport, rear_port_position=2, type=PortTypeChoices.TYPE_8P8C),
+                FrontPort(device=patch_panel, name='Front Port 3', rear_port=rearport, rear_port_position=3, type=PortTypeChoices.TYPE_8P8C),
+                FrontPort(device=patch_panel, name='Front Port 4', rear_port=rearport, rear_port_position=4, type=PortTypeChoices.TYPE_8P8C),
+            ))
+
+    def test_direct_connection(self):
+        """
+
+        [Device 1] ----- [Device 2]
+             Iface1     Iface1
+
+        """
+        # Create cable
+        cable = Cable(
+            termination_a=Interface.objects.get(device__name='Device 1', name='Interface 1'),
+            termination_b=Interface.objects.get(device__name='Device 2', name='Interface 1')
         )
-        self.device1 = Device.objects.create(
-            device_type=devicetype, device_role=devicerole, name='Test Device 1', site=site
+        cable.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')
+
+        # 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
+        cable.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_patch(self):
+        """
+                     1               2               3
+        [Device 1] ----- [Panel 1] ----- [Panel 2] ----- [Device 2]
+             Iface1     FP1     RP1     RP1     FP1     Iface1
+
+        """
+        # Create cables
+        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')
         )
-        self.device2 = Device.objects.create(
-            device_type=devicetype, device_role=devicerole, name='Test Device 2', site=site
+        cable1.save()
+        cable2 = Cable(
+            termination_a=RearPort.objects.get(device__name='Panel 1', name='Rear Port 1'),
+            termination_b=RearPort.objects.get(device__name='Panel 2', name='Rear Port 1')
         )
-        self.interface1 = Interface.objects.create(device=self.device1, name='eth0')
-        self.interface2 = Interface.objects.create(device=self.device2, name='eth0')
-        self.panel1 = Device.objects.create(
-            device_type=devicetype, device_role=devicerole, name='Test Panel 1', site=site
+        cable2.save()
+        cable3 = Cable(
+            termination_a=FrontPort.objects.get(device__name='Panel 2', name='Front Port 1'),
+            termination_b=Interface.objects.get(device__name='Device 2', name='Interface 1')
         )
-        self.panel2 = Device.objects.create(
-            device_type=devicetype, device_role=devicerole, name='Test Panel 2', site=site
+        cable3.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')
+
+        # 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_multiple_patches(self):
+        """
+                     1               2               3               4               5
+        [Device 1] ----- [Panel 1] ----- [Panel 2] ----- [Panel 3] ----- [Panel 4] ----- [Device 2]
+             Iface1     FP1     RP1     RP1     FP1     FP1     RP1     RP1     FP1     Iface1
+
+        """
+        # Create cables
+        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')
         )
-        self.rear_port1 = RearPort.objects.create(
-            device=self.panel1, name='Rear Port 1', type=PortTypeChoices.TYPE_8P8C
+        cable1.save()
+        cable2 = Cable(
+            termination_a=RearPort.objects.get(device__name='Panel 1', name='Rear Port 1'),
+            termination_b=RearPort.objects.get(device__name='Panel 2', name='Rear Port 1')
         )
-        self.front_port1 = FrontPort.objects.create(
-            device=self.panel1, name='Front Port 1', type=PortTypeChoices.TYPE_8P8C, rear_port=self.rear_port1
+        cable2.save()
+        cable3 = Cable(
+            termination_a=FrontPort.objects.get(device__name='Panel 2', name='Front Port 1'),
+            termination_b=FrontPort.objects.get(device__name='Panel 3', name='Front Port 1')
         )
-        self.rear_port2 = RearPort.objects.create(
-            device=self.panel2, name='Rear Port 2', type=PortTypeChoices.TYPE_8P8C
+        cable3.save()
+        cable4 = Cable(
+            termination_a=RearPort.objects.get(device__name='Panel 3', name='Rear Port 1'),
+            termination_b=RearPort.objects.get(device__name='Panel 4', name='Rear Port 1')
         )
-        self.front_port2 = FrontPort.objects.create(
-            device=self.panel2, name='Front Port 2', type=PortTypeChoices.TYPE_8P8C, rear_port=self.rear_port2
+        cable4.save()
+        cable5 = Cable(
+            termination_a=FrontPort.objects.get(device__name='Panel 4', name='Front Port 1'),
+            termination_b=Interface.objects.get(device__name='Device 2', name='Interface 1')
         )
+        cable5.save()
 
-    def test_path_completion(self):
+        # 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')
 
-        # First segment
-        cable1 = Cable(termination_a=self.interface1, termination_b=self.front_port1)
-        cable1.save()
-        interface1 = Interface.objects.get(pk=self.interface1.pk)
-        self.assertIsNone(interface1.connected_endpoint)
-        self.assertIsNone(interface1.connection_status)
+        # 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)
 
-        # Second segment
-        cable2 = Cable(termination_a=self.rear_port1, termination_b=self.rear_port2)
-        cable2.save()
-        interface1 = Interface.objects.get(pk=self.interface1.pk)
-        self.assertIsNone(interface1.connected_endpoint)
-        self.assertIsNone(interface1.connection_status)
+        # Delete cable 3
+        cable3.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)
 
-        # Third segment
+    def test_connection_via_stacked_rear_ports(self):
+        """
+                     1               2               3               4               5
+        [Device 1] ----- [Panel 1] ----- [Panel 2] ----- [Panel 3] ----- [Panel 4] ----- [Device 2]
+             Iface1     FP1     RP1     FP1     RP1     RP1     FP1     RP1     FP1     Iface1
+
+        """
+        # Create cables
+        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_a=RearPort.objects.get(device__name='Panel 1', name='Rear Port 1'),
+            termination_b=FrontPort.objects.get(device__name='Panel 2', name='Front Port 1')
+        )
+        cable2.save()
         cable3 = Cable(
-            termination_a=self.front_port2,
-            termination_b=self.interface2,
-            status=CableStatusChoices.STATUS_PLANNED
+            termination_a=RearPort.objects.get(device__name='Panel 2', name='Rear Port 1'),
+            termination_b=RearPort.objects.get(device__name='Panel 3', name='Rear Port 1')
         )
         cable3.save()
-        interface1 = Interface.objects.get(pk=self.interface1.pk)
-        self.assertEqual(interface1.connected_endpoint, self.interface2)
-        self.assertFalse(interface1.connection_status)
+        cable4 = Cable(
+            termination_a=FrontPort.objects.get(device__name='Panel 3', name='Front Port 1'),
+            termination_b=RearPort.objects.get(device__name='Panel 4', name='Rear Port 1')
+        )
+        cable4.save()
+        cable5 = Cable(
+            termination_a=FrontPort.objects.get(device__name='Panel 4', name='Front Port 1'),
+            termination_b=Interface.objects.get(device__name='Device 2', name='Interface 1')
+        )
+        cable5.save()
 
-        # Switch third segment from planned to connected
-        cable3.status = CableStatusChoices.STATUS_CONNECTED
-        cable3.save()
-        interface1 = Interface.objects.get(pk=self.interface1.pk)
-        self.assertEqual(interface1.connected_endpoint, self.interface2)
-        self.assertTrue(interface1.connection_status)
+        # 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')
+
+        # 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 3
+        cable3.delete()
 
-    def test_path_teardown(self):
+        # Refresh endpoints
+        endpoint_a.refresh_from_db()
+        endpoint_b.refresh_from_db()
 
-        # Build the path
-        cable1 = Cable(termination_a=self.interface1, termination_b=self.front_port1)
+        # 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_circuit(self):
+        """
+                     1               2
+        [Device 1] ----- [Circuit] ----- [Device 2]
+             Iface1     A         Z     Iface1
+
+        """
+        # Create cables
+        cable1 = Cable(
+            termination_a=Interface.objects.get(device__name='Device 1', name='Interface 1'),
+            termination_b=CircuitTermination.objects.get(term_side='A')
+        )
         cable1.save()
-        cable2 = Cable(termination_a=self.rear_port1, termination_b=self.rear_port2)
+        cable2 = Cable(
+            termination_a=CircuitTermination.objects.get(term_side='Z'),
+            termination_b=Interface.objects.get(device__name='Device 2', name='Interface 1')
+        )
         cable2.save()
-        cable3 = Cable(termination_a=self.front_port2, termination_b=self.interface2)
-        cable3.save()
-        interface1 = Interface.objects.get(pk=self.interface1.pk)
-        self.assertEqual(interface1.connected_endpoint, self.interface2)
-        self.assertTrue(interface1.connection_status)
 
-        # Remove a cable
-        cable2.delete()
-        interface1 = Interface.objects.get(pk=self.interface1.pk)
-        self.assertIsNone(interface1.connected_endpoint)
-        self.assertIsNone(interface1.connection_status)
-        interface2 = Interface.objects.get(pk=self.interface2.pk)
-        self.assertIsNone(interface2.connected_endpoint)
-        self.assertIsNone(interface2.connection_status)
+        # 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')
+
+        # 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 circuit
+        circuit = Circuit.objects.first().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_patched_circuit(self):
+        """
+                     1               2               3               4
+        [Device 1] ----- [Panel 1] ----- [Circuit] ----- [Panel 2] ----- [Device 2]
+             Iface1     FP1     RP1     A         Z     RP1     FP1     Iface1
+
+        """
+        # Create cables
+        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_a=RearPort.objects.get(device__name='Panel 1', name='Rear Port 1'),
+            termination_b=CircuitTermination.objects.get(term_side='A')
+        )
+        cable2.save()
+        cable3 = Cable(
+            termination_a=CircuitTermination.objects.get(term_side='Z'),
+            termination_b=RearPort.objects.get(device__name='Panel 2', name='Rear Port 1')
+        )
+        cable3.save()
+        cable4 = Cable(
+            termination_a=FrontPort.objects.get(device__name='Panel 2', name='Front Port 1'),
+            termination_b=Interface.objects.get(device__name='Device 2', name='Interface 1')
+        )
+        cable4.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')
+
+        # 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 circuit
+        circuit = Circuit.objects.first().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)