feat: prevent duplicate component creation from templates (#143)

Author: mfiedorowicz

docker/docker-compose.yaml

@@ -13,7 +13,7 @@ services:
     env_file: netbox/env/netbox.env
     user: 'unit:root'
     healthcheck:
-      start_period: 60s
+      start_period: 180s
       timeout: 3s
       interval: 15s
       test: "curl -f http://localhost:8080/netbox/api/ || exit 1"

netbox_diode_plugin/api/applier.py

@@ -54,19 +54,67 @@ def apply_changeset(change_set: ChangeSet, request) -> ChangeSetResult:
         id=change_set.id,
     )
 
+def _is_auto_created_component(object_type: str) -> bool:
+    """Check if the object type is auto-created from templates."""
+    auto_created_components = [
+        "dcim.consoleport",
+        "dcim.consoleserverport",
+        "dcim.powerport",
+        "dcim.poweroutlet",
+        "dcim.interface",
+        "dcim.rearport",
+        "dcim.frontport",
+        "dcim.modulebay",
+        "dcim.devicebay",
+        "dcim.inventoryitem",
+    ]
+    return object_type in auto_created_components
+
+
+def _try_find_and_update_existing_instance(data: dict, object_type: str, serializer_class, request):
+    """Try to find existing auto-created instance and update it."""
+    try:
+        instance = find_existing_object(data, object_type)
+        if instance:
+            serializer = serializer_class(instance, data=data, partial=True, context={"request": request})
+            serializer.is_valid(raise_exception=True)
+            return serializer.save()
+    except (ValueError, TypeError) as e:
+        logger.debug(f"Could not find existing {object_type}: {e}")
+    return None
+
+
+def _create_or_find_instance(data: dict, object_type: str, serializer_class, request):
+    """Create new instance or find existing one on conflict."""
+    serializer = serializer_class(data=data, context={"request": request})
+    try:
+        serializer.is_valid(raise_exception=True)
+        return serializer.save()
+    except ValidationError as e:
+        instance = find_existing_object(data, object_type)
+        if not instance:
+            raise e
+        return instance
+
+
 def _apply_change(data: dict, model_class: models.Model, change: Change, created: dict, request):
     serializer_class = get_serializer_for_model(model_class)
     change_type = change.change_type
+
     if change_type == ChangeType.CREATE.value:
-        serializer = serializer_class(data=data, context={"request": request})
-        try:
-            serializer.is_valid(raise_exception=True)
-            instance = serializer.save()
-        except ValidationError as e:
-            instance = find_existing_object(data, change.object_type)
-            if not instance:
-                raise e
-        created[change.ref_id] = instance
+        # For component types that may be auto-created from e.g. DeviceType or ModuleType templates,
+        # try to find existing object first before attempting to create.
+        # This prevents duplicates when components are instantiated during Device/Module save()
+        instance = None
+        if _is_auto_created_component(change.object_type):
+            instance = _try_find_and_update_existing_instance(data, change.object_type, serializer_class, request)
+
+        if not instance:
+            instance = _create_or_find_instance(data, change.object_type, serializer_class, request)
+
+        # Always add the instance to created dict so it can be referenced by subsequent changes
+        if change.ref_id:
+            created[change.ref_id] = instance
 
     elif change_type == ChangeType.UPDATE.value:
         if object_id := change.object_id:

netbox_diode_plugin/tests/test_api_apply_change_set.py

@@ -931,3 +931,206 @@ def test_apply_two_changes_that_create_the_same_object_return_200(self):
             ],
         }
         _ = self.send_request(payload2)
