Feat(eos_validate_state): Added the validation for DPS interface reac…

…hability (#4154)

ansible_collections/arista/avd/molecule/eos_validate_state/intended/test_catalogs/dc1-wan1-catalog.yml

@@ -84,6 +84,24 @@ anta.tests.connectivity:
     result_overwrite:
       custom_field: 'Source: P2P Interface Ethernet2 (IP: 10.255.255.3) - Destination:
         dc1-leaf1b Ethernet6 (IP: 10.255.255.2)'
+- VerifyReachability:
+    hosts:
+    - destination: 10.255.1.1
+      repeat: 1
+      source: 10.255.1.1
+      vrf: default
+    result_overwrite:
+      custom_field: 'Source: Dps1 (IP: 10.255.1.1) - Destination: dc1-wan1 Dps1 (IP:
+        10.255.1.1)'
+- VerifyReachability:
+    hosts:
+    - destination: 10.255.1.2
+      repeat: 1
+      source: 10.255.1.1
+      vrf: default
+    result_overwrite:
+      custom_field: 'Source: Dps1 (IP: 10.255.1.1) - Destination: dc1-wan2 Dps1 (IP:
+        10.255.1.2)'
 anta.tests.field_notices:
 - VerifyFieldNotice44Resolution: null
 - VerifyFieldNotice72Resolution: null

ansible_collections/arista/avd/molecule/eos_validate_state/intended/test_catalogs/dc1-wan2-catalog.yml

@@ -84,6 +84,24 @@ anta.tests.connectivity:
     result_overwrite:
       custom_field: 'Source: P2P Interface Ethernet2 (IP: 10.255.255.7) - Destination:
         dc1-leaf1b Ethernet7 (IP: 10.255.255.6)'
+- VerifyReachability:
+    hosts:
+    - destination: 10.255.1.1
+      repeat: 1
+      source: 10.255.1.2
+      vrf: default
+    result_overwrite:
+      custom_field: 'Source: Dps1 (IP: 10.255.1.2) - Destination: dc1-wan1 Dps1 (IP:
+        10.255.1.1)'
+- VerifyReachability:
+    hosts:
+    - destination: 10.255.1.2
+      repeat: 1
+      source: 10.255.1.2
+      vrf: default
+    result_overwrite:
+      custom_field: 'Source: Dps1 (IP: 10.255.1.2) - Destination: dc1-wan2 Dps1 (IP:
+        10.255.1.2)'
 anta.tests.field_notices:
 - VerifyFieldNotice44Resolution: null
 - VerifyFieldNotice72Resolution: null

ansible_collections/arista/avd/molecule/eos_validate_state/reports/test_results/dc1-wan1-results.json

@@ -188,6 +188,24 @@
     "description": "Verifies that the provided LLDP neighbors are connected properly.",
     "custom_field": "Local: Ethernet2 - Remote: dc1-leaf1b Ethernet6"
   },
+  {
+    "name": "dc1-wan1",
+    "test": "VerifyReachability",
+    "categories": [
+      "connectivity"
+    ],
+    "description": "Test the network reachability to one or many destination IP(s).",
+    "custom_field": "Source: Dps1 (IP: 10.255.1.1) - Destination: dc1-wan1 Dps1 (IP: 10.255.1.1)"
+  },
+  {
+    "name": "dc1-wan1",
+    "test": "VerifyReachability",
+    "categories": [
+      "connectivity"
+    ],
+    "description": "Test the network reachability to one or many destination IP(s).",
+    "custom_field": "Source: Dps1 (IP: 10.255.1.1) - Destination: dc1-wan2 Dps1 (IP: 10.255.1.2)"
+  },
   {
     "name": "dc1-wan1",
     "test": "VerifyReachability",

ansible_collections/arista/avd/molecule/eos_validate_state/reports/test_results/dc1-wan2-results.json

@@ -188,6 +188,24 @@
     "description": "Verifies that the provided LLDP neighbors are connected properly.",
     "custom_field": "Local: Ethernet2 - Remote: dc1-leaf1b Ethernet7"
   },
