fix: consider object type in fingerprint all deduplication

Author: ltucker

netbox_diode_plugin/api/matcher.py

@@ -735,7 +735,7 @@ def _get_refs(expr) -> set[str]:
         logger.warning(f"Unhandled expression type for _get_refs: {type(expr)}")
     return refs
 
-def _fingerprint_all(data: dict) -> str:
+def _fingerprint_all(data: dict, object_type: str|None = None) -> str:
     """
     Returns a fingerprint of the data based on all fields.
 
@@ -745,7 +745,7 @@ def _fingerprint_all(data: dict) -> str:
     if data is None:
         return None
 
-    values = []
+    values = ["object_type", object_type]
     for k, v in sorted(data.items()):
         if k.startswith("_"):
             continue
@@ -776,7 +776,7 @@ def fingerprints(data: dict, object_type: str) -> list[str]:
         fp = matcher.fingerprint(data)
         if fp is not None:
             fps.append(fp)
-    fp = _fingerprint_all(data)
+    fp = _fingerprint_all(data, object_type)
     fps.append(fp)
     return fps
 

netbox_diode_plugin/tests/test_api_generate_diff.py

@@ -3,6 +3,7 @@
 """Diode NetBox Plugin - Tests."""
 
 import logging
+from collections import defaultdict
 from types import SimpleNamespace
 from unittest import mock
 from uuid import uuid4
@@ -377,6 +378,39 @@ def test_vlangroup_error(self):
         }
         _ = self.send_request(payload)
 
+    def test_generate_diff_dedupe_different_object_types(self):
+        """Test generate diff dedupe different object types with same values."""
+        payload = {
+            "timestamp": 1,
+            "object_type": "dcim.device",
+            "entity": {
+                "device": {
+                    "name": "Cat8000V",
+                    "role": {"name": "undefined"},
+                    "site": {"name": "undefined"},
+                    "serial": "9OBXJHNNU5V",
+                    "status": "active",
+                    "platform": {"name": "ios", "manufacturer": {"name": "undefined"}},
+                    "device_type": {"model": "C8000V", "manufacturer": {"name": "undefined"}}
+                },
+            },
+        }
+        response = self.send_request(payload)
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        cs = response.json().get("change_set", {})
+        self.assertIsNotNone(cs.get("id"))
+        changes = cs.get("changes", [])
+        self.assertEqual(len(changes), 6)
+        by_object_type = defaultdict(int)
+        for change in changes:
+            by_object_type[change.get("object_type")] += 1
+
+        self.assertEqual(by_object_type["dcim.device"], 1)
+        self.assertEqual(by_object_type["dcim.manufacturer"], 1)
+        self.assertEqual(by_object_type["dcim.platform"], 1)
+        self.assertEqual(by_object_type["dcim.devicetype"], 1)
+        self.assertEqual(by_object_type["dcim.site"], 1)
+        self.assertEqual(by_object_type["dcim.devicerole"], 1)
 
     def send_request(self, payload, status_code=status.HTTP_200_OK):
         """Post the payload to the url and return the response."""