addresses #156

Author: jtdub

hier_config/platforms/driver_base.py

@@ -1,5 +1,6 @@
 from abc import ABC, abstractmethod
 from collections.abc import Callable, Iterable
+from re import Match, search
 
 from pydantic import Field, PositiveInt
 
@@ -10,6 +11,7 @@
     IdempotentCommandsAvoidRule,
     IdempotentCommandsRule,
     IndentAdjustRule,
+    MatchRule,
     NegationDefaultWhenRule,
     NegationDefaultWithRule,
     OrderingRule,
@@ -133,10 +135,18 @@ def idempotent_for(
         other_children: Iterable[HConfigChild],
     ) -> HConfigChild | None:
         for rule in self.rules.idempotent_commands:
-            if config.is_lineage_match(rule.match_rules):
-                for other_child in other_children:
-                    if other_child.is_lineage_match(rule.match_rules):
-                        return other_child
+            if not config.is_lineage_match(rule.match_rules):
+                continue
+
+            config_key = self._idempotency_key(config, rule.match_rules)
+
+            for other_child in other_children:
+                if not other_child.is_lineage_match(rule.match_rules):
+                    continue
+
+                if self._idempotency_key(other_child, rule.match_rules) == config_key:
+                    return other_child
+
         return None
 
     def negate_with(self, config: HConfigChild) -> str | None:
@@ -154,6 +164,171 @@ def swap_negation(self, child: HConfigChild) -> HConfigChild:
 
         return child
 
+    def _idempotency_key(
+        self,
+        config: HConfigChild,
+        match_rules: tuple[MatchRule, ...],
+    ) -> tuple[str, ...]:
+        lineage = tuple(config.lineage())
+        if len(lineage) != len(match_rules):
+            return ()
+
+        return tuple(map(self._idempotency_component_key, lineage, match_rules))
+
+    def _idempotency_component_key(
+        self,
+        child: HConfigChild,
+        rule: MatchRule,
+    ) -> str:
+        text = child.text
+        normalized_text = text.removeprefix(self.negation_prefix)
+
+        parts: list[str] = []
+        parts.extend(self._key_from_equals(rule.equals, text))
+        parts.extend(self._key_from_prefix(rule.startswith, normalized_text))
+        parts.extend(self._key_from_suffix(rule.endswith, normalized_text))
+        parts.extend(self._key_from_contains(rule.contains, normalized_text))
+        parts.extend(self._key_from_regex(rule.re_search, normalized_text, text))
+
+        if not parts:
+            parts.append(f"text|{normalized_text}")
+
+        return ";".join(parts)
+
+    @staticmethod
+    def _key_from_equals(
+        equals: str | frozenset[str] | None, text: str
+    ) -> list[str]:
+        if equals is None:
+            return []
+        if isinstance(equals, str):
+            return [f"equals|{equals}"]
+        return [f"equals|{text}"]
+
+    def _key_from_prefix(
+        self,
+        prefix: str | tuple[str, ...] | None,
+        normalized_text: str,
+    ) -> list[str]:
+        if prefix is None:
+            return []
+        matched = self._match_prefix(normalized_text, prefix)
+        if matched is None:
+            return []
+        return [f"startswith|{matched}"]
+
+    def _key_from_suffix(
+        self,
+        suffix: str | tuple[str, ...] | None,
+        normalized_text: str,
+    ) -> list[str]:
+        if suffix is None:
+            return []
+        matched = self._match_suffix(normalized_text, suffix)
+        if matched is None:
+            return []
+        return [f"endswith|{matched}"]
+
+    def _key_from_contains(
+        self,
+        contains: str | tuple[str, ...] | None,
+        normalized_text: str,
+    ) -> list[str]:
+        if contains is None:
+            return []
+        matched = self._match_contains(normalized_text, contains)
+        if matched is None:
+            return []
+        return [f"contains|{matched}"]
+
+    def _key_from_regex(
+        self,
+        pattern: str | None,
+        normalized_text: str,
+        original_text: str,
+    ) -> list[str]:
+        if pattern is None:
+            return []
+
+        match = search(pattern, normalized_text)
+        match_source = normalized_text
+        if match is None:
+            match = search(pattern, original_text)
+            match_source = original_text
+
+        if match is None:
+            return []
+
+        regex_key = self._normalize_regex_key(pattern, match_source, match)
+        return [f"re|{regex_key}"]
+
+    @staticmethod
+    def _match_prefix(value: str, prefix: str | tuple[str, ...]) -> str | None:
+        if isinstance(prefix, tuple):
+            matches = [candidate for candidate in prefix if value.startswith(candidate)]
+            if matches:
+                return max(matches, key=len)
+            return None
+
+        if value.startswith(prefix):
+            return prefix
+
+        return None
+
+    @staticmethod
+    def _match_suffix(value: str, suffix: str | tuple[str, ...]) -> str | None:
+        if isinstance(suffix, tuple):
+            matches = [candidate for candidate in suffix if value.endswith(candidate)]
+            if matches:
+                return max(matches, key=len)
+            return None
+
+        if value.endswith(suffix):
+            return suffix
+
+        return None
+
+    @staticmethod
+    def _match_contains(
+        value: str, contains: str | tuple[str, ...]
+    ) -> str | None:
+        if isinstance(contains, tuple):
+            matches = [candidate for candidate in contains if candidate in value]
+            if matches:
+                return max(matches, key=len)
+            return None
+
+        if contains in value:
+            return contains
+
+        return None
+
+    @staticmethod
+    def _normalize_regex_key(pattern: str, value: str, match: Match[str]) -> str:
+        result = match.group(0)
+
+        if match.re.groups:
+            groups = tuple(g or "" for g in match.groups())
+            if any(groups):
+                normalized_groups = tuple(group.strip() for group in groups)
+                if any(normalized_groups):
+                    return "|".join(normalized_groups)
+
+        trimmed_pattern = pattern.rstrip("$")
+        for suffix in (".*", ".+"):
+            if trimmed_pattern.endswith(suffix):
+                candidate_pattern = trimmed_pattern[: -len(suffix)]
+                if not candidate_pattern:
+                    break
+                trimmed_match = search(candidate_pattern, value)
+                if trimmed_match is not None:
+                    candidate = trimmed_match.group(0).strip()
+                    if candidate:
+                        return candidate
+                break
+
+        return result.strip()
+
     @property
     def declaration_prefix(self) -> str:
         return ""

tests/test_hier_config.py

@@ -485,6 +485,42 @@ def test_future_config(platform_a: Platform) -> None:
     )
 
 
+def test_future_preserves_bgp_neighbor_description() -> None:
+    platform = Platform.ARISTA_EOS
+    running_raw = """router bgp 1
+  neighbor 2.2.2.2 description neighbor2
+  neighbor 2.2.2.2 remote-as 2
+  !
+"""
+    change_raw = """router bgp 1
+  neighbor 3.3.3.3 description neighbor3
+  neighbor 3.3.3.3 remote-as 3
+"""
+
+    running_config = get_hconfig(platform, running_raw)
+    change_config = get_hconfig(platform, change_raw)
+
+    future_config = running_config.future(change_config)
+    expected_future = (
+        "router bgp 1",
+        "  neighbor 3.3.3.3 description neighbor3",
+        "  neighbor 3.3.3.3 remote-as 3",
+        "  neighbor 2.2.2.2 description neighbor2",
+        "  neighbor 2.2.2.2 remote-as 2",
+        "  exit",
+    )
+    assert future_config.dump_simple(sectional_exiting=True) == expected_future
+
+    rollback_config = future_config.config_to_get_to(running_config)
+    expected_rollback = (
+        "router bgp 1",
+        "  no neighbor 3.3.3.3 description neighbor3",
+        "  no neighbor 3.3.3.3 remote-as 3",
+        "  exit",
+    )
+    assert rollback_config.dump_simple(sectional_exiting=True) == expected_rollback
+
+
 def test_difference1(platform_a: Platform) -> None:
     rc = ("a", " a1", " a2", " a3", "b")
     step = ("a", " a1", " a2", " a3", " a4", " a5", "b", "c", "d", " d1")