+
+    def test_module_bay_from_template_no_duplicate(self):
+        """Test that module bays created from templates are reused and updated, not duplicated."""
+        from dcim.models import Module, ModuleBay, ModuleBayTemplate, ModuleType
+
+        # Create a device type with a module bay template
+        device_type = DeviceType.objects.create(
+            manufacturer=Manufacturer.objects.first(),
+            model="Device with Module Bay Template",
+            slug="device-with-module-bay-template",
+        )
+
+        # Create module bay template
+        ModuleBayTemplate.objects.create(
+            device_type=device_type,
+            name="Tray",
+        )
+
+        # Create module type
+        module_type = ModuleType.objects.create(
+            manufacturer=Manufacturer.objects.first(),
+            model="Test Module Type",
+        )
+
+        # Step 1: Create a device - this will auto-create module bay "Tray" from template
+        device_payload = {
+            "id": str(uuid.uuid4()),
+            "changes": [
+                {
+                    "change_id": str(uuid.uuid4()),
+                    "change_type": "create",
+                    "object_version": None,
+                    "object_type": "dcim.device",
+                    "object_id": None,
+                    "ref_id": "device-1",
+                    "data": {
+                        "name": "Test Device with Module Bay",
+                        "device_type": device_type.id,
+                        "role": self.roles[0].id,
+                        "site": self.sites[0].id,
+                    },
+                },
+            ],
+        }
+        self.send_request(device_payload)
+
+        # Verify device was created
+        device = Device.objects.get(name="Test Device with Module Bay")
+
+        # Verify module bay was auto-created from template
+        module_bays_before = ModuleBay.objects.filter(device=device, name="Tray")
+        self.assertEqual(module_bays_before.count(), 1)
+        module_bay = module_bays_before.first()
+        self.assertIsNone(module_bay.module)  # No module installed yet
+        self.assertEqual(module_bay.description, "")  # Template has no description
+
+        # Step 2: Create a module with the module bay - should reuse existing bay and update it
+        module_payload = {
+            "id": str(uuid.uuid4()),
+            "changes": [
+                {
+                    "change_id": str(uuid.uuid4()),
+                    "change_type": "create",
+                    "object_version": None,
+                    "object_type": "dcim.modulebay",
+                    "object_id": None,
+                    "ref_id": "modulebay-1",
+                    "data": {
+                        "name": "Tray",
+                        "device": device.id,
+                        "description": "Ingested module bay",
+                    },
+                },
+                {
+                    "change_id": str(uuid.uuid4()),
+                    "change_type": "create",
+                    "object_version": None,
+                    "object_type": "dcim.module",
+                    "object_id": None,
+                    "ref_id": "module-1",
+                    "new_refs": ["module_bay"],
+                    "data": {
+                        "device": device.id,
+                        "module_bay": "modulebay-1",
+                        "module_type": module_type.id,
+                        "description": "Ingested module",
+                    },
+                },
+            ],
+        }
+        self.send_request(module_payload)
+
+        # Verify NO duplicate module bays were created
+        module_bays_after = ModuleBay.objects.filter(device=device, name="Tray")
+        self.assertEqual(
+            module_bays_after.count(),
+            1,
+            "Module bay should be reused, not duplicated"
+        )
+
+        # Verify the module bay was updated with the description
+        module_bay.refresh_from_db()
+        self.assertEqual(
+            module_bay.description,
+            "Ingested module bay",
+            "Module bay should be updated with ingested data"
+        )
+
+        # Verify module was created successfully
+        modules = Module.objects.filter(device=device, module_bay=module_bay)
+        self.assertEqual(modules.count(), 1)
+        module = modules.first()
+        self.assertEqual(module.module_type, module_type)
+        self.assertEqual(module.description, "Ingested module")
+
+    def test_interface_from_template_no_duplicate(self):
+        """Test that interfaces created from templates are reused and updated, not duplicated."""
+        from dcim.models import InterfaceTemplate
+
+        # Create a device type with an interface template
+        device_type = DeviceType.objects.create(
+            manufacturer=Manufacturer.objects.first(),
+            model="Device with Interface Template",
+            slug="device-with-interface-template",
+        )
+
+        # Create interface template
+        InterfaceTemplate.objects.create(
+            device_type=device_type,
+            name="eth0",
+            type="1000base-t",
+        )
+
+        # Step 1: Create a device - this will auto-create interface "eth0" from template
+        device_payload = {
+            "id": str(uuid.uuid4()),
+            "changes": [
+                {
+                    "change_id": str(uuid.uuid4()),
+                    "change_type": "create",
+                    "object_version": None,
+                    "object_type": "dcim.device",
+                    "object_id": None,
+                    "ref_id": "device-1",
+                    "data": {
+                        "name": "Test Device with Interface",
+                        "device_type": device_type.id,
+                        "role": self.roles[0].id,
+                        "site": self.sites[0].id,
+                    },
+                },
+            ],
+        }
+        self.send_request(device_payload)
+
+        # Verify device was created
+        device = Device.objects.get(name="Test Device with Interface")
+
+        # Verify interface was auto-created from template
+        interfaces_before = Interface.objects.filter(device=device, name="eth0")
+        self.assertEqual(interfaces_before.count(), 1)
+        interface = interfaces_before.first()
+        self.assertEqual(interface.description, "")  # Template has no description
+
+        # Step 2: Try to create the same interface with additional data - should reuse and update
+        interface_payload = {
+            "id": str(uuid.uuid4()),
+            "changes": [
+                {
+                    "change_id": str(uuid.uuid4()),
+                    "change_type": "create",
+                    "object_version": None,
+                    "object_type": "dcim.interface",
+                    "object_id": None,
+                    "ref_id": "interface-1",
+                    "data": {
+                        "name": "eth0",
+                        "device": device.id,
+                        "type": "1000base-t",
+                        "description": "Ingested interface",
+                        "enabled": True,
+                    },
+                },
+            ],
+        }
+        self.send_request(interface_payload)
+
+        # Verify NO duplicate interfaces were created
+        interfaces_after = Interface.objects.filter(device=device, name="eth0")
+        self.assertEqual(
+            interfaces_after.count(),
+            1,
+            "Interface should be reused, not duplicated"
+        )
+
+        # Verify the interface was updated with the description
+        interface.refresh_from_db()
+        self.assertEqual(
+            interface.description,
+            "Ingested interface",
+            "Interface should be updated with ingested data"
+        )
+        self.assertTrue(interface.enabled)