fix: add additional 4.3 logical matchers (#117)

* fix: add versioned NetBox 4.3 logical matcher for service model

* fix: adds additional logical matchers for types with constraint changes in 4.3

Author: ltucker

netbox_diode_plugin/api/compat.py

@@ -28,19 +28,11 @@ def apply_entity_migrations(data: dict, object_type: str):
 
 def _register_migration(func, min_version, max_version, object_type):
     """Registers a migration function."""
-    min_version = version.parse(min_version)
-    max_version = version.parse(max_version) if max_version else None
-    current_version = _current_netbox_version()
-
-    if current_version < min_version:
-        logger.debug(f"Skipping migration {func.__name__} for {object_type}: min version {min_version}")
-        return
-    if max_version and current_version > max_version:
-        logger.debug(f"Skipping migration {func.__name__} for {object_type}: max version {max_version}")
-        return
-
-    logger.debug(f"Registering migration {func.__name__} for {object_type}.")
-    _MIGRATIONS_BY_OBJECT_TYPE[object_type].append(func)
+    if in_version_range(min_version, max_version):
+        logger.debug(f"Registering migration {func.__name__} for {object_type}.")
+        _MIGRATIONS_BY_OBJECT_TYPE[object_type].append(func)
+    else:
+        logger.debug(f"Skipping migration {func.__name__} for {object_type}: {min_version} to {max_version}.")
 
 @cache
 def _current_netbox_version():
@@ -51,6 +43,17 @@ def _current_netbox_version():
         logger.exception("Failed to determine current version of NetBox.")
         return (0, 0, 0)
 
+def in_version_range(min_version: str | None, max_version: str | None):
+    """Returns True if the current version of NetBox is within the given version range."""
+    min_version = version.parse(min_version) if min_version else None
+    max_version = version.parse(max_version) if max_version else None
+    current_version = _current_netbox_version()
+    if min_version and current_version < min_version:
+        return False
+    if max_version and current_version > max_version:
+        return False
+    return True
+
 def diode_migration(min_version: str, max_version: str | None, object_type: str):
     """Decorator to mark a function as a diode migration."""
     def decorator(func):

netbox_diode_plugin/api/matcher.py

@@ -17,6 +17,7 @@
 from extras.models.customfields import CustomField
 
 from .common import UnresolvedReference
+from .compat import in_version_range
 from .plugin_utils import content_type_id, get_object_type, get_object_type_model
 
 logger = logging.getLogger(__name__)
@@ -162,18 +163,28 @@
             name="logical_service_name_no_device_or_vm",
             model_class=get_object_type_model("ipam.service"),
             condition=Q(device__isnull=True, virtual_machine__isnull=True),
+            max_version="4.2.99",
         ),
         ObjectMatchCriteria(
             fields=("name", "device"),
             name="logical_service_name_on_device",
             model_class=get_object_type_model("ipam.service"),
             condition=Q(device__isnull=False),
+            max_version="4.2.99",
         ),
         ObjectMatchCriteria(
             fields=("name", "virtual_machine"),
             name="logical_service_name_on_vm",
             model_class=get_object_type_model("ipam.service"),
             condition=Q(virtual_machine__isnull=False),
+            max_version="4.2.99",
+        ),
+        ObjectMatchCriteria(
+            fields=("name", "parent_object_type", "parent_object_id"),
+            name="logical_service_name_on_parent",
+            model_class=get_object_type_model("ipam.service"),
+            condition=Q(parent_object_type__isnull=False),
+            min_version="4.3.0"
         ),
     ],
     "dcim.modulebay": lambda: [
@@ -201,6 +212,32 @@
             model_class=get_object_type_model("ipam.fhrpgroup"),
         )
     ],
+    "tenancy.contact": lambda: [
+        ObjectMatchCriteria(
+            # contacts are unconstrained in 4.3.0
+            # in 4.2 they are constrained by unique name per group
+            fields=("name", ),
+            name="logical_contact_name",
+            model_class=get_object_type_model("tenancy.contact"),
+            min_version="4.3.0",
+        )
+    ],
+    "dcim.devicerole": lambda: [
+        ObjectMatchCriteria(
+            fields=("name",),
+            name="logical_device_role_name_no_parent",
+            model_class=get_object_type_model("dcim.devicerole"),
+            condition=Q(parent__isnull=True),
+            min_version="4.3.0",
+        ),
+        ObjectMatchCriteria(
+            fields=("slug",),
+            name="logical_device_role_slug_no_parent",
+            model_class=get_object_type_model("dcim.devicerole"),
+            condition=Q(parent__isnull=True),
+            min_version="4.3.0",
+        )
+    ],
 }
 
 @dataclass
@@ -223,6 +260,9 @@ class ObjectMatchCriteria:
     model_class: type[models.Model] | None = None
     name: str | None = None
 
+    min_version: str | None = None
+    max_version: str | None = None
+
     def __hash__(self):
         """Hash the object match criteria."""
         return hash((self.fields, self.expressions, self.condition, self.model_class.__name__, self.name))
@@ -414,6 +454,9 @@ class CustomFieldMatcher:
     custom_field: str
     model_class: type[models.Model]
 
+    min_version: str | None = None
+    max_version: str | None = None
+
     def fingerprint(self, data: dict) -> str|None:
         """Fingerprint the custom field value."""
         if not self.has_required_fields(data):
@@ -450,6 +493,9 @@ class GlobalIPNetworkIPMatcher:
     model_class: type[models.Model]
     name: str
 
+    min_version: str | None = None
+    max_version: str | None = None
+
     def _check_condition(self, data: dict) -> bool:
         """Check the condition for the custom field."""
         return data.get(self.vrf_field, None) is None
@@ -510,6 +556,9 @@ class VRFIPNetworkIPMatcher:
     model_class: type[models.Model]
     name: str
 
+    min_version: str | None = None
+    max_version: str | None = None
+
     def _check_condition(self, data: dict) -> bool:
         """Check the condition for the custom field."""
         return data.get(self.vrf_field, None) is not None
@@ -584,6 +633,9 @@ class AutoSlugMatcher:
     slug_field: str
     model_class: type[models.Model]
 
+    min_version: str | None = None
+    max_version: str | None = None
+
     def fingerprint(self, data: dict) -> str|None:
         """Fingerprint the custom field value."""
         if not self.has_required_fields(data):
@@ -649,7 +701,10 @@ def _get_autoslug_matchers(model_class) -> list:
 @lru_cache(maxsize=256)
 def _get_model_matchers(model_class) -> list[ObjectMatchCriteria]:
     object_type = get_object_type(model_class)
-    matchers = _LOGICAL_MATCHERS.get(object_type, lambda: [])()
+    matchers = [
+        x for x in _LOGICAL_MATCHERS.get(object_type, lambda: [])()
+        if in_version_range(x.min_version, x.max_version)
+    ]
 
     # collect single fields that are unique
     for field in model_class._meta.fields: