feat(anta): Added the test case to verify Update error counters for B…

…GP neighbors (#775)

anta/custom_types.py

@@ -167,3 +167,4 @@ def validate_regex(value: str) -> str:
 Hostname = Annotated[str, Field(pattern=REGEXP_TYPE_HOSTNAME)]
 Port = Annotated[int, Field(ge=1, le=65535)]
 RegexString = Annotated[str, AfterValidator(validate_regex)]
+BgpUpdateError = Literal["inUpdErrWithdraw", "inUpdErrIgnore", "inUpdErrDisableAfiSafi", "disabledAfiSafi", "lastUpdErrTime"]

anta/tests/routing/bgp.py

@@ -14,7 +14,7 @@
 from pydantic.v1.utils import deep_update
 from pydantic_extra_types.mac_address import MacAddress
 
-from anta.custom_types import Afi, MultiProtocolCaps, Safi, Vni
+from anta.custom_types import Afi, BgpUpdateError, MultiProtocolCaps, Safi, Vni
 from anta.models import AntaCommand, AntaTemplate, AntaTest
 from anta.tools import get_item, get_value
 
@@ -1226,3 +1226,94 @@ def test(self) -> None:
             self.result.is_success()
         else:
             self.result.is_failure(f"Following BGP peers are not configured or hold and keep-alive timers are not correct:\n{failures}")
+
+
+class VerifyBGPPeerUpdateErrors(AntaTest):
+    """Verifies BGP update error counters for the provided BGP IPv4 peer(s).
+
+    By default, all update error counters will be checked for any non-zero values.
+    An optional list of specific update error counters can be provided for granular testing.
+
+    Note: For "disabledAfiSafi" error counter field, checking that it's not "None" versus 0.
+
+    Expected Results
+    ----------------
+    * Success: The test will pass if the BGP peer's update error counter(s) are zero/None.
+    * Failure: The test will fail if the BGP peer's update error counter(s) are non-zero/not None/Not Found or
+    peer is not configured.
+
+    Examples
+    --------
+    ```yaml
+    anta.tests.routing:
+      bgp:
+        - VerifyBGPPeerUpdateErrors:
+            bgp_peers:
+              - peer_address: 172.30.11.1
+                vrf: default
+                update_error_filter:
+                  - inUpdErrWithdraw
+    ```
+    """
+
+    name = "VerifyBGPPeerUpdateErrors"
+    description = "Verifies the update error counters of a BGP IPv4 peer."
+    categories: ClassVar[list[str]] = ["bgp"]
+    commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaTemplate(template="show bgp neighbors {peer} vrf {vrf}", revision=3)]
+
+    class Input(AntaTest.Input):
+        """Input model for the VerifyBGPPeerUpdateErrors test."""
+
+        bgp_peers: list[BgpPeer]
+        """List of BGP peers"""
+
+        class BgpPeer(BaseModel):
+            """Model for a BGP peer."""
+
+            peer_address: IPv4Address
+            """IPv4 address of a BGP peer."""
+            vrf: str = "default"
+            """Optional VRF for BGP peer. If not provided, it defaults to `default`."""
+            update_errors: list[BgpUpdateError] | None = None
+            """Optional list of update error counters to be verified. If not provided, test will verifies all the update error counters."""
+
+    def render(self, template: AntaTemplate) -> list[AntaCommand]:
+        """Render the template for each BGP peer in the input list."""
+        return [template.render(peer=str(bgp_peer.peer_address), vrf=bgp_peer.vrf) for bgp_peer in self.inputs.bgp_peers]
+
+    @AntaTest.anta_test
+    def test(self) -> None:
+        """Main test function for VerifyBGPPeerUpdateErrors."""
+        failures: dict[Any, Any] = {}
+
+        for command, input_entry in zip(self.instance_commands, self.inputs.bgp_peers):
+            peer = command.params.peer
+            vrf = command.params.vrf
+            update_error_counters = input_entry.update_errors
+
+            # Verify BGP peer.
+            if not (peer_list := get_value(command.json_output, f"vrfs.{vrf}.peerList")) or (peer_detail := get_item(peer_list, "peerAddress", peer)) is None:
+                failures[peer] = {vrf: "Not configured"}
+                continue
+
+            # Getting the BGP peer's error counters output.
+            error_counters_output = peer_detail.get("peerInUpdateErrors", {})
+
+            # In case update error counters not provided, It will check all the update error counters.
+            if not update_error_counters:
+                update_error_counters = error_counters_output
+
+            # verifying the error counters.
+            error_counters_not_ok = {
+                ("disabledAfiSafi" if error_counter == "disabledAfiSafi" else error_counter): value
+                for error_counter in update_error_counters
+                if (value := error_counters_output.get(error_counter, "Not Found")) != "None" and value != 0
+            }
+            if error_counters_not_ok:
+                failures[peer] = {vrf: error_counters_not_ok}
+
+        # Check if any failures
+        if not failures:
+            self.result.is_success()
+        else:
+            self.result.is_failure(f"The following BGP peers are not configured or have non-zero update error counters:\n{failures}")

examples/tests.yaml

@@ -567,6 +567,13 @@ anta.tests.routing:
             vrf: default
             hold_time: 180
             keep_alive_time: 60
+    - VerifyBGPPeerUpdateErrors:
+        bgp_peers:
+          - peer_address: 10.100.0.8
+            vrf: default
+            update_errors:
+              - inUpdErrWithdraw
+              - inUpdErrIgnore
   ospf:
     - VerifyOSPFNeighborState:
     - VerifyOSPFNeighborCount: