feat(anta.tests): Added testcases for AVT feature

anta/models.py

@@ -567,6 +567,7 @@ async def wrapper(
             if not self.collected:
                 await self.collect()
                 if self.result.result != "unset":
+                    AntaTest.update_progress()
                     return self.result
 
                 if cmds := self.failed_commands:
@@ -575,8 +576,9 @@ async def wrapper(
                         msg = f"Test {self.name} has been skipped because it is not supported on {self.device.hw_model}: {GITHUB_SUGGESTION}"
                         self.logger.warning(msg)
                         self.result.is_skipped("\n".join(unsupported_commands))
-                        return self.result
-                    self.result.is_error(message="\n".join([f"{c.command} has failed: {', '.join(c.errors)}" for c in cmds]))
+                    else:
+                        self.result.is_error(message="\n".join([f"{c.command} has failed: {', '.join(c.errors)}" for c in cmds]))
+                    AntaTest.update_progress()
                     return self.result
 
             try:

anta/tests/avt.py

@@ -0,0 +1,232 @@
+# 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.
+"""Module related to Adaptive virtual topology tests."""
+
+# Mypy does not understand AntaTest.Input typing
+# mypy: disable-error-code=attr-defined
+from __future__ import annotations
+
+from ipaddress import IPv4Address
+from typing import ClassVar
+
+from pydantic import BaseModel
+
+from anta.models import AntaCommand, AntaTemplate, AntaTest
+from anta.tools import get_value
+
+
+class VerifyAVTPathHealth(AntaTest):
+    """
+    Verifies the status of all Adaptive Virtual Topology (AVT) paths for all VRFs.
+
+    Expected Results
+    ----------------
+    * Success: The test will pass if all AVT paths for all VRFs are active and valid.
+    * Failure: The test will fail if the AVT path is not configured or if any AVT path under any VRF is either inactive or invalid.
+
+    Examples
+    --------
+    ```yaml
+    anta.tests.avt:
+      - VerifyAVTPathHealth:
+    ```
+    """
+
+    name = "VerifyAVTPathHealth"
+    description = "Verifies the status of all AVT paths for all VRFs."
+    categories: ClassVar[list[str]] = ["avt"]
+    commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show adaptive-virtual-topology path")]
+
+    @AntaTest.anta_test
+    def test(self) -> None:
+        """Main test function for VerifyAVTPathHealth."""
+        # Initialize the test result as success
+        self.result.is_success()
+
+        # Get the command output
+        command_output = self.instance_commands[0].json_output.get("vrfs", {})
+
+        # Check if AVT is configured
+        if not command_output:
+            self.result.is_failure("Adaptive virtual topology paths are not configured.")
+            return
+
+        # Iterate over each VRF
+        for vrf, vrf_data in command_output.items():
+            # Iterate over each AVT path
+            for profile, avt_path in vrf_data.get("avts", {}).items():
+                for path, flags in avt_path.get("avtPaths", {}).items():
+                    # Get the status of the AVT path
+                    valid = flags["flags"]["valid"]
+                    active = flags["flags"]["active"]
+
+                    # Check the status of the AVT path
+                    if not valid and not active:
+                        self.result.is_failure(f"AVT path {path} for profile {profile} in VRF {vrf} is invalid and not active.")
+                    elif not valid:
+                        self.result.is_failure(f"AVT path {path} for profile {profile} in VRF {vrf} is invalid.")
+                    elif not active:
+                        self.result.is_failure(f"AVT path {path} for profile {profile} in VRF {vrf} is not active.")
+
+
+class VerifyAVTSpecificPath(AntaTest):
+    """
+    Verifies the status and type of an Adaptive Virtual Topology (AVT) path for a specified VRF.
+
+    Expected Results
+    ----------------
+    * Success: The test will pass if the AVT path is active, valid, and matches the specified type (direct/multihop) for the given VRF.
+               If no path type is specified, both direct and multihop paths are considered.
+    * Failure: The test will fail if the AVT path is not configured or if the AVT path is not active, valid, or does not match the specified type for the given VRF.
+
+    Examples
+    --------
+    ```yaml
+    anta.tests.avt:
+      - VerifyAVTSpecificPath:
+          avt_paths:
+            - avt_name: CONTROL-PLANE-PROFILE
+              vrf: default
+              destination: 10.101.255.2
+              next_hop: 10.101.255.1
+              path_type: direct
+    ```
+    """
+
+    name = "VerifyAVTSpecificPath"
+    description = "Verifies the status and type of an AVT path for a specified VRF."
+    categories: ClassVar[list[str]] = ["avt"]
+    commands: ClassVar[list[AntaCommand | AntaTemplate]] = [
+        AntaTemplate(template="show adaptive-virtual-topology path vrf {vrf} avt {avt_name} destination {destination}")
+    ]
+
+    class Input(AntaTest.Input):
+        """Input model for the VerifyAVTSpecificPath test."""
+
+        avt_paths: list[AVTPaths]
+        """List of AVT paths to verify."""
+
+        class AVTPaths(BaseModel):
+            """Model for the details of AVT paths."""
+
+            vrf: str = "default"
+            """The VRF for the AVT path. Defaults to 'default' if not provided."""
+            avt_name: str
+            """Name of the adaptive virtual topology."""
+            destination: IPv4Address
+            """The IPv4 address of the AVT peer."""
+            next_hop: IPv4Address
+            """The IPv4 address of the next hop for the AVT peer."""
+            path_type: str | None = None
+            """The type of the AVT path. If not provided, both 'direct' and 'multihop' paths are considered."""
+
+    def render(self, template: AntaTemplate) -> list[AntaCommand]:
+        """Render the template for each input AVT path/peer."""
+        return [template.render(vrf=path.vrf, avt_name=path.avt_name, destination=path.destination) for path in self.inputs.avt_paths]
+
+    @AntaTest.anta_test
+    def test(self) -> None:
+        """Main test function for VerifyAVTSpecificPath."""
+        # Assume the test is successful until a failure is detected
+        self.result.is_success()
+
+        # Process each command in the instance
+        for command, input_avt in zip(self.instance_commands, self.inputs.avt_paths):
+            # Extract the command output and parameters
+            vrf = command.params.vrf
+            avt_name = command.params.avt_name
+            peer = str(command.params.destination)
+
+            command_output = command.json_output.get("vrfs", {})
+
+            # If no AVT is configured, mark the test as failed and skip to the next command
+            if not command_output:
+                self.result.is_failure(f"AVT configuration for peer '{peer}' under topology '{avt_name}' in VRF '{vrf}' is not found.")
+                continue
+
+            # Extract the AVT paths
+            avt_paths = get_value(command_output, f"{vrf}.avts.{avt_name}.avtPaths")
+            next_hop, input_path_type = str(input_avt.next_hop), input_avt.path_type
+
+            # Initialize flags for next-hop and path type
+            nexthop_path_found, path_type_found = False, False
+
+            # Check each AVT path
+            for path, path_data in avt_paths.items():
+                # If the path does not match the expected next hop, skip to the next path
+                if path_data.get("nexthopAddr") != next_hop:
+                    continue
+
+                nexthop_path_found = True
+                actual_path = get_value(path_data, "flags.directPath")
+                path_type = "direct" if actual_path else "multihop"
+
+                # If the path type does not match the expected path type, skip to the next path
+                if path_type != input_path_type and input_path_type:
+                    continue
+
+                path_type_found = True
+                valid = get_value(path_data, "flags.valid")
+                active = get_value(path_data, "flags.active")
+
+                # Construct the failure message prefix
+                failed_log = f"AVT path '{path}' for topology '{avt_name}' in VRF '{vrf}'"
+
+                # Check the path status and type against the expected values
+                if not all([valid, active]):
+                    failure_reasons = []
+                    if not active:
+                        failure_reasons.append("inactive")
+                    if not valid:
+                        failure_reasons.append("invalid")
+                    self.result.is_failure(f"{failed_log} is {', '.join(failure_reasons)}.")
+
+            # If no matching next hop or path type was found, mark the test as failed
+            if not nexthop_path_found or not path_type_found:
+                self.result.is_failure(
+                    f"No '{input_path_type}' path found with next-hop address '{next_hop}' for AVT peer '{peer}' under topology '{avt_name}' in VRF '{vrf}'."
+                )
+
+
+class VerifyAVTRole(AntaTest):
+    """
+    Verifies the Adaptive Virtual Topology (AVT) role of a device.
+
+    Expected Results
+    ----------------
+    * Success: The test will pass if the AVT role of the device matches the expected role.
+    * Failure: The test will fail if the AVT is not configured or if the AVT role does not match the expected role.
+
+    Examples
+    --------
+    ```yaml
+    anta.tests.avt:
+      - VerifyAVTRole:
+          role: edge
+    ```
+    """
+
+    name = "VerifyAVTRole"
+    description = "Verifies the AVT role of a device."
+    categories: ClassVar[list[str]] = ["avt"]
+    commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show adaptive-virtual-topology path")]
+
+    class Input(AntaTest.Input):
+        """Input model for the VerifyAVTRole test."""
+
+        role: str
+        """Expected AVT role of the device."""
+
+    @AntaTest.anta_test
+    def test(self) -> None:
+        """Main test function for VerifyAVTRole."""
+        # Initialize the test result as success
+        self.result.is_success()
+
+        # Get the command output
+        command_output = self.instance_commands[0].json_output
+
+        # Check if the AVT role matches the expected role
+        if self.inputs.role != command_output.get("role"):
+            self.result.is_failure(f"Expected AVT role as `{self.inputs.role}`, but found `{command_output.get('role')}` instead.")

docs/api/tests.avt.md

@@ -0,0 +1,20 @@
+---
+anta_title: ANTA catalog for Adaptive Virtual Topology (AVT) tests
+---
+<!--
+  ~ 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.
+  -->
+
+::: anta.tests.avt
+    options:
+      show_root_heading: false
+      show_root_toc_entry: false
+      show_bases: false
+      merge_init_into_class: false
+      anta_hide_test_module_description: true
+      show_labels: true
+      filters:
+        - "!test"
+        - "!render"

docs/api/tests.md

@@ -13,6 +13,7 @@ This section describes all the available tests provided by the ANTA package.
 Here are the tests that we currently provide:
 
 - [AAA](tests.aaa.md)
+- [Adaptive Virtual Topology](tests.avt.md)
 - [BFD](tests.bfd.md)
 - [Configuration](tests.configuration.md)
 - [Connectivity](tests.connectivity.md)

examples/tests.yaml

@@ -50,6 +50,18 @@ anta.tests.aaa:
         - commands
         - dot1x
 
+anta.tests.avt:
+  - VerifyAVTPathHealth:
+  - VerifyAVTSpecificPath:
+      avt_paths:
+        - avt_name: CONTROL-PLANE-PROFILE
+          vrf: default
+          destination: 10.101.255.2
+          next_hop: 10.101.255.1
+          path_type: direct
+  - VerifyAVTRole:
+      role: edge
+
 anta.tests.bfd:
   - VerifyBFDSpecificPeers:
       bfd_peers:

mkdocs.yml

@@ -178,6 +178,7 @@ nav:
   - Tests Documentation:
     - Overview: api/tests.md
     - AAA: api/tests.aaa.md
+    - Adaptive Virtual Topology: api/tests.avt.md
     - BFD: api/tests.bfd.md
     - Configuration: api/tests.configuration.md
     - Connectivity: api/tests.connectivity.md