refactor: centralize template claim ambiguity

docs/adr/0013-centralize-template-claim-ambiguity.md

@@ -0,0 +1,32 @@
+---
+status: accepted
+---
+
+# Centralize template claim ambiguity
+
+The engine and installed-family discovery each implemented the same two-sided uniqueness rule.
+Exhaustive checks over 4096 relations established their equivalence.
+The family package owns this rule through immutable `TemplateClaim` values and
+`resolve_template_claims`, exported at the package boundary.
+
+The primitive accepts the complete relation for one selection scope.
+Callers must retain claims from templates that claim multiple labels.
+Repeated edges count once, and duplicate claimant IDs are invalid.
+A template with multiple labels or a label with multiple templates disqualifies
+every involved claim. Accepted pairs retain input claimant order.
+The primitive returns accepted pairs and rendered collision messages.
+It uses only the standard library and does not discover candidates.
+
+The engine retains regex matching, comparison forms, exact-name precedence,
+forced-base ordering and the preference for channel `:0`.
+Its admission guard stays at the same point in `_collect_unrenamed`, before
+interfaces that intend one family collapse, as required by ADR 0011.
+
+Message construction belongs to the primitive so callers cannot drift in wording.
+The primitive does not log. Callers emit its messages through their own logger,
+which preserves the engine warning source and keeps the decision free of side effects.
+
+Increment 1 adopts the primitive only in the engine.
+Installed-family adoption belongs to increment 2.
+The `_singly_claimed` rule counts member primary keys across complete family
+candidates and remains separate.

netbox_interface_name_rules/engine.py

@@ -87,40 +87,19 @@ def supports_vc_position_token():
 
 
 def _unambiguous_claims(candidates, matchers, module):  # pragma: no cover - requires vc_position token support
-    """Return the labels of *candidates* that exactly one drifted ``{vc_position}`` template claims.
-
-    *candidates* pairs a label with the name forms it is compared under.  Both sides of the claim
-    have to be unique: a template matching two labels, or a label matched by two templates,
-    disqualifies everything involved with a warning rather than renaming a guess.
-    """
-    claims = defaultdict(list)
-    claimants = defaultdict(list)
-    for index, matcher in enumerate(matchers):
-        for label, forms in candidates:
-            if any(matcher.pattern.fullmatch(form) for form in forms):
-                claims[index].append(label)
-                claimants[label].append(index)
-
-    ambiguous = {index for index, claimed in claims.items() if len(claimed) > 1}
-    for index in sorted(ambiguous):
-        logger.warning(
-            "Interface template %r of %s could name any of %s since this device's virtual-chassis "
-            "position changed; skipping them all rather than renaming a guess.",
-            matchers[index].template_name,
-            module,
-            sorted(claims[index]),
+    """Build drift claims and delegate admission to the family package."""
+    claims = tuple(
+        family_ops.TemplateClaim(
+            index,
+            matcher.template_name,
+            tuple(label for label, forms in candidates if any(matcher.pattern.fullmatch(form) for form in forms)),
         )
-    for label, indexes in claimants.items():
-        if len(indexes) > 1:
-            logger.warning(
-                "Interface %r on %s could be the drifted name of any of the templates %s; "
-                "skipping it rather than renaming a guess.",
-                label,
-                module,
-                sorted(matchers[index].template_name for index in indexes),
-            )
-            ambiguous.update(indexes)
-    return [claims[index][0] for index in sorted(claims) if index not in ambiguous]
+        for index, matcher in enumerate(matchers)
+    )
+    accepted, messages = family_ops.resolve_template_claims(claims, module=module, label_kind="interface name")
+    for message in messages:
+        logger.warning("%s", message)
+    return [label for _, label in accepted]
 
 
 def _drifted_candidates(interfaces, matchers, module):  # pragma: no cover - requires vc_position token support

netbox_interface_name_rules/family/__init__.py

@@ -11,6 +11,7 @@
     plan_module_families,
 )
 from .capabilities import supports_channelization
