fix: expand support for cycle breaking, add additional logical matchers

netbox_diode_plugin/api/common.py

@@ -2,16 +2,17 @@
 # Copyright 2025 NetBox Labs Inc
 """Diode NetBox Plugin - API - Common types and utilities."""
 
-from collections import defaultdict
 import logging
 import uuid
+from collections import defaultdict
 from dataclasses import dataclass, field
 from enum import Enum
 
 from django.apps import apps
-from django.contrib.contenttypes.fields import GenericRelation, GenericForeignKey
+from django.contrib.contenttypes.fields import GenericForeignKey, GenericRelation
 from django.contrib.contenttypes.models import ContentType
 from django.core.exceptions import ValidationError
+from django.db import models
 from rest_framework import status
 
 logger = logging.getLogger("netbox.diode_data")
@@ -108,31 +109,7 @@ def validate(self) -> dict[str, list[str]]:
             if change.before:
                 change_data.update(change.before)
 
-            # check that there is some value for every required
-            # reference field, but don't validate the actual reference.
-            excluded_relation_fields = []
-            rel_errors = defaultdict(list)
-            for f in model._meta.get_fields():
-                if isinstance(f, (GenericRelation, GenericForeignKey)):
-                    excluded_relation_fields.append(f.name)
-                    continue
-                if not f.is_relation:
-                    continue
-                field_name = f.name
-                excluded_relation_fields.append(field_name)
-
-                if hasattr(f, "related_model") and f.related_model == ContentType:
-                    change_data.pop(field_name, None)
-                    base_field = field_name[:-5]
-                    excluded_relation_fields.append(base_field + "_id")
-                    value = change_data.pop(base_field + "_id", None)
-                else:
-                    value = change_data.pop(field_name, None)
-
-                if not f.null and not f.blank and not f.many_to_many:
-                    # this field is a required relation...
-                    if value is None:
-                        rel_errors[f.name].append(f"Field {f.name} is required")
+            excluded_relation_fields, rel_errors = self._validate_relations(change_data, model)
             if rel_errors:
                 errors[change.object_type] = rel_errors
 
@@ -144,6 +121,36 @@ def validate(self) -> dict[str, list[str]]:
 
         return errors or None
 
+    def _validate_relations(self, change_data: dict, model: models.Model) -> tuple[list[str], dict]:
+        # check that there is some value for every required
+        # reference field, but don't validate the actual reference.
+        # the fields are removed from the change_data so that other
+        # fields can be validated by instantiating the model.
+        excluded_relation_fields = []
+        rel_errors = defaultdict(list)
+        for f in model._meta.get_fields():
+            if isinstance(f, (GenericRelation, GenericForeignKey)):
+                excluded_relation_fields.append(f.name)
+                continue
+            if not f.is_relation:
+                continue
+            field_name = f.name
+            excluded_relation_fields.append(field_name)
+
+            if hasattr(f, "related_model") and f.related_model == ContentType:
+                change_data.pop(field_name, None)
+                base_field = field_name[:-5]
+                excluded_relation_fields.append(base_field + "_id")
+                value = change_data.pop(base_field + "_id", None)
+            else:
+                value = change_data.pop(field_name, None)
+
+            if not f.null and not f.blank and not f.many_to_many:
+                # this field is a required relation...
+                if value is None:
+                    rel_errors[f.name].append(f"Field {f.name} is required")
+        return excluded_relation_fields, rel_errors
+
 
 @dataclass
 class ChangeSetResult:

netbox_diode_plugin/api/matcher.py

@@ -9,6 +9,7 @@
 from typing import Type
 
 from core.models import ObjectType as NetBoxType
+from django.conf import settings
 from django.contrib.contenttypes.fields import ContentType
 from django.core.exceptions import FieldDoesNotExist
 from django.db import models
@@ -30,11 +31,44 @@
 _LOGICAL_MATCHERS = {
     "dcim.macaddress": lambda: [
         ObjectMatchCriteria(
-            # consider a matching mac address within the same parent object
-            # to be the same object although not technically required to be.
             fields=("mac_address", "assigned_object_type", "assigned_object_id"),
             name="logical_mac_address_within_parent",
             model_class=get_object_type_model("dcim.macaddress"),
+            condition=Q(assigned_object_id__isnull=False),
+        ),
+        ObjectMatchCriteria(
+            fields=("mac_address", "assigned_object_type", "assigned_object_id"),
+            name="logical_mac_address_within_parent",
+            model_class=get_object_type_model("dcim.macaddress"),
+            condition=Q(assigned_object_id__isnull=True),
+        ),
+    ],
+    "ipam.ipaddress": lambda: [
+        ObjectMatchCriteria(
+            fields=("address", ),
+            name="logical_ip_address_global_no_vrf",
+            model_class=get_object_type_model("ipam.ipaddress"),
+            condition=Q(vrf__isnull=True),
+        ),
+        ObjectMatchCriteria(
+            fields=("address", "assigned_object_type", "assigned_object_id"),
+            name="logical_ip_address_within_vrf",
+            model_class=get_object_type_model("ipam.ipaddress"),
+            condition=Q(vrf__isnull=False)
+        ),
+    ],
+    "ipam.prefix": lambda: [
+         ObjectMatchCriteria(
+            fields=("prefix",),
+            name="logical_prefix_global_no_vrf",
+            model_class=get_object_type_model("ipam.prefix"),
+            condition=Q(vrf__isnull=True),
+        ),
+        ObjectMatchCriteria(
+            fields=("prefix", "vrf_id"),
+            name="logical_prefix_within_vrf",
+            model_class=get_object_type_model("ipam.prefix"),
+            condition=Q(vrf__isnull=False),
         ),
     ],
 }
@@ -404,22 +438,3 @@ def find_existing_object(data: dict, object_type: str):
         logger.error(f"      -> No object found for matcher {matcher.name}")
     logger.error("  * No matchers found an existing object")
     return None
-
-def merge_data(a: dict, b: dict) -> dict:
-    """
-    Merges two structures.
-
-    If there are any conflicts, an error is raised.
-    Ignores conflicts in fields that start with an underscore,
-    preferring a's value.
-    """
-    if a is None or b is None:
-        raise ValueError("Cannot merge None values")
-    merged = a.copy()
-    for k, v in b.items():
-        if k.startswith("_"):
-            continue
-        if k in merged and merged[k] != v:
-            raise ValueError(f"Conflict merging {a} and {b} on {k}: {merged[k]} and {v}")
-        merged[k] = v
-    return merged