feat(anta): Verify cvx status

anta/input_models/cvx.py

@@ -0,0 +1,19 @@
+# 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 containing input models for CVX tests."""
+
+from __future__ import annotations
+
+from typing import Literal
+
+from pydantic import BaseModel
+
+from anta.custom_types import Hostname
+
+
+class CVXPeers(BaseModel):
+    """Model for a CVX Cluster Peer."""
+
+    peer_name: Hostname
+    registration_state: Literal["Connecting", "Connected", "Registration error", "Registration complete", "Unexpected peer state"] = "Registration complete"

anta/tests/cvx.py

@@ -7,12 +7,21 @@
 # mypy: disable-error-code=attr-defined
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, ClassVar
+from typing import TYPE_CHECKING, Any, ClassVar, Literal, TypedDict
 
 from anta.models import AntaCommand, AntaTest
 
 if TYPE_CHECKING:
     from anta.models import AntaTemplate
+from anta.input_models.cvx import CVXPeers
+from anta.tools import get_value
+
+
+class Peer(TypedDict):
+    """Class for holding CVX Peer information."""
+
+    peer_name: str
+    registration_state: str
 
 
 class VerifyMcsClientMounts(AntaTest):
@@ -87,3 +96,137 @@ def test(self) -> None:
         cluster_status = command_output["clusterStatus"]
         if (cluster_state := cluster_status.get("enabled")) != self.inputs.enabled:
             self.result.is_failure(f"Management CVX status is not valid: {cluster_state}")
+
+
+class VerifyCVXClusterStatus(AntaTest):
+    """Verifies the CVX Server Cluster status.
+
+    Expected Results
+    ----------------
+    * Success: The test will pass if the CVX Server Cluster is enabled.
+    * Failure: The test will fail if any of the following conditions are met:
+        - If the CVX Status is disabled
+        - If the peers are not in "Registration ok" state
+
+    Examples
+    --------
+    ```yaml
+    anta.tests.cvx:
+      - VerifyCVXClusterStatus:
+          enabled: true
+          cluster_mode: true
+          role: Master
+          peer_status:
+            - peer_name : cvx-red-2
+              registration_state: Registration complete
+            - peer_name: cvx-red-3
+              registration_state: Registration complete
+    ```
+    """
+
+    name = "VerifyCVXClusterStatus"
+    description = "Verifies the CVX Cluster Status."
+    categories: ClassVar[list[str]] = ["cvx"]
+    commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show cvx", revision=1)]
+
+    class Input(AntaTest.Input):
+        """Input model for the VerifyCVXClusterStatus test."""
+
+        enabled: bool
+        """Whether management CVX must be enabled (True) or disabled (False)."""
+        cluster_mode: bool
+        role: Literal["Master", "Standby", "Disconnected"] = "Master"
+        peer_status: list[CVXPeers]
+        CVXPeers: ClassVar[type[CVXPeers]] = CVXPeers
+
+    @AntaTest.anta_test
+    def test(self) -> None:
+        """Run the main test for VerifyCVXClusterStatus."""
+        command_output = self.instance_commands[0].json_output
+        self.result.is_success()
+
+        # Validate cluster status
+        if not self._validate_cluster_status(command_output):
+            return
+
+        # Validate cluster mode and enabled status
+        if not self._validate_cluster_mode_and_enabled(command_output):
+            return
+
+        # Validate peer status
+        if not self._validate_peer_status(command_output.get("clusterStatus")):
+            return
+
+    def _validate_cluster_status(self, command_output: dict[str, Any]) -> bool:
+        """Check if the cluster status is available."""
+        cluster_status = command_output.get("clusterStatus")
+        if not cluster_status:
+            self.result.is_failure("CVX Server is not a cluster")
+            return False
+        return True
+
+    def _validate_cluster_mode_and_enabled(self, command_output: dict[str, Any]) -> bool:
+        """Check if the cluster mode and enabled status are correct."""
+        if not command_output.get("enabled"):
+            self.result.is_failure("CVX Server status is not enabled")
+            return False
+
+        cluster_status = command_output.get("clusterStatus")
+        if not (command_output.get("clusterMode") and cluster_status):
+            self.result.is_failure("CVX Server is not a cluster")
+            return False
+        return True
+
+    def _validate_peer_status(self, cluster_status: dict[str, Any] | None) -> bool:
+        """Check peer statuses in the cluster."""
+        if cluster_status is None:
+            self.result.is_failure("Cluster status is missing")
+            return False
+
+        peer_cluster = cluster_status.get("peerStatus")
+
+        # Check if peer_cluster is None or not sized
+        if not peer_cluster or not isinstance(peer_cluster, dict):
+            self.result.is_failure("Peer status data is invalid")
+            return False
+
+        # Check peer count
+        if len(peer_cluster) != len(self.inputs.peer_status):
+            self.result.is_failure("Unexpected number of peers")
+
+        # Check cluster role
+        cluster_role = cluster_status.get("role")
+        if cluster_role != self.inputs.role:
+            self.result.is_failure(f"CVX Role is not valid: {cluster_role}")
+            return False
+
+        # Check each peer
+        for peer in self.inputs.peer_status:
+            if not self._validate_individual_peer(peer, peer_cluster):
+                continue
+        return True
+
+    def _validate_individual_peer(self, peer: Peer, peer_cluster: dict[str, Any]) -> bool:
+        """Check an individual peer in the cluster.
+
+        Args:
+            peer (Peer): The peer object containing expected peer details.
+            peer_cluster (dict[str, Any]): The dictionary representing the peer statuses in the cluster.
+
+        Returns
+        -------
+            bool: True if the peer is valid, False otherwise.
+        """
+        # Retrieve the peer status from the peer cluster
+        eos_peer_status = get_value(peer_cluster, peer.peer_name, separator="..")
+        if eos_peer_status is None:
+            self.result.is_failure(f"{peer.peer_name} is not present")
+            return False
+
+        # Validate the registration state of the peer
+        peer_reg_state = eos_peer_status.get("registrationState")
+        if peer_reg_state != peer.registration_state:
+            self.result.is_failure(f"{peer.peer_name} registration state is not complete: {peer_reg_state}")
+            return False
+
+        return True