+  {
+    "name": "dc1-wan2",
+    "test": "VerifyReachability",
+    "categories": [
+      "connectivity"
+    ],
+    "description": "Test the network reachability to one or many destination IP(s).",
+    "custom_field": "Source: Dps1 (IP: 10.255.1.2) - Destination: dc1-wan1 Dps1 (IP: 10.255.1.1)"
+  },
+  {
+    "name": "dc1-wan2",
+    "test": "VerifyReachability",
+    "categories": [
+      "connectivity"
+    ],
+    "description": "Test the network reachability to one or many destination IP(s).",
+    "custom_field": "Source: Dps1 (IP: 10.255.1.2) - Destination: dc1-wan2 Dps1 (IP: 10.255.1.2)"
+  },
   {
     "name": "dc1-wan2",
     "test": "VerifyReachability",

ansible_collections/arista/avd/plugins/plugin_utils/eos_validate_state_utils/avdtestbase.py

@@ -53,6 +53,11 @@ def vtep_mapping(self) -> list[tuple[str, str]]:
         """Return the vtep_mapping from the ConfigManager instance."""
         return self.config_manager.vtep_mapping
 
+    @property
+    def dps_mapping(self) -> list[tuple[str, str]]:
+        """Return the dps_mapping from the ConfigManager instance."""
+        return self.config_manager.dps_mapping
+
     def render(self) -> dict:
         """Return the test_definition of the class.
 

ansible_collections/arista/avd/plugins/plugin_utils/eos_validate_state_utils/config_manager.py

@@ -50,6 +50,7 @@ def __init__(self, device_name: str, hostvars: Mapping) -> None:
         self.structured_config = self.get_host_structured_config(host=device_name)
         self.loopback0_mapping = self.get_loopback0_mapping()
         self.vtep_mapping = self.get_vtep_mapping()
+        self.dps_mapping = self.get_dps_mapping()
 
     def get_host_structured_config(self, host: str) -> Mapping:
         """Get a specified host's structured configuration from the hostvars.
@@ -84,6 +85,35 @@ def get_vtep_mapping(self) -> list[tuple[str, str]]:
         """Get the vtep_mapping list."""
         return self._get_loopback_mappings["vtep_mapping"]
 
+    def get_dps_mapping(self) -> list[tuple[str, str]]:
+        """Get the dps_mapping list."""
+        return self._get_loopback_mappings["dps_mapping"]
+
+    def _get_ip_address(self, host: str, interfaces: list, vtep_interface: str) -> tuple[str, str] | None:
+        """Retrieve the IP address for a given VTEP interface on a host.
+
+        Parameters
+        ----------
+        host: str
+            The hostname of the device.
+        interfaces: list
+            List of interface dictionaries.
+        vtep_interface: str
+            The name of the VTEP interface.
+
+        Returns:
+        -------
+        tuple | None
+            A tuple containing the hostname and IP address if found, else None.
+        """
+        if (loopback_interface := get_item(interfaces, "name", vtep_interface)) is None:
+            LOGGER.warning("Host '%s' interface '%s' is missing.", host, vtep_interface)
+        elif (loopback_ip := loopback_interface.get("ip_address")) is None:
+            LOGGER.warning("Host '%s' variable 'ip_address' of interface '%s' is missing.", host, vtep_interface)
+        else:
+            return (host, str(ip_interface(loopback_ip).ip))
+        return None  # Ensure a value is always returned
+
     @cached_property
     def _get_loopback_mappings(self) -> dict:
         """Generate the loopback mappings for the eos_validate_state tests, which are used in AvdTestBase subclasses.
@@ -93,13 +123,16 @@ def _get_loopback_mappings(self) -> dict:
             dict: A dictionary containing:
             - "loopback0_mapping": A list of tuples where each tuple contains a hostname and its Loopback0 IP address.
             - "vtep_mapping": A list of tuples where each tuple contains a hostname and its VTEP IP address if `Vxlan1` is the source_interface.
-
+            - "dps_mapping": A list of tuples where each tuple contains a hostname and its VTEP IP address if `Dps` is in the source_interface.
         """
-        results = {"loopback0_mapping": [], "vtep_mapping": []}
+        results = {"loopback0_mapping": [], "vtep_mapping": [], "dps_mapping": []}
 
         for host in self.hostvars:
             host_struct_cfg = self.get_host_structured_config(host)
             loopback_interfaces = host_struct_cfg.get("loopback_interfaces", [])
+            dps_interfaces = host_struct_cfg.get("dps_interfaces", [])
+
+            # Handle Loopback0 interface
             if (loopback0 := get_item(loopback_interfaces, "name", "Loopback0")) is not None:
                 if (loopback_ip := loopback0.get("ip_address")) is None:
                     LOGGER.warning("Host '%s' variable 'ip_address' of interface 'Loopback0' is missing.", host)
@@ -111,13 +144,17 @@ def _get_loopback_mappings(self) -> dict:
             vtep_interface = default(
                 get(host_struct_cfg, "vxlan_interface.vxlan1.vxlan.source_interface"), get(host_struct_cfg, "vxlan_interface.Vxlan1.vxlan.source_interface")
             )
+            if vtep_interface is None:
+                continue
+
+            # Determine the correct mapping based on the interface name
+            if "Dps" in vtep_interface:
+                ip_address = self._get_ip_address(host, dps_interfaces, vtep_interface)
+                if ip_address:
+                    results["dps_mapping"].append(ip_address)
+            else:
+                ip_address = self._get_ip_address(host, loopback_interfaces, vtep_interface)
+                if ip_address:
+                    results["vtep_mapping"].append(ip_address)
 
-            # NOTE: For now we exclude WAN VTEPs from the vtep_mapping
-            if vtep_interface is not None and "Dps" not in vtep_interface:
-                if (loopback_interface := get_item(loopback_interfaces, "name", vtep_interface)) is None:
-                    LOGGER.warning("Host '%s' interface '%s' is missing.", host, vtep_interface)
-                elif (loopback_ip := loopback_interface.get("ip_address")) is None:
-                    LOGGER.warning("Host '%s' variable 'ip_address' of interface '%s' is missing.", host, vtep_interface)
-                else:
-                    results["vtep_mapping"].append((host, str(ip_interface(loopback_ip).ip)))
         return results

ansible_collections/arista/avd/plugins/plugin_utils/eos_validate_state_utils/mixins.py

@@ -111,6 +111,18 @@ def is_subinterface(self, interface: dict) -> bool:
         """
         return "." in interface.get("name", "")
 
+    def is_vtep(self) -> bool:
+        """Check if the host is a VTEP by verifying the presence of a VXLAN interface."""
+        return get(self.structured_config, "vxlan_interface") is not None
+
+    def is_wan_vtep(self) -> bool:
+        """Check if the host is a WAN VTEP by verifying the presence of a VXLAN interface and Dps in the source interface."""
+        return self.is_vtep() and "Dps" in get(
+            self.structured_config,
+            "vxlan_interface.vxlan1.vxlan.source_interface",
+            (get(self.structured_config, "vxlan_interface.Vxlan1.vxlan.source_interface", "")),
+        )
+
 
 class ValidationMixin:
     """Mixin class for the eos_validate_state tests.

ansible_collections/arista/avd/roles/eos_designs/docs/how-to/wan.md

@@ -989,6 +989,7 @@ wan_virtual_topologies:
 | AvdTestBGP | VerifyBGPSpecificPeers | Validate the state of BGP Address Family sessions, including `Path-Selection` for AutoVPN, `Link-State` and `IPv4/IPv6 SR-TE` for CV Pathfinder. |
 | AvdTestIPSecurity | VerifySpecificIPSecConn | Validate the establishment of IP security connections for each static peer under the `router path-selection` section of the configuration. |
 | AvdTestStun | VerifyStunClient | Validate the presence of a STUN client translation for a given source IPv4 address and port. The list of expected translations for each device is built by searching local interfaces in each path-group. |
+| AvdTestDpsReachability | VerifyReachability | Validate DPS reachability between devices. |
 
 !!! note
     More WAN-related tests are available directly in ANTA and can be added using custom catalogs.

ansible_collections/arista/avd/roles/eos_validate_state/README.md

@@ -89,6 +89,9 @@ Figure 1 below provides a visualization of the role's inputs, outputs, and tasks
 - AvdTestLoopback0Reachability
   - VerifyReachability: Validate loopback reachability between devices.
 
+- AvdTestDpsReachability
+  - VerifyReachability: Validate DPS reachability between devices.
+
 - AvdTestLLDPTopology
   - VerifyLLDPNeighbors: Validate LLDP topology.
 

ansible_collections/arista/avd/roles/eos_validate_state/python_modules/constants.py

@@ -8,6 +8,7 @@
 from .tests import (
     AvdTestAPIHttpsSSL,
     AvdTestBGP,
+    AvdTestDpsReachability,
     AvdTestHardware,
     AvdTestInbandReachability,
     AvdTestInterfacesState,
@@ -42,6 +43,7 @@
     AvdTestAPIHttpsSSL: {},
     AvdTestIPSecurity: {},
     AvdTestStun: {},
+    AvdTestDpsReachability: {},
 }
 """
 A dict of all AVD eos_validate_state test classes.

ansible_collections/arista/avd/roles/eos_validate_state/python_modules/tests/__init__.py

@@ -1,7 +1,7 @@
 # Copyright (c) 2023-2024 Arista Networks, Inc.
 # Use of this source code is governed by the Apache License 2.0
 # that can be found in the LICENSE file.
-from .avdtestconnectivity import AvdTestInbandReachability, AvdTestLLDPTopology, AvdTestLoopback0Reachability, AvdTestP2PIPReachability
+from .avdtestconnectivity import AvdTestDpsReachability, AvdTestInbandReachability, AvdTestLLDPTopology, AvdTestLoopback0Reachability, AvdTestP2PIPReachability
 from .avdtesthardware import AvdTestHardware
 from .avdtestinterfaces import AvdTestInterfacesState
 from .avdtestmlag import AvdTestMLAG
@@ -25,4 +25,5 @@
     "AvdTestAPIHttpsSSL",
     "AvdTestStun",
     "AvdTestIPSecurity",
+    "AvdTestDpsReachability",
 ]

ansible_collections/arista/avd/roles/eos_validate_state/python_modules/tests/avdtestconnectivity.py

@@ -126,16 +126,12 @@ def test_definition(self) -> dict | None:
         anta_tests = []
 
         # Skip the test if the host is not a VTEP (no VXLAN interface)
-        if get(self.structured_config, "vxlan_interface") is None:
+        if not self.is_vtep():
             LOGGER.info("Host is not a VTEP since it doesn't have a VXLAN interface. %s is skipped.", self.__class__.__name__)
             return None
 
         # TODO: For now, we exclude WAN VTEPs from testing
-        # TODO: Remove the support of Vxlan1 in AVD 6.0.0 version
-        if "Dps" in default(
-            get(self.structured_config, "vxlan_interface.vxlan1.vxlan.source_interface"),
-            get(self.structured_config, "vxlan_interface.Vxlan1.vxlan.source_interface"),
-        ):
+        if self.is_wan_vtep():
             LOGGER.info("Host is a VTEP with a DPS source interface for VXLAN. For now, WAN VTEPs are excluded. %s is skipped.", self.__class__.__name__)
             return None
 
@@ -161,6 +157,52 @@ def test_definition(self) -> dict | None:
         return {self.anta_module: anta_tests} if anta_tests else None
 
 
+class AvdTestDpsReachability(AvdTestBase):
+    """AvdTestDpsReachability class for DPS reachability tests."""
+
+    anta_module = "anta.tests.connectivity"
+
+    @cached_property
+    def test_definition(self) -> dict | None:
+        """
+        Generates the proper ANTA test definition for all DPS reachability tests.
+
+        Returns:
+            dict | None: ANTA test definition if there are tests to run, otherwise None.
+        """
+        anta_tests = []
+
+        # Skip the test if the host is not a WAN VTEP
+        if not self.is_wan_vtep():
+            LOGGER.info("Host is not a WAN VTEP. %s is skipped.", self.__class__.__name__)
+            return None
+
+        # TODO: Remove the support of Vxlan1 in AVD 6.0.0 version
+        dps_source_interface = default(
+            get(self.structured_config, "vxlan_interface.vxlan1.vxlan.source_interface"),
+            get(self.structured_config, "vxlan_interface.Vxlan1.vxlan.source_interface"),
+        )
+        dps_ip = self.get_interface_ip("dps_interfaces", dps_source_interface)
+        if not dps_ip:
+            return None
+        src_ip = str(ip_interface(dps_ip).ip)
+        for dst_node, dst_ip in self.dps_mapping:
+            if not self.is_peer_available(dst_node):
+                continue
+
+            custom_field = f"Source: {dps_source_interface} (IP: {src_ip}) - Destination: {dst_node} {dps_source_interface} (IP: {dst_ip})"
+            anta_tests.append(
+                {
+                    "VerifyReachability": {
+                        "hosts": [{"source": src_ip, "destination": dst_ip, "vrf": "default", "repeat": 1}],
+                        "result_overwrite": {"custom_field": custom_field},
+                    }
+                }
+            )
+
+        return {self.anta_module: anta_tests} if anta_tests else None
+
+
 class AvdTestLLDPTopology(AvdTestBase):
     """AvdTestLLDPTopology class for the LLDP topology tests."""