+from .claims import TemplateClaim, resolve_template_claims
 from .conversion import (
     conversion_offered,
     convert_rule_families,
@@ -93,6 +94,7 @@
     "ProspectiveInterface",
     "ProspectiveMember",
     "StructuralFamilyPlan",
+    "TemplateClaim",
     "apply_rule_to_modules",
     "channelized_family_names",
     "conversion_offered",
@@ -123,6 +125,7 @@
     "plan_prospective_families",
     "plan_structural_family",
     "preview_rule_conversions",
+    "resolve_template_claims",
     "resolved_template_names",
     "supports_channelization",
     "template_channel_suffixes",

netbox_interface_name_rules/family/claims.py

@@ -0,0 +1,65 @@
+# SPDX-License-Identifier: Apache-2.0
+# Copyright (C) 2025 Marcin Zieba <marcinpsk@gmail.com>
+"""Resolve two-sided uniqueness for complete template claim relations."""
+
+from collections import defaultdict
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True, slots=True)
+class TemplateClaim:
+    """Labels claimed by one template in a selection scope."""
+
+    claimant_id: int
+    template_name: str
+    labels: tuple[str, ...]
+
+    def __post_init__(self):
+        object.__setattr__(self, "labels", tuple(self.labels))
+
+
+def resolve_template_claims(claims, *, module, label_kind):
+    """Return accepted pairs in claimant order and rendered collision messages.
+
+    Pass the complete relation, including templates with multiple labels.
+    Repeated edges count once. Claimant IDs must be unique.
+    The caller emits the messages through its own logger.
+    """
+    if label_kind not in ("interface name", "family base"):
+        raise ValueError("label_kind must be 'interface name' or 'family base'")
+
+    by_id = {}
+    claimants = defaultdict(list)
+    ambiguous = set()
+    messages = []
+    for claim in claims:
+        if claim.claimant_id in by_id:
+            raise ValueError(f"Duplicate claimant_id: {claim.claimant_id}")
+        labels = tuple(dict.fromkeys(claim.labels))
+        by_id[claim.claimant_id] = (claim.template_name, labels)
+        for label in labels:
+            claimants[label].append(claim.claimant_id)
+        if len(labels) > 1:
+            ambiguous.add(claim.claimant_id)
+            messages.append(
+                f"Interface template {claim.template_name!r} of {module} could name any of {sorted(labels)} "
+                "since this device's virtual-chassis position changed; "
+                "skipping them all rather than renaming a guess."
+            )
+
+    subject = "Interface" if label_kind == "interface name" else "Family base"
+    for label, ids in claimants.items():
+        if len(ids) > 1:
+            messages.append(
+                f"{subject} {label!r} on {module} could be the drifted name of any of the templates "
+                f"{sorted(by_id[claimant_id][0] for claimant_id in ids)}; "
+                "skipping it rather than renaming a guess."
+            )
+            ambiguous.update(ids)
+
+    accepted = tuple(
+        (claimant_id, labels[0])
+        for claimant_id, (_, labels) in by_id.items()
+        if labels and claimant_id not in ambiguous
+    )
+    return accepted, tuple(messages)

netbox_interface_name_rules/tests/test_family_claims.py

