feat: module bay types

core/change_detector.py

@@ -6,13 +6,13 @@
 
 import os
 from dataclasses import dataclass, field
-from functools import lru_cache
-from typing import Any, List, Optional
 from enum import Enum
+from functools import lru_cache
+from typing import Any
 
-from core.component_registry import COMPONENT_TYPES
-from core.normalization import normalize_values
+from core.component_registry import BY_YAML_KEY, COMPONENT_TYPES
 from core.formatting import log_property_diffs
+from core.normalization import is_explicit_list, normalize_values
 from core.schema_reader import load_properties_for_type
 
 
@@ -26,6 +26,114 @@ class ChangeType(Enum):
     COMPONENT_REMOVED = "component_removed"
 
 
+def _manufacturer_slug_of(yaml_data):
+    """Return the owning manufacturer's slug from a parsed definition.
+
+    ``core.repo`` rewrites every YAML manufacturer to ``{"slug": ...}`` before the importer
+    sees it, so that is the usual shape here.
+    """
+    manufacturer = (yaml_data or {}).get("manufacturer")
+    if isinstance(manufacturer, dict):
+        return manufacturer.get("slug") or ""
+    return manufacturer or ""
+
+
+def _relation_properties(comp_type):
+    """Return the relation field names the registry declares for *comp_type*."""
+    component = BY_YAML_KEY.get(comp_type)
+    return component.relations if component else ()
+
+
+def _is_relation_list(value):
+    """Return True when *value* is a list of non-empty strings, the only shape a reference takes.
+
+    Blank is checked after stripping, matching the catalog: accepting "   " here only defers
+    the rejection to the write path, where it skips the whole component's update.
+    """
+    return is_explicit_list(value) and all(isinstance(item, str) and item.strip() for item in value)
+
+
+def _relation_change(prop, yaml_comp, netbox_comp, catalog=None, manufacturer=None, handle=None):
+    """Return a PropertyChange when a relation differs, or None when there is nothing to do.
+
+    A relation is an unordered set of related objects.  Where the catalog is available the
+    comparison is between *identities*, ``(manufacturer slug, slug)``, because two
+    manufacturers may define the same class name and a bay holding the wrong one reads as
+    equal on names alone.  Without a catalog it falls back to comparing names, which is
+    what a caller that has not wired one can still do.
+
+    An omitted key leaves the relation unmanaged and an empty list clears it, but a bare
+    ``module_bay_types:`` parses as None and a malformed entry is not a name: those are left
+    unmanaged, because reading them as empty would clear a restriction nobody removed.  A
+    field the query did not return is skipped for the same reason.
+    """
+    if prop not in yaml_comp:
+        return None
+    declared = yaml_comp.get(prop)
+    if not _is_relation_list(declared):
+        if handle is not None:
+            handle.log(
+                f"Ignored {prop} on {yaml_comp.get('name', 'Unknown')!r}: expected a list of names, got {declared!r}"
+            )
+        return None
+    netbox_value = getattr(netbox_comp, prop, _MISSING)
+    if netbox_value is _MISSING:
+        return None
+
+    if catalog is not None and manufacturer:
+        from core.module_bay_types import ModuleBayTypeError
+
+        try:
+            wanted = catalog.identities_for(manufacturer, declared)
+        except ModuleBayTypeError:
+            # Reported as a change so the write path sees it; "no change" is silent.
+            return PropertyChange(
+                property_name=prop,
+                old_value=sorted(_relation_names(netbox_value)),
+                new_value=sorted(set(declared)),
+            )
+        current = _relation_identities(netbox_value)
+        if current is not None and wanted == current:
+            return None
+        if current is not None:
+            return PropertyChange(
+                property_name=prop, old_value=sorted(_relation_names(netbox_value)), new_value=sorted(set(declared))
+            )
+
+    yaml_names = frozenset(declared)
+    netbox_names = frozenset(_relation_names(netbox_value))
+    if yaml_names == netbox_names:
+        return None
+    return PropertyChange(property_name=prop, old_value=sorted(netbox_names), new_value=sorted(yaml_names))
+
+
+def _relation_identities(value):
+    """Return {(manufacturer slug, slug)} for a relation, or None if the shape lacks them.
+
+    A read path that returns only ``id`` and ``name`` cannot answer the scope question, so
+    the caller falls back to comparing names rather than inventing an identity.
+    """
+    identities = set()
+    for item in value or []:
+        slug = item.get("slug") if isinstance(item, dict) else getattr(item, "slug", None)
+        manufacturer = item.get("manufacturer") if isinstance(item, dict) else getattr(item, "manufacturer", None)
+        owner = manufacturer.get("slug") if isinstance(manufacturer, dict) else getattr(manufacturer, "slug", None)
+        if not slug or not owner:
+            return None
+        identities.add((owner, slug))
+    return frozenset(identities)
+
+
+def _relation_names(value):
+    """Return the ``name`` of each related object NetBox returned for a relation."""
+    names = []
+    for item in value or []:
+        name = item.get("name") if isinstance(item, dict) else getattr(item, "name", None)
+        if name is not None:
+            names.append(name)
+    return names
+
+
 @dataclass
 class PropertyChange:
     """Represents a single property change."""
@@ -42,7 +150,7 @@ class ComponentChange:
     component_type: str  # e.g., "interfaces", "power-ports"
     component_name: str
     change_type: ChangeType
-    property_changes: List[PropertyChange] = field(default_factory=list)
+    property_changes: list[PropertyChange] = field(default_factory=list)
 
 
 @dataclass
@@ -53,9 +161,9 @@ class DeviceTypeChange:
     model: str
     slug: str
     is_new: bool = False
-    property_changes: List[PropertyChange] = field(default_factory=list)
-    component_changes: List[ComponentChange] = field(default_factory=list)
-    netbox_id: Optional[int] = None
+    property_changes: list[PropertyChange] = field(default_factory=list)
+    component_changes: list[ComponentChange] = field(default_factory=list)
+    netbox_id: int | None = None
 
     @property
     def has_changes(self) -> bool:
@@ -72,8 +180,8 @@ def has_updates(self) -> bool:
 class ChangeReport:
     """Aggregated change report for all device types."""
 
-    new_device_types: List[DeviceTypeChange] = field(default_factory=list)
-    modified_device_types: List[DeviceTypeChange] = field(default_factory=list)
+    new_device_types: list[DeviceTypeChange] = field(default_factory=list)
+    modified_device_types: list[DeviceTypeChange] = field(default_factory=list)
     unchanged_count: int = 0
 
 