examples/tests.yaml

@@ -103,6 +103,15 @@ anta.tests.cvx:
   - VerifyMcsClientMounts:
   - VerifyManagementCVX:
       enabled: true
+  - VerifyCVXClusterStatus:
+          enabled: true
+          cluster_mode: true
+          role: Master
+          peer_status:
+            - peer_name : cvx-red-2
+              registration_state: Registration complete
+            - peer_name: cvx-red-3
+              registration_state: Registration complete
 
 anta.tests.connectivity:
   - VerifyReachability:

tests/units/anta_tests/test_cvx.py

@@ -7,7 +7,7 @@
 
 from typing import Any
 
-from anta.tests.cvx import VerifyManagementCVX, VerifyMcsClientMounts
+from anta.tests.cvx import VerifyCVXClusterStatus, VerifyManagementCVX, VerifyMcsClientMounts
 from tests.units.anta_tests import test
 
 DATA: list[dict[str, Any]] = [
@@ -146,4 +146,174 @@
         "inputs": {"enabled": False},
         "expected": {"result": "failure", "messages": ["Management CVX status is not valid: None"]},
     },
+    {
+        "name": "success-all",
+        "test": VerifyCVXClusterStatus,
+        "eos_data": [
+            {
+                "enabled": True,
+                "clusterMode": True,
+                "clusterStatus": {
+                    "role": "Master",
+                    "peerStatus": {
+                        "cvx-red-2": {"peerName": "cvx-red-2", "registrationState": "Registration complete"},
+                        "cvx-red-3": {"peerName": "cvx-red-3", "registrationState": "Registration complete"},
+                    },
+                },
+            }
+        ],
+        "inputs": {
+            "enabled": True,
+            "cluster_mode": True,
+            "role": "Master",
+            "peer_status": [
+                {"peer_name": "cvx-red-2", "registrationState": "Registration complete"},
+                {"peer_name": "cvx-red-3", "registrationState": "Registration complete"},
+            ],
+        },
+        "expected": {"result": "success"},
+    },
+    {
+        "name": "failure-invalid-role",
+        "test": VerifyCVXClusterStatus,
+        "eos_data": [
+            {
+                "enabled": True,
+                "clusterMode": True,
+                "clusterStatus": {
+                    "role": "Standby",
+                    "peerStatus": {
+                        "cvx-red-2": {"peerName": "cvx-red-2", "registrationState": "Registration complete"},
+                        "cvx-red-3": {"peerName": "cvx-red-3", "registrationState": "Registration complete"},
+                    },
+                },
+            }
+        ],
+        "inputs": {
+            "enabled": True,
+            "cluster_mode": True,
+            "role": "Master",
+            "peer_status": [
+                {"peer_name": "cvx-red-2", "registrationState": "Registration complete"},
+                {"peer_name": "cvx-red-3", "registrationState": "Registration complete"},
+            ],
+        },
+        "expected": {"result": "failure", "messages": ["CVX Role is not valid: Standby"]},
+    },
+    {
+        "name": "failure-cvx-enabled",
+        "test": VerifyCVXClusterStatus,
+        "eos_data": [
+            {
+                "enabled": False,
+                "clusterMode": True,
+                "clusterStatus": {
+                    "role": "Master",
+                    "peerStatus": {},
+                },
+            }
+        ],
+        "inputs": {
+            "enabled": True,
+            "cluster_mode": True,
+            "role": "Master",
+            "peer_status": [],
+        },
+        "expected": {"result": "failure", "messages": ["CVX Server status is not enabled"]},
+    },
+    {
+        "name": "failure-cluster-enabled",
+        "test": VerifyCVXClusterStatus,
+        "eos_data": [
+            {
+                "enabled": True,
+                "clusterMode": False,
+                "clusterStatus": {},
+            }
+        ],
+        "inputs": {
+            "enabled": True,
+            "cluster_mode": True,
+            "role": "Master",
+            "peer_status": [],
+        },
+        "expected": {"result": "failure", "messages": ["CVX Server is not a cluster"]},
+    },
+    {
+        "name": "failure-missing-peers",
+        "test": VerifyCVXClusterStatus,
+        "eos_data": [
+            {
+                "enabled": True,
+                "clusterMode": True,
+                "clusterStatus": {
+                    "role": "Master",
+                    "peerStatus": {
+                        "cvx-red-2": {"peerName": "cvx-red-2", "registrationState": "Registration complete"},
+                    },
+                },
+            }
+        ],
+        "inputs": {
+            "enabled": True,
+            "cluster_mode": True,
+            "role": "Master",
+            "peer_status": [
+                {"peer_name": "cvx-red-2", "registrationState": "Registration complete"},
+                {"peer_name": "cvx-red-3", "registrationState": "Registration complete"},
+            ],
+        },
+        "expected": {"result": "failure", "messages": ["Unexpected number of peers", "cvx-red-3 is not present"]},
+    },
+    {
+        "name": "failure-invalid-peers",
+        "test": VerifyCVXClusterStatus,
+        "eos_data": [
+            {
+                "enabled": True,
+                "clusterMode": True,
+                "clusterStatus": {
+                    "role": "Master",
+                    "peerStatus": {},
+                },
+            }
+        ],
+        "inputs": {
+            "enabled": True,
+            "cluster_mode": True,
+            "role": "Master",
+            "peer_status": [
+                {"peer_name": "cvx-red-2", "registrationState": "Registration complete"},
+                {"peer_name": "cvx-red-3", "registrationState": "Registration complete"},
+            ],
+        },
+        "expected": {"result": "failure", "messages": ["Peer status data is invalid"]},
+    },
+    {
+        "name": "failure-registration-error",
+        "test": VerifyCVXClusterStatus,
+        "eos_data": [
+            {
+                "enabled": True,
+                "clusterMode": True,
+                "clusterStatus": {
+                    "role": "Master",
+                    "peerStatus": {
+                        "cvx-red-2": {"peerName": "cvx-red-2", "registrationState": "Registration error"},
+                        "cvx-red-3": {"peerName": "cvx-red-3", "registrationState": "Registration complete"},
+                    },
+                },
+            }
+        ],
+        "inputs": {
+            "enabled": True,
+            "cluster_mode": True,
+            "role": "Master",
+            "peer_status": [
+                {"peer_name": "cvx-red-2", "registrationState": "Registration complete"},
+                {"peer_name": "cvx-red-3", "registrationState": "Registration complete"},
+            ],
+        },
+        "expected": {"result": "failure", "messages": ["cvx-red-2 registration state is not complete: Registration error"]},
+    },
 ]