@@ -0,0 +1,111 @@
+# SPDX-License-Identifier: Apache-2.0
+# Copyright (C) 2025 Marcin Zieba <marcinpsk@gmail.com>
+"""Tests for complete template claim relations."""
+
+from django.test import SimpleTestCase
+
+from netbox_interface_name_rules import family
+
+
+class TemplateClaimsTest(SimpleTestCase):
+    def test_claim_copies_labels_from_a_mutable_list(self):
+        labels = ["x"]
+        claim = family.TemplateClaim(1, "A", labels)
+        self.assertIsInstance(claim.labels, tuple)
+        labels.append("y")
+        self.assertEqual(claim.labels, ("x",))
+
+    def test_empty_relation(self):
+        self.assertEqual(
+            family.resolve_template_claims((), module="module", label_kind="interface name"),
+            ((), ()),
+        )
+
+    def test_duplicate_edges_count_once_and_keep_claimant_order(self):
+        claims = (
+            family.TemplateClaim(9, "A", ("z", "z")),
+            family.TemplateClaim(2, "B", ()),
+            family.TemplateClaim(1, "C", ("x",)),
+        )
+        self.assertEqual(
+            family.resolve_template_claims(iter(claims), module="module", label_kind="interface name"),
+            (((9, "z"), (1, "x")), ()),
+        )
+
+    def test_claimant_with_two_labels_is_rejected(self):
+        claims = (family.TemplateClaim(1, "A", ("y", "x", "y")),)
+        self.assertEqual(
+            family.resolve_template_claims(claims, module="module", label_kind="interface name"),
+            (
+                (),
+                (
+                    (
+                        "Interface template 'A' of module could name any of ['x', 'y'] since this device's "
+                        "virtual-chassis position changed; skipping them all rather than renaming a guess."
+                    ),
+                ),
+            ),
+        )
+
+    def test_label_with_two_claimants_is_rejected(self):
+        claims = (family.TemplateClaim(1, "B", ("x", "x")), family.TemplateClaim(2, "A", ("x",)))
+        for label_kind, subject in (("interface name", "Interface"), ("family base", "Family base")):
+            with self.subTest(label_kind=label_kind):
+                self.assertEqual(
+                    family.resolve_template_claims(claims, module="module", label_kind=label_kind),
+                    (
+                        (),
+                        (
+                            (
+                                f"{subject} 'x' on module could be the drifted name of any of the templates "
+                                "['A', 'B']; skipping it rather than renaming a guess."
+                            ),
+                        ),
+                    ),
+                )
+
+    def test_complete_mixed_relation_rejects_shared_label(self):
+        claims = (
+            family.TemplateClaim(1, "A", ("x", "y")),
+            family.TemplateClaim(2, "B", ("y",)),
+            family.TemplateClaim(3, "C", ("z",)),
+        )
+        self.assertEqual(
+            family.resolve_template_claims(claims, module="module", label_kind="family base"),
+            (
+                ((3, "z"),),
+                (
+                    (
+                        "Interface template 'A' of module could name any of ['x', 'y'] since this device's "
+                        "virtual-chassis position changed; skipping them all rather than renaming a guess."
+                    ),
+                    (
+                        "Family base 'y' on module could be the drifted name of any of the templates "
+                        "['A', 'B']; skipping it rather than renaming a guess."
+                    ),
+                ),
+            ),
+        )
+
+    def test_duplicate_claimant_id_raises(self):
+        claims = (family.TemplateClaim(1, "A", ()), family.TemplateClaim(1, "B", ("x",)))
+        with self.assertRaisesRegex(ValueError, "Duplicate claimant_id"):
+            family.resolve_template_claims(claims, module="module", label_kind="interface name")
+
+    def test_invalid_label_kind_raises(self):
+        with self.assertRaisesRegex(ValueError, "label_kind"):
+            family.resolve_template_claims((), module="module", label_kind="port")
+
+    def test_claim_is_immutable(self):
+        from dataclasses import FrozenInstanceError
+
+        claim = family.TemplateClaim(1, "A", ("x",))
+        with self.assertRaises(FrozenInstanceError):
+            claim.claimant_id = 2
+
+    def test_primitive_does_not_log_collisions(self):
+        claims = (family.TemplateClaim(1, "A", ("x", "y")),)
+        with self.assertNoLogs("netbox_interface_name_rules", level="WARNING"):
+            accepted, messages = family.resolve_template_claims(claims, module="module", label_kind="interface name")
+        self.assertEqual(accepted, ())
+        self.assertEqual(len(messages), 1)

netbox_interface_name_rules/tests/test_vc_drift.py

@@ -356,6 +356,26 @@ def test_a_matcher_that_claims_two_interfaces_renames_neither(self):
         for candidate in ("xe-0/0/3", "xe-1/0/3"):
             self.assertIn(candidate, output)
 
+    def test_drift_warning_uses_the_engine_logger(self):
+        module, bay = self._install_on(self.device, self.decoy_type, "3")
+        rename_out_of_band(Interface.objects.get(module=module, name="mgmt-3"), "xe-0/0/3")
+        self._renumber(2)
+        InterfaceNameRule.objects.create(module_type=self.decoy_type, name_template="et-{base}")
+
+        with self.assertLogs(engine.logger, level="WARNING") as logs:
+            renamed = apply_interface_name_rules(module, bay)
+
+        self.assertEqual(renamed, 0)
+        self.assertEqual(self._names(module), ["xe-0/0/3", "xe-1/0/3"])
+        self.assertEqual(len(logs.records), 1)
+        self.assertEqual(logs.records[0].name, engine.__name__)
+        self.assertEqual(
+            logs.records[0].getMessage(),
+            f"Interface template 'xe-{{vc_position:0}}/0/{{module}}' of {module} could name any of "
+            "['xe-0/0/3', 'xe-1/0/3'] since this device's virtual-chassis position changed; "
+            "skipping them all rather than renaming a guess.",
+        )
+
     def test_a_forced_re_apply_does_not_break_out_an_ambiguous_pair(self):
         """The same claim reached through the force path, where distinct targets hide the collision.