@@ -149,7 +257,7 @@ def __init__(self, device_types_instance, handle, remove_unmanaged_types: bool =
         self.verbose = verbose
         self.remove_unmanaged_types = remove_unmanaged_types
 
-    def detect_changes(self, device_types: List[dict], progress=None) -> ChangeReport:
+    def detect_changes(self, device_types: list[dict], progress=None) -> ChangeReport:
         """Analyze all device types and generate a change report.
 
         Args:
@@ -200,7 +308,7 @@ def detect_changes(self, device_types: List[dict], progress=None) -> ChangeRepor
 
         return report
 
-    def _compare_device_type_properties(self, yaml_data: dict, netbox_dt) -> List[PropertyChange]:
+    def _compare_device_type_properties(self, yaml_data: dict, netbox_dt) -> list[PropertyChange]:
         """Compare YAML device type properties against NetBox device type.
 
         Args:
@@ -241,7 +349,7 @@ def _compare_device_type_properties(self, yaml_data: dict, netbox_dt) -> List[Pr
         return changes
 
     @staticmethod
-    def _compare_image_properties(yaml_data: dict, netbox_dt) -> List[PropertyChange]:
+    def _compare_image_properties(yaml_data: dict, netbox_dt) -> list[PropertyChange]:
         """Compare image properties between YAML and NetBox device type.
 
         YAML uses boolean flags (front_image: true) meaning "an image should exist",
@@ -280,7 +388,7 @@ def _compare_components(
         yaml_data: dict,
         device_type_id: int,
         parent_type: str = "device",
-    ) -> List[ComponentChange]:
+    ) -> list[ComponentChange]:
         """Compare all components between YAML and cached NetBox data.
 
         Args:
@@ -310,9 +418,9 @@ def _compare_components(
             # same as an empty list so chassis YAMLs that omit (e.g.) interfaces can
             # still drive cleanup of stale templates in NetBox.
             if yaml_key in yaml_data or self.remove_unmanaged_types:
-                for existing_name in existing_components.keys():
+                for existing_name in existing_components:
                     if existing_name not in yaml_component_names:
-                        changes.append(
+                        changes.append(  # noqa: PERF401
                             ComponentChange(
                                 component_type=yaml_key,
                                 component_name=existing_name,
@@ -339,7 +447,11 @@ def _compare_components(
                     # Check for property changes on existing component
                     existing = existing_components[comp_name]
                     prop_changes = self._compare_component_properties(
-                        yaml_comp, existing, component.compare_properties, comp_type=yaml_key
+                        yaml_comp,
+                        existing,
+                        component.compare_properties,
+                        comp_type=yaml_key,
+                        manufacturer=_manufacturer_slug_of(yaml_data),
                     )
                     if prop_changes:
                         changes.append(
@@ -353,14 +465,31 @@ def _compare_components(
 
         return changes
 
+    def _relation_catalog(self):
+        """Return the module bay type catalog, or None when this detector has no real one.
+
+        Tests build the detector around a stand-in for DeviceTypes, and a stand-in cannot
+        answer what a reference means.  Returning None there keeps the name comparison,
+        which is what those tests are asserting on.
+        """
+        from core.module_bay_types import ModuleBayTypeCatalog
+
+        catalog = getattr(self.device_types, "module_bay_type_catalog", None)
+        return catalog if isinstance(catalog, ModuleBayTypeCatalog) else None
+
     def _compare_component_properties(
         self,
         yaml_comp: dict,
         netbox_comp,
-        properties: List[str],
+        properties: list[str],
         comp_type: str = "",
-    ) -> List[PropertyChange]:
-        """Compare properties between YAML and NetBox component."""
+        manufacturer: str = "",
+    ) -> list[PropertyChange]:
+        """Compare properties between YAML and NetBox component.
+
+        *manufacturer* is the owning manufacturer's slug, used to resolve a relation
+        reference to the object it means rather than the name it is written as.
+        """
         changes = []
 
         for prop in properties:
@@ -389,7 +518,11 @@ def _compare_component_properties(
                     # GraphQL response lacked both mappings and rear_port_position;
                     # treat as unmanaged to avoid a false COMPONENT_CHANGED.
                     continue
-                has_names = any(m.get("rear_port_name") is not None for m in canonical)
+                # Compare by name whenever one is available: an empty M2M list has no name to
+                # infer from but still needs the named path, and the pre-4.5 query returns one.
+                has_names = bool(getattr(netbox_comp, "_mappings_m2m", False)) or any(
+                    m.get("rear_port_name") is not None for m in canonical
+                )
                 if has_names:
                     # NetBox >= 4.5: compare with rear port names
                     netbox_set: frozenset = frozenset(
@@ -426,6 +559,14 @@ def _compare_component_properties(
                     )
                 continue
 
+            if prop in _relation_properties(comp_type):
+                relation_change = _relation_change(
+                    prop, yaml_comp, netbox_comp, self._relation_catalog(), manufacturer, self.handle
+                )
+                if relation_change is not None:
+                    changes.append(relation_change)
+                continue
+
             # Only compare properties explicitly present in the YAML component;
             # an omitted property means the YAML doesn't manage it (absent key != removal).
             if prop not in yaml_comp:
@@ -490,7 +631,7 @@ def _log_modified_summary(self, report: ChangeReport) -> None:
         if parts:
             self.handle.log(f"  Breakdown: {', '.join(parts)}")
 
-    def _log_property_diffs(self, prop_changes: List[PropertyChange], indent: str) -> None:
+    def _log_property_diffs(self, prop_changes: list[PropertyChange], indent: str) -> None:
         """Emit diff-u style lines for *prop_changes* at the given *indent*."""
         log_property_diffs(
             [(pc.property_name, pc.old_value, pc.new_value) for pc in prop_changes],

core/compat.py

@@ -1,61 +1,27 @@
 """NetBox version-compatibility helpers.
 
-Centralises filter parameter names that changed between NetBox releases so
-that every caller uses the same logic and drift is impossible.
-
-NetBox 4.1 renamed several filter keys on the DCIM endpoints:
-
-    devicetype_id  →  device_type_id
-    moduletype_id  →  module_type_id
-
-Any code that constructs endpoint filter kwargs should call the helpers
-here rather than inlining ``"device_type_id" if new_filters else "devicetype_id"``.
+Centralises the version line each feature sits behind so that every caller uses the
+same logic and drift is impossible.
 """
 
 from __future__ import annotations
 
+import re
 
-def device_type_filter_key(new_filters: bool) -> str:
-    """Return the correct filter parameter name for device-type component queries.
+# Selecting or sending the relation below this release fails the whole query.
+MODULE_BAY_TYPE_MINIMUM_VERSION = (4, 7)
 
-    Args:
-        new_filters: ``True`` for NetBox ≥ 4.1 (returns ``"device_type_id"``);
-            ``False`` for older releases (returns ``"devicetype_id"``).
-    """
-    return "device_type_id" if new_filters else "devicetype_id"
 
+def parse_netbox_version(version) -> tuple[int, int]:
+    """Return ``(major, minor)`` from a NetBox version string.
 
-def module_type_filter_key(new_filters: bool) -> str:
-    """Return the correct filter parameter name for module-type component queries.
-
-    Args:
-        new_filters: ``True`` for NetBox ≥ 4.1 (returns ``"module_type_id"``);
-            ``False`` for older releases (returns ``"moduletype_id"``).
+    Padded to two parts so a single-component string cannot raise, and tolerant of the
+    suffixes NetBox ships ("4.7.0-beta2").
     """
-    return "module_type_id" if new_filters else "moduletype_id"
-
-
-def device_type_filter_kwargs(device_type_id: int, *, new_filters: bool) -> dict:
-    """Return filter kwargs for querying components of a device type.
+    raw = [int(re.sub(r"\D.*", "", part.strip()) or "0") for part in str(version).split(".")]
+    return tuple(([*raw, 0, 0])[:2])  # type: ignore[return-value]
 
-    Args:
-        device_type_id: NetBox ID of the device type.
-        new_filters: ``True`` for NetBox ≥ 4.1; ``False`` for older releases.
 
-    Returns:
-        A dict suitable for unpacking into ``endpoint.filter(**kwargs)``.
-    """
-    return {device_type_filter_key(new_filters): device_type_id}
-
-
-def module_type_filter_kwargs(module_type_id: int, *, new_filters: bool) -> dict:
-    """Return filter kwargs for querying components of a module type.
-
-    Args:
-        module_type_id: NetBox ID of the module type.
-        new_filters: ``True`` for NetBox ≥ 4.1; ``False`` for older releases.
-
-    Returns:
-        A dict suitable for unpacking into ``endpoint.filter(**kwargs)``.
-    """
-    return {module_type_filter_key(new_filters): module_type_id}
+def supports_module_bay_types(version) -> bool:
+    """Return True when this NetBox release exposes the module bay type relation."""
+    return parse_netbox_version(version) >= MODULE_BAY_TYPE_MINIMUM_VERSION