feat: OBS-397 - [NetBox Diode Plugin] create API endpoint for retrieving NetBox object state

diode-netbox-plugin/README.md

@@ -23,3 +23,22 @@ PLUGINS_CONFIG = {
     }
 }
 ```
+
+## Running Tests
+
+a) Start the container in diode/diode-server:
+```bash
+make docker-compose-up
+```
+
+b) Enter in the Diode-Netbox container:
+
+```bash
+docker exec -it diode-netbox-1 /bin/bash
+```
+
+c) Execute the tests:
+```bash
+./manage.py test --keepdb netbox_diode_plugin.tests.test_object_state.ObjectStateTestCase
+```
+

diode-netbox-plugin/netbox_diode_plugin/api/serializers.py

@@ -0,0 +1,50 @@
+#!/usr/bin/env python
+# Copyright 2024 NetBox Labs Inc
+"""Diode Netbox Plugin - Serializer."""
+
+from extras.models import ObjectChange
+from rest_framework import serializers
+from utilities.api import get_serializer_for_model
+
+
+class ObjectStateSerializer(serializers.Serializer):
+    """Object State Serializer."""
+
+    object_type = serializers.SerializerMethodField(read_only=True)
+    object_change_id = serializers.SerializerMethodField(read_only=True)
+    object = serializers.SerializerMethodField(read_only=True)
+
+    def get_object_type(self, instance):
+        """
+        Get the object type from context sent from view.
+
+        Return a string with the format "app.model".
+        """
+        return self.context.get("object_type")
+
+    def get_object_change_id(self, instance):
+        """
+        Get the object changed based on instance ID.
+
+        Return the ID of last change.
+        """
+        object_changed = (
+            ObjectChange.objects.filter(changed_object_id=instance.id)
+            .order_by("-id")
+            .values_list("id", flat=True)
+        )
+        return object_changed[0] if len(object_changed) > 0 else None
+
+    def get_object(self, instance):
+        """
+        Get the serializer based on instance model.
+
+        Get the data from the model according to its ID.
+        Return the object according to serializer defined in the Netbox.
+        """
+        serializer = get_serializer_for_model(instance)
+
+        object_data = instance.__class__.objects.filter(id=instance.id)
+
+        context = {"request": self.context.get("request")}
+        return serializer(object_data, context=context, many=True).data[0]

diode-netbox-plugin/netbox_diode_plugin/api/urls.py

@@ -0,0 +1,15 @@
+#!/usr/bin/env python
+# Copyright 2024 NetBox Labs Inc
+"""Diode Netbox Plugin - API URLs."""
+
+from django.urls import include, path
+from netbox.api.routers import NetBoxRouter
+
+from .views import ObjectStateView
+
+router = NetBoxRouter()
+
+urlpatterns = [
+    path("object-state/", ObjectStateView.as_view()),
+    path("", include(router.urls)),
+]

diode-netbox-plugin/netbox_diode_plugin/api/views.py

@@ -0,0 +1,73 @@
+#!/usr/bin/env python
+# Copyright 2024 NetBox Labs Inc
+"""Diode Netbox Plugin - API Views."""
+
+from django.contrib.contenttypes.models import ContentType
+from django.db.models import Q
+from extras.models import CachedValue
+from netbox.search import LookupTypes
+from rest_framework import views
+from rest_framework.exceptions import ValidationError
+from rest_framework.response import Response
+
+from netbox_diode_plugin.api.serializers import ObjectStateSerializer
+
+
+class ObjectStateView(views.APIView):
+    """ObjectState view."""
+
+    authentication_classes = []  # disables authentication
+    permission_classes = []
+
+    def get(self, request, *args, **kwargs):
+        """
+        Return a JSON with object_type, object_change_id, and object.
+
+        Search for objects according to object type.
+        If the obj_type parameter is not in the parameters, raise a ValidationError.
+        When object ID is provided in the request, search using it in the model specified by object type.
+        If ID is not provided, use the q parameter for searching.
+        Lookup is iexact
+        """
+        obj_type = self.request.query_params.get("obj_type", None)
+
+        if not obj_type:
+            raise ValidationError("obj_type parameter is required")
+
+        app_label, model_name = obj_type.split(".")
+        object_type = ContentType.objects.get_by_natural_key(app_label, model_name)
+        object_type_model = object_type.model_class()
+
+        object_id = self.request.query_params.get("id", None)
+
+        if object_id:
+            queryset = object_type_model.objects.filter(id=object_id)
+        else:
+            lookup = LookupTypes.EXACT
+            search_value = self.request.query_params.get("q", None)
+            if not search_value:
+                raise ValidationError("id or q parameter is required")
+
+            query_filter = Q(**{f"value__{lookup}": search_value})
+            query_filter &= Q(object_type__in=[object_type])
+
+            object_id_in_cached_value = CachedValue.objects.filter(
+                query_filter
+            ).values_list("object_id", flat=True)
+
+            queryset = object_type_model.objects.filter(
+                id__in=object_id_in_cached_value
+            )
+
+        serializer = ObjectStateSerializer(
+            queryset,
+            many=True,
+            context={
+                "request": request,
+                "object_type": f"{obj_type}",
+            },
+        )
+
+        if len(serializer.data) > 0:
+            return Response(serializer.data[0])
+        return Response(serializer.data)

diode-netbox-plugin/netbox_diode_plugin/tests/__init__.py

@@ -0,0 +1,3 @@
+#!/usr/bin/env python
+# Copyright 2024 NetBox Labs Inc
+"""Diode Netbox Plugin."""

diode-netbox-plugin/netbox_diode_plugin/tests/test_object_state.py

@@ -0,0 +1,96 @@
+#!/usr/bin/env python
+# Copyright 2024 NetBox Labs Inc
+"""Diode Netbox Plugin - Tests for ObjectStateView."""
+
+from dcim.models import Site
+from django.core.management import call_command
+from rest_framework import status
+from utilities.testing import APITestCase
+
+
+class ObjectStateTestCase(APITestCase):
+    """ObjectState test cases."""
+
+    def setUp(self):
+        """Set up test."""
+        sites = (
+            Site(
+                id=1,
+                name="Site 1",
+                slug="site-1",
+                facility="Alpha",
+                description="First test site",
+                physical_address="123 Fake St Lincoln NE 68588",
+                shipping_address="123 Fake St Lincoln NE 68588",
+                comments="Lorem ipsum etcetera",
+            ),
+            Site(
+                id=2,
+                name="Site 2",
+                slug="site-2",
+                facility="Bravo",
+                description="Second test site",
+                physical_address="725 Cyrus Valleys Suite 761 Douglasfort NE 57761",
+                shipping_address="725 Cyrus Valleys Suite 761 Douglasfort NE 57761",
+                comments="Lorem ipsum etcetera",
+            ),
+            Site(
+                id=3,
+                name="Site 3",
+                slug="site-3",
+                facility="Charlie",
+                description="Third test site",
+                physical_address="2321 Dovie Dale East Cristobal AK 71959",
+                shipping_address="2321 Dovie Dale East Cristobal AK 71959",
+                comments="Lorem ipsum etcetera",
+            ),
+        )
+        Site.objects.bulk_create(sites)
+
+        # call_command is because the searching using q parameter uses CachedValue to get the object ID
+        call_command("reindex")
+
+        self.url = "/api/plugins/diode/object-state/"
+
+    def test_return_object_state_using_id(self):
+        """Test searching using id parameter."""
+        query_parameters = {"id": 1, "obj_type": "dcim.site"}
+
+        response = self.client.get(self.url, query_parameters)
+
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(response.json().get("object").get("name"), "Site 1")
+
+    def test_return_object_state_using_q(self):
+        """Test searching using q parameter."""
+        query_parameters = {"q": "Site 2", "obj_type": "dcim.site"}
+
+        response = self.client.get(self.url, query_parameters)
+
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(response.json().get("object").get("name"), "Site 2")
+
+    def test_object_not_found_return_empty(self):
+        """Test empty searching."""
+        query_parameters = {"q": "Site 10", "obj_type": "dcim.site"}
+
+        response = self.client.get(self.url, query_parameters)
+
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(response.json(), [])
+
+    def test_missing_obj_type_return_400(self):
+        """Test API behavior with missing object type."""
+        query_parameters = {"q": "Site 10", "obj_type": ""}
+
+        response = self.client.get(self.url, query_parameters)
+
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+
+    def test_missing_q_and_id_parameters_return_400(self):
+        """Test API behavior with missing q and ID parameters."""
+        query_parameters = {"obj_type": "dcim.site"}
+
+        response = self.client.get(self.url, query_parameters)
+
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)

diode-netbox-plugin/netbox_diode_plugin/views.py

@@ -11,5 +11,5 @@ class DisplayStateView(View):
 
     def get(self, request):
         """Render a display state template."""
-        return render(request, "netbox_diode_plugin/display_state.html")
+        return render(request, "diode/display_state.html")