diff --git a/docs/adr/0013-centralize-template-claim-ambiguity.md b/docs/adr/0013-centralize-template-claim-ambiguity.md new file mode 100644 index 00000000..2f516271 --- /dev/null +++ b/docs/adr/0013-centralize-template-claim-ambiguity.md @@ -0,0 +1,35 @@ +--- +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. + +The engine and installed-family discovery both use the primitive. +Installed-family discovery passes the complete historical base relation and emits +collision messages through its own logger with the label kind `family base`. +Current resolved bases remain unconditional and precede accepted historical bases. +Historical base extraction and duplicate handling remain unchanged. +The `_singly_claimed` rule counts member primary keys across complete family +candidates and remains separate. diff --git a/netbox_interface_name_rules/engine.py b/netbox_interface_name_rules/engine.py index 5fb43abc..97ffee4b 100644 --- a/netbox_interface_name_rules/engine.py +++ b/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 diff --git a/netbox_interface_name_rules/family/__init__.py b/netbox_interface_name_rules/family/__init__.py index e73cddcc..5c3519e2 100644 --- a/netbox_interface_name_rules/family/__init__.py +++ b/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", diff --git a/netbox_interface_name_rules/family/claims.py b/netbox_interface_name_rules/family/claims.py new file mode 100644 index 00000000..f4f52084 --- /dev/null +++ b/netbox_interface_name_rules/family/claims.py @@ -0,0 +1,65 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright (C) 2025 Marcin Zieba +"""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) diff --git a/netbox_interface_name_rules/family/conversion.py b/netbox_interface_name_rules/family/conversion.py index 1d12b675..538ae280 100644 --- a/netbox_interface_name_rules/family/conversion.py +++ b/netbox_interface_name_rules/family/conversion.py @@ -126,7 +126,7 @@ def plan_module_conversions( channelization_supported = supports_channelization() plans = [] claimed = set() - for base_name, source_base in flat_family_bases(rule, variables, interfaces, catalog): + for base_name, source_base in flat_family_bases(module, rule, variables, interfaces, catalog): names = family_names_for(rule, variables, base_name, source_base) if names is None: continue diff --git a/netbox_interface_name_rules/family/installed.py b/netbox_interface_name_rules/family/installed.py index ad34d3bb..0c255879 100644 --- a/netbox_interface_name_rules/family/installed.py +++ b/netbox_interface_name_rules/family/installed.py @@ -9,6 +9,7 @@ from ..choices import BreakoutModeChoices from ..naming import evaluate_name_template +from .claims import TemplateClaim, resolve_template_claims from .domain import ( FamilyStatus, FamilyTopology, @@ -73,32 +74,19 @@ def _historical_bases(rule, variables, template, interfaces): # pragma: no cove return tuple(sorted(bases)) -def _ambiguous_bases(historical_by_template): - """Return every historical base that more than one template could claim.""" - ambiguous = {base_name for bases in historical_by_template.values() if len(bases) > 1 for base_name in bases} - single_claims: dict[str, int] = {} - for bases in historical_by_template.values(): - if len(bases) == 1: # pragma: no cover - historical matchers require VC token support - single_claims[bases[0]] = single_claims.get(bases[0], 0) + 1 - ambiguous.update(base_name for base_name, count in single_claims.items() if count > 1) - return ambiguous +def _source_bases(template, historical_bases): + """Return the template's own base and its accepted historical bases.""" + return (template.resolved, *historical_bases) -def _source_bases(template, historical_bases, ambiguous_bases): - """Return the template's own base and every historical base it alone claims.""" - unambiguous = tuple( - base_name for base_name in historical_bases if len(historical_bases) == 1 and base_name not in ambiguous_bases - ) - return (template.resolved, *unambiguous) - - -def flat_family_bases(rule, variables, interfaces, catalog): +def flat_family_bases(module, rule, variables, interfaces, catalog): """Return ``(template base, source base)`` for every base a flat family could be named from. The template base is the name the rule resolves for this module now; the source base is the one - an installed family still spells, which differs after a virtual-chassis renumber. A historical - base more than one template could claim is dropped: the rows it names are not certainly one - family's, and neither renaming nor converting them is this plugin's guess to make. + an installed family still spells, which differs after a virtual-chassis renumber. + A historical base that more than one template could claim is dropped. + Every historical base of a template that claims multiple bases is also dropped. + These claims do not identify one family with certainty, so the plugin does not rename or convert them. """ if rule.channel_count <= 0: return () @@ -106,11 +94,21 @@ def flat_family_bases(rule, variables, interfaces, catalog): historical_by_template = { template.pk: _historical_bases(rule, variables, template, interfaces) for template in templates } - ambiguous_bases = _ambiguous_bases(historical_by_template) + accepted, messages = resolve_template_claims( + tuple( + TemplateClaim(template.pk, template.template_name, historical_by_template[template.pk]) + for template in templates + ), + module=module, + label_kind="family base", + ) + for message in messages: + logger.warning(message) + accepted_by_template = {pk: (base,) for pk, base in accepted} return tuple( (template.resolved, source_base) for template in templates - for source_base in _source_bases(template, historical_by_template[template.pk], ambiguous_bases) + for source_base in _source_bases(template, accepted_by_template.get(template.pk, ())) ) @@ -135,7 +133,7 @@ def _singly_claimed(candidates): return [candidate for candidate in candidates if all(claims[member.pk] == 1 for member in candidate[2])] -def flat_family_candidates(rule, variables, interfaces, catalog): +def flat_family_candidates(module, rule, variables, interfaces, catalog): """Return complete, unambiguous flat-family candidates on this module. A flat family carries the names the rule's channel range spells, and a flat rule and the @@ -146,7 +144,7 @@ def flat_family_candidates(rule, variables, interfaces, catalog): if not by_name: return [] candidates = [] - for base_name, source_base in flat_family_bases(rule, variables, interfaces, catalog): + for base_name, source_base in flat_family_bases(module, rule, variables, interfaces, catalog): names = family_names_for(rule, variables, base_name, source_base) if names is None: continue @@ -159,11 +157,11 @@ def flat_family_candidates(rule, variables, interfaces, catalog): return _singly_claimed(candidates) -def _flat_candidates(rule, variables, interfaces, catalog): +def _flat_candidates(module, rule, variables, interfaces, catalog): """Return the flat families a flat-mode rule owns on this module.""" if rule.breakout_mode != BreakoutModeChoices.FLAT: return [] - return flat_family_candidates(rule, variables, interfaces, catalog) + return flat_family_candidates(module, rule, variables, interfaces, catalog) def _flat_plan(module, target_names, interfaces): @@ -346,7 +344,7 @@ def plan_installed_families(module, rule, variables, interfaces=None) -> Install plans = _channelized_plans(module, rule, variables, interfaces, catalog) plans.extend( _flat_plan(module, target_names, members) - for _base_name, target_names, members in _flat_candidates(rule, variables, interfaces, catalog) + for _base_name, target_names, members in _flat_candidates(module, rule, variables, interfaces, catalog) ) plans.sort(key=lambda plan: plan.member_pks[0]) return InstalledFamilyPlanSet(module_id=module.pk, plans=tuple(plans)) diff --git a/netbox_interface_name_rules/tests/test_family_claims.py b/netbox_interface_name_rules/tests/test_family_claims.py new file mode 100644 index 00000000..79849ef1 --- /dev/null +++ b/netbox_interface_name_rules/tests/test_family_claims.py @@ -0,0 +1,111 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright (C) 2025 Marcin Zieba +"""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) diff --git a/netbox_interface_name_rules/tests/test_vc_drift.py b/netbox_interface_name_rules/tests/test_vc_drift.py index 4223385d..59c6181f 100644 --- a/netbox_interface_name_rules/tests/test_vc_drift.py +++ b/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. @@ -790,7 +810,46 @@ def test_a_family_matcher_that_captures_two_bases_is_not_offered(self): self._join(VirtualChassis.objects.create(name="vcconv-vc2"), 5, device=standalone) self._switch_to_channelized(rule) - self.assertEqual(find_convertible_families(rule).candidates, ()) + with self.assertLogs("netbox_interface_name_rules.family.installed", level="WARNING") as logs: + self.assertEqual(find_convertible_families(rule).candidates, ()) + + warnings = " ".join(logs.output) + self.assertIn("Family base", warnings) + self.assertIn("xe-{vc_position:0}/0/{module}", warnings) + self.assertIn("xe-{vc_position:9}/0/{module}", warnings) + self.assertIn(str(module), warnings) + + def test_an_unrelated_family_survives_overlapping_historical_claims(self): + """A multi-base claim rejects its shared base but leaves an unrelated family available.""" + module_type = _token_module_type( + self.manufacturer, + "VcConv-OVERLAP", + "xe-{vc_position}/0/{module}", + "xe-1/{vc_position}/{module}", + "et-{vc_position}/0/{module}", + ) + rule = self._flat_rule(module_type, "brk-{base}:{channel}") + module, _ = self._install_on(self.device, module_type, "3") + for channel in range(4): + rename_out_of_band( + Interface.objects.get(module=module, name=f"brk-xe-1/1/3:{channel}"), + f"brk-xe-2/0/3:{channel}", + ) + self._renumber(5) + self._switch_to_channelized(rule) + + with self.assertLogs("netbox_interface_name_rules.family.installed", level="WARNING") as logs: + candidates = find_convertible_families(rule).candidates + + self.assertEqual(len(candidates), 1) + self.assertTrue(candidates[0].convertible, candidates[0].reason) + self.assertEqual(list(candidates[0].current_names), [f"brk-et-1/0/3:{channel}" for channel in range(4)]) + self.assertEqual(candidates[0].new_names[0], "et-0/0/3") + self.assertIn( + f"Family base 'xe-1/0/3' on {module} could be the drifted name of any of the templates " + "['xe-1/{vc_position}/{module}', 'xe-{vc_position}/0/{module}']", + " ".join(logs.output), + ) def test_a_rule_without_a_base_is_identified_after_a_renumber(self): """Drift-immune by construction — asserted, not assumed, so the fix cannot regress it."""