feat: OBS-399 - netbox diode plugin to listen on post_save signals

diode-netbox-plugin/netbox_diode_plugin/signals.py

@@ -4,4 +4,65 @@
 
 import logging
 
+from django.contrib.auth import get_user_model
+from django.contrib.contenttypes.models import ContentType
+from django.db.models.signals import post_save
+from django.dispatch import receiver
+from django.forms import model_to_dict
+from extras.models import ObjectChange
+from users.models import Token
+
+from netbox_diode_plugin.diode_reconciler_sdk.client import DiodeReconcilerClient
+
 logger = logging.getLogger("netbox.netbox_diode_plugin")
+
+User = get_user_model()
+
+
+def get_netbox_to_diode_token():
+    """Get token for NETBOX_TO_DIODE."""
+    user = get_user_model().objects.get(username="NETBOX_TO_DIODE")
+    return Token.objects.get(user=user)
+
+
+@receiver(post_save)
+def handle_notify_diode(instance, created, sender, update_fields, **kwargs):
+    """Handle notify reconciliation."""
+    logger.debug("Handling notify reconciliation.")
+
+    content_type = ContentType.objects.get_for_model(sender, for_concrete_model=False)
+    app_label = content_type.app_label
+    model_name = content_type.model  # noqa
+
+    if app_label in ["dcim", "ipam"]:
+
+        object_changed = (
+            ObjectChange.objects.filter(changed_object_id=instance.id)
+            .order_by("id")
+            .last()
+        )
+        object_id = instance.id  # noqa
+        object_type = f"{app_label}.{model_name}"  # noqa
+        object_changed_id = object_changed if object_changed else None  # noqa
+        object = {model_name: model_to_dict(instance)}  # noqa
+
+        try:
+            sdk = DiodeReconcilerClient(  # noqa
+                "diode-reconciler:8081", get_netbox_to_diode_token()
+            )
+
+            # Comment out because the DiodeReconcilerClient need some adjustments.
+
+            # sdk.add_object_state(
+            #     object_id=object_id,
+            #     object_type=object_type,
+            #     object_change_id=object_changed_id,
+            #     object=object,
+            # )
+        except Exception as e:
+            logger.error(e)
+            return False
+
+        return True
+
+    return None

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

@@ -26,6 +26,10 @@ class BaseApplyChangeSet(APITestCase):
 
     def setUp(self):
         """Set up test."""
+        # Necessary to use with signals.
+        self.user_netbox_to_diode = User.objects.create_user(username="NETBOX_TO_DIODE")
+        Token.objects.create(user=self.user_netbox_to_diode)
+
         self.user = User.objects.create_user(username="testcommonuser")
         self.add_permissions("netbox_diode_plugin.add_diode")
         self.user_token = Token.objects.create(user=self.user)

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

@@ -0,0 +1,40 @@
+from dcim.models import Site
+from django.contrib.auth import get_user_model
+from django.test import TestCase
+from users.models import Token
+
+from netbox_diode_plugin.signals import handle_notify_diode
+
+User = get_user_model()
+
+
+class HandleNotifyDiodeTestCase(TestCase):
+    """Test Handle Notify Diode class."""
+
+    def setUp(self):
+        """Set up."""
+        self.user_netbox_to_diode = User.objects.create_user(username="NETBOX_TO_DIODE")
+        Token.objects.create(user=self.user_netbox_to_diode)
+
+        self.site = Site(
+            id=10,
+            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",
+        )
+        self.site.save()
+
+    def test_handle_notify_diode_success(self):
+        """Test handle notify diode success."""
+        instance = Site.objects.get(id=10)
+        self.assertTrue(handle_notify_diode(instance, True, Site, None))
+
+    def test_handle_notify_diode_failure(self):
+        """Test handle notify diode failure."""
+        Token.objects.filter(user=self.user_netbox_to_diode).delete()
+        instance = Site.objects.get(id=10)
+        self.assertFalse(handle_notify_diode(instance, True, Site, None))