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,14 @@
 # mypy: disable-error-code=attr-defined
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, ClassVar
+from typing import TYPE_CHECKING, Any, ClassVar, Literal
 
 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 VerifyMcsClientMounts(AntaTest):
@@ -87,3 +89,77 @@ 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 all of the following conditions is met:
+        - CVX Enabled state is true
+        - Cluster Mode is true
+        - Role is either Master or Standby.
+        - peer_status matches defined state
+    * Failure: The test will fail if any of the success conditions is not met.
+
+    Examples
+    --------
+    ```yaml
+    anta.tests.cvx:
+      - VerifyCVXClusterStatus:
+          role: Master
+          peer_status:
+            - peer_name : cvx-red-2
+              registration_state: Registration complete
+            - peer_name: cvx-red-3
+              registration_state: Registration error
+    ```
+    """
+
+    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."""
+
+        role: Literal["Master", "Standby", "Disconnected"] = "Master"
+        peer_status: list[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 Server enabled status
+        if not command_output.get("enabled"):
+            self.result.is_failure("CVX Server status is not enabled")
+
+        # Validate cluster status and mode
+        if not (cluster_status := command_output.get("clusterStatus")) or not command_output.get("clusterMode"):
+            self.result.is_failure("CVX Server is not a cluster")
+            return
+
+        # Check cluster role
+        if (cluster_role := cluster_status.get("role")) != self.inputs.role:
+            self.result.is_failure(f"CVX Role is not valid: {cluster_role}")
+            return
+
+        # Validate peer status
+        peer_cluster = cluster_status.get("peerStatus", {})
+
+        # Check peer count
+        if (num_of_peers := len(peer_cluster)) != (expected_num_of_peers := len(self.inputs.peer_status)):
+            self.result.is_failure(f"Unexpected number of peers {num_of_peers} vs {expected_num_of_peers}")
+
+        # Check each peer
+        for peer in self.inputs.peer_status:
+            # Retrieve the peer status from the peer cluster
+            if (eos_peer_status := get_value(peer_cluster, peer.peer_name, separator="..")) is None:
+                self.result.is_failure(f"{peer.peer_name} is not present")
+                continue
+
+            # Validate the registration state of the peer
+            if (peer_reg_state := eos_peer_status.get("registrationState")) != peer.registration_state:
+                self.result.is_failure(f"{peer.peer_name} registration state is not complete: {peer_reg_state}")

examples/tests.yaml

@@ -135,6 +135,14 @@ anta.tests.connectivity:
           df_bit: True
           size: 100
 anta.tests.cvx:
+  - VerifyCVXClusterStatus:
+      # Verifies the CVX Server Cluster status.
+      role: Master
+      peer_status:
+        - peer_name : cvx-red-2
+          registration_state: Registration complete
+        - peer_name: cvx-red-3
+          registration_state: Registration error
   - VerifyManagementCVX:
       # Verifies the management CVX global status.
       enabled: true

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,160 @@
         "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": {
+            "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": {
+            "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": {
+            "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": {
+            "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": {
+            "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 1 vs 2", "cvx-red-3 is not present"]},
+    },
+    {
+        "name": "failure-invalid-peers",
+        "test": VerifyCVXClusterStatus,
+        "eos_data": [
+            {
+                "enabled": True,
+                "clusterMode": True,
+                "clusterStatus": {
+                    "role": "Master",
+                    "peerStatus": {},
+                },
+            }
+        ],
+        "inputs": {
+            "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 0 vs 2", "cvx-red-2 is not present", "cvx-red-3 is not present"]},
+    },
+    {
+        "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": {
+            "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"]},
+    },
 ]