Introduced a 'trace' API endpoint for cable terminations

netbox/dcim/api/serializers.py

@@ -533,6 +533,19 @@ class CableSerializer(ValidatedModelSerializer):
         return self._get_termination(obj, 'b')
 
 
+class TracedCableSerializer(serializers.ModelSerializer):
+    """
+    Used only while tracing a cable path.
+    """
+    url = serializers.HyperlinkedIdentityField(view_name='dcim-api:cable-detail')
+
+    class Meta:
+        model = Cable
+        fields = [
+            'id', 'url', 'type', 'status', 'label', 'color', 'length', 'length_unit',
+        ]
+
+
 class NestedCableSerializer(serializers.Serializer):
     url = serializers.HyperlinkedIdentityField(view_name='dcim-api:cable-detail')
 

netbox/dcim/api/views.py

@@ -22,7 +22,9 @@ from dcim.models import (
 from extras.api.serializers import RenderedGraphSerializer
 from extras.api.views import CustomFieldModelViewSet
 from extras.models import Graph, GRAPH_TYPE_INTERFACE, GRAPH_TYPE_SITE
-from utilities.api import IsAuthenticatedOrLoginNotRequired, FieldChoicesViewSet, ModelViewSet, ServiceUnavailable
+from utilities.api import (
+    get_serializer_for_model, IsAuthenticatedOrLoginNotRequired, FieldChoicesViewSet, ModelViewSet, ServiceUnavailable,
+)
 from . import serializers
 from .exceptions import MissingFilterException
 
@@ -43,6 +45,40 @@ class DCIMFieldChoicesViewSet(FieldChoicesViewSet):
     )
 
 
+# Mixins
+
+class CableTraceMixin(object):
+
+    @action(detail=True, url_path='trace')
+    def trace(self, request, pk):
+        """
+        Trace a complete cable path and return each segment as a three-tuple of (termination, cable, termination).
+        """
+        obj = get_object_or_404(self.queryset.model, pk=pk)
+
+        # Initialize the path array
+        path = []
+
+        for near_end, cable, far_end in obj.trace():
+
+            # Serialize each object
+            serializer_a = get_serializer_for_model(near_end, prefix='Nested')
+            x = serializer_a(near_end, context={'request': request}).data
+            if cable is not None:
+                y = serializers.TracedCableSerializer(cable, context={'request': request}).data
+            else:
+                y = None
+            if far_end is not None:
+                serializer_b = get_serializer_for_model(far_end, prefix='Nested')
+                z = serializer_b(far_end, context={'request': request}).data
+            else:
+                z = None
+
+            path.append((x, y, z))
+
+        return Response(path)
+
+
 #
 # Regions
 #
@@ -329,7 +365,7 @@ class DeviceViewSet(CustomFieldModelViewSet):
 # Device components
 #
 
-class ConsolePortViewSet(ModelViewSet):
+class ConsolePortViewSet(CableTraceMixin, ModelViewSet):
     queryset = ConsolePort.objects.select_related(
         'device', 'connected_endpoint__device', 'cable'
     ).prefetch_related(
@@ -339,7 +375,7 @@ class ConsolePortViewSet(ModelViewSet):
     filter_class = filters.ConsolePortFilter
 
 
-class ConsoleServerPortViewSet(ModelViewSet):
+class ConsoleServerPortViewSet(CableTraceMixin, ModelViewSet):
     queryset = ConsoleServerPort.objects.select_related(
         'device', 'connected_endpoint__device', 'cable'
     ).prefetch_related(
@@ -349,7 +385,7 @@ class ConsoleServerPortViewSet(ModelViewSet):
     filter_class = filters.ConsoleServerPortFilter
 
 
-class PowerPortViewSet(ModelViewSet):
+class PowerPortViewSet(CableTraceMixin, ModelViewSet):
     queryset = PowerPort.objects.select_related(
         'device', 'connected_endpoint__device', 'cable'
     ).prefetch_related(
@@ -359,7 +395,7 @@ class PowerPortViewSet(ModelViewSet):
     filter_class = filters.PowerPortFilter
 
 
-class PowerOutletViewSet(ModelViewSet):
+class PowerOutletViewSet(CableTraceMixin, ModelViewSet):
     queryset = PowerOutlet.objects.select_related(
         'device', 'connected_endpoint__device', 'cable'
     ).prefetch_related(
@@ -369,7 +405,7 @@ class PowerOutletViewSet(ModelViewSet):
     filter_class = filters.PowerOutletFilter
 
 
-class InterfaceViewSet(ModelViewSet):
+class InterfaceViewSet(CableTraceMixin, ModelViewSet):
     queryset = Interface.objects.select_related(
         'device', 'connected_endpoint__device', 'cable'
     ).prefetch_related(

netbox/dcim/models.py

@@ -76,6 +76,55 @@ class CableTermination(models.Model):
     class Meta:
         abstract = True
 
+    def trace(self, position=1):
+        """
+        Return a list representing a complete cable path, with each individual segment represented as a three-tuple:
+            [
+                (termination A, cable, termination B),
+                (termination C, cable, termination D),
+                (termination E, cable, termination F)
+            ]
+        """
+        def get_peer_port(termination, position=1):
+
+            # Map a front port to its corresponding rear port
+            if isinstance(termination, FrontPort):
+                return termination.rear_port, termination.rear_port_position
+
+            # Map a rear port/position to its corresponding front port
+            elif isinstance(termination, RearPort):
+                if position not in range(1, termination.positions + 1):
+                    raise Exception("Invalid position for {} ({} positions): {})".format(
+                        termination, termination.positions, position
+                    ))
+                peer_port = FrontPort.objects.get(
+                    rear_port=termination,
+                    rear_port_position=position,
+                )
+                return peer_port, 1
+
+            # Termination is not a pass-through port
+            else:
+                return None, None
+
+        if not self.cable:
+            return [(self, None, None)]
+
+        far_end = self.cable.termination_b if self.cable.termination_a == self else self.cable.termination_a
+        path = [(self, self.cable, far_end)]
+
+        peer_port, position = get_peer_port(far_end, position)
+        if peer_port is None:
+            return path
+
+        next_segment = peer_port.trace(position)
+        if next_segment is None:
+            return path + [(peer_port, None, None)]
+
+        return path + next_segment
+
+
+    # TODO: Deprecate in favor of obj.cable
     def get_connected_cable(self):
         """
         Return the connected cable if one exists; else None. Assign the far end of the connection on the Cable instance.