aristanetworks · titom73 · May 26, 2023 · May 9, 2023 · May 10, 2023
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
@@ -58,4 +58,4 @@ repos:
           - types-paramiko
           - types-requests
         # Do not run mypy on tests but only on code and scripts
-        files: ^(anta|scripts)/
+        files: ^(anta|scripts|tests)/
diff --git a/anta/models.py b/anta/models.py
@@ -9,9 +9,9 @@
 from abc import ABC, abstractmethod
 from copy import deepcopy
 from functools import wraps
-from typing import TYPE_CHECKING, Any, Callable, ClassVar, Coroutine, Dict, Literal, Optional, TypeVar, Union
+from typing import TYPE_CHECKING, Any, Callable, ClassVar, Coroutine, Dict, Literal, Optional, Type, TypeVar, Union
 
-from pydantic import BaseModel
+from pydantic import BaseModel, validator
 
 from anta.result_manager.models import TestResult
 from anta.tools.misc import exc_to_str
@@ -22,7 +22,7 @@
 F = TypeVar("F", bound=Callable[..., Any])
 
 
-class AntaTestCommand(BaseModel):
+class AntaTestTemplate(BaseModel):
     """Class to define a test command with its API version
 
     Attributes:
@@ -32,26 +32,39 @@ class AntaTestCommand(BaseModel):
         output: collected output either dict for json or str for text
     """
 
-    command: str
+    template: str
     version: Union[int, Literal["latest"]] = "latest"
     ofmt: str = "json"
-    output: Optional[Union[Dict[str, Any], str]]
-    is_dynamic: bool = False
 
 
-class AntaTestTemplate(BaseModel):
+class AntaTestCommand(BaseModel):
     """Class to define a test command with its API version
 
     Attributes:
         command(str): Test command
         version: eAPI version - valid values are integers or the string "latest" - default is "latest"
         ofmt(str):  eAPI output - json or text - default is json
         output: collected output either dict for json or str for text
+        template Optional(AntaTestTemplate): Template used to generate the command
+        template_params Optional(dict): params used in the template to generate the command
     """
 
-    template: str
+    command: str
     version: Union[int, Literal["latest"]] = "latest"
     ofmt: str = "json"
+    output: Optional[Union[Dict[str, Any], str]]
+    template: Optional[AntaTestTemplate] = None
+    template_params: Optional[Dict[str, str]]
+
+    @validator("template_params")
+    def prevent_none_when_template_is_set(cls: Type[AntaTestTemplate], value: Optional[Dict[str, str]]) -> Optional[Dict[str, str]]:
+        """
+        Raises if template is set but no params are given
+        """
+        if hasattr(cls, "template") and cls.template is not None:
+            assert value is not None
+
+        return value
 
 
 class AntaTestFilter(ABC):
@@ -127,7 +140,8 @@ def __init__(
                     command=tpl.template.format(**param),
                     ofmt=tpl.ofmt,
                     version=tpl.version,
-                    is_dynamic=True,
+                    template=tpl,
+                    template_params=param,
                 )
                 for param in template_params
             )

diff --git a/anta/tests/hardware.py b/anta/tests/hardware.py
@@ -40,11 +40,11 @@ def test(self, manufacturers: Optional[List[str]] = None) -> None:
 
 class VerifyTemperature(AntaTest):
     """
-    Verifies device temparture is currently OK.
+    Verifies device temparture is currently OK (temperatureOK).
     """
 
     name = "VerifyTemperature"
-    description = "Verifies device temparture is currently OK"
+    description = "Verifies device temparture is currently OK (temperatureOK)"
     categories = ["hardware"]
     commands = [AntaTestCommand(command="show system environment temperature", ofmt="json")]
 
@@ -72,7 +72,7 @@ class VerifyTransceiversTemperature(AntaTest):
 
     @skip_on_platforms(["cEOSLab", "vEOS-lab"])
     @AntaTest.anta_test
-    def test(self) -> None:  # type: ignore[override]
+    def test(self) -> None:
         """Run VerifyTransceiversTemperature validation"""
         command_output = cast(Dict[str, Dict[Any, Any]], self.instance_commands[0].output)
         sensors = command_output["tempSensors"] if "tempSensors" in command_output.keys() else ""
@@ -91,32 +91,71 @@ def test(self) -> None:  # type: ignore[override]
             self.result.messages.append(str(wrong_sensors))
 
 
+class VerifyEnvironmentSystemCooling(AntaTest):
+    """
+    Verifies the System Cooling is ok.
+    """
+
+    name = "VerifyEnvironmentSystemCooling"
+    description = "Verifies the fans status is OK for fans"
+    categories = ["hardware"]
+    commands = [AntaTestCommand(command="show system environment cooling", ofmt="json")]
+
+    @skip_on_platforms(["cEOSLab", "vEOS-lab"])
+    @AntaTest.anta_test
+    def test(self) -> None:
+        """Run VerifyEnvironmentCooling validation"""
+
+        command_output = cast(Dict[str, Dict[Any, Any]], self.instance_commands[0].output)
+        sys_status = command_output["systemStatus"] if "systemStatus" in command_output.keys() else ""
+
+        self.result.is_success()
+        if sys_status != "coolingOk":
+            self.result.is_failure(f"Device System cooling is not OK: {sys_status}")
+
+
 class VerifyEnvironmentCooling(AntaTest):
     """
-    Verifies the fans status is OK.
+    Verifies the fans status is in the accepted states list.
+
+    Default accepted states list is ['ok']
     """
 
     name = "VerifyEnvironmentCooling"
-    description = "Verifies the fans status is OK"
+    description = "Verifies the fans status is OK for fans"
     categories = ["hardware"]
     commands = [AntaTestCommand(command="show system environment cooling", ofmt="json")]
 
     @skip_on_platforms(["cEOSLab", "vEOS-lab"])
     @AntaTest.anta_test
-    def test(self) -> None:  # type: ignore[override]
+    def test(self, accepted_states: Optional[List[str]] = None) -> None:
         """Run VerifyEnvironmentCooling validation"""
+        if accepted_states is None:
+            accepted_states = ["ok"]
 
         command_output = cast(Dict[str, Dict[Any, Any]], self.instance_commands[0].output)
-        sys_status = command_output["systemStatus"] if "systemStatus" in command_output.keys() else ""
-        if sys_status == "coolingOk":
-            self.result.is_success()
-        else:
-            self.result.is_failure("Device cooling is not OK")
+        self.result.is_success()
+        # First go through power supplies fans
+        for power_supply in command_output.get("powerSupplySlots", []):
+            for fan in power_supply.get("fans", []):
+                if (state := fan["status"]) not in accepted_states:
+                    if self.result.result == "success":
+                        self.result.is_failure(f"Some fans state are not in the accepted list: {accepted_states}.")
+                    self.result.is_failure(f"Fan {fan['label']} on PowerSupply {power_supply['label']} has state '{state}'.")
+        # Then go through Fan Trays
+        for fan_tray in command_output.get("fanTraySlots", []):
+            for fan in fan_tray.get("fans", []):
+                if (state := fan["status"]) not in accepted_states:
+                    if self.result.result == "success":
+                        self.result.is_failure(f"Some fans state are not in the accepted list: {accepted_states}.")
+                    self.result.is_failure(f"Fan {fan['label']} on Fan Tray {fan_tray['label']} has state '{state}'.")
 
 
 class VerifyEnvironmentPower(AntaTest):
     """
-    Verifies the power supplied status is OK.
+    Verifies the power supplied status is in the accepted states list
+
+    The default accepted states list is ['ok']
     """
 
     name = "VerifyEnvironmentPower"
@@ -126,15 +165,19 @@ class VerifyEnvironmentPower(AntaTest):
 
     @skip_on_platforms(["cEOSLab", "vEOS-lab"])
     @AntaTest.anta_test
-    def test(self) -> None:  # type: ignore[override]
+    def test(self, accepted_states: Optional[List[str]] = None) -> None:
         """Run VerifyEnvironmentPower validation"""
+        if accepted_states is None:
+            accepted_states = ["ok"]
         command_output = cast(Dict[str, Dict[Any, Any]], self.instance_commands[0].output)
         power_supplies = command_output["powerSupplies"] if "powerSupplies" in command_output.keys() else "{}"
-        wrong_power_supplies = {powersupply: {"state": value["state"]} for powersupply, value in dict(power_supplies).items() if value["state"] != "ok"}
+        wrong_power_supplies = {
+            powersupply: {"state": value["state"]} for powersupply, value in dict(power_supplies).items() if value["state"] not in accepted_states
+        }
         if not wrong_power_supplies:
             self.result.is_success()
         else:
-            self.result.is_failure("The following power supplies are not OK")
+            self.result.is_failure(f"The following power supplies states are not in the accepted_states list {accepted_states}")
             self.result.messages.append(str(wrong_power_supplies))
 
 
@@ -149,7 +192,7 @@ class VerifyAdverseDrops(AntaTest):
     commands = [AntaTestCommand(command="show hardware counter drop", ofmt="json")]
 
     @AntaTest.anta_test
-    def test(self) -> None:  # type: ignore[override]
+    def test(self) -> None:
         """Run VerifyAdverseDrops validation"""
         command_output = cast(Dict[str, Dict[Any, Any]], self.instance_commands[0].output)
         total_adverse_drop = command_output["totalAdverseDrops"] if "totalAdverseDrops" in command_output.keys() else ""

diff --git a/anta/tests/routing/bgp.py b/anta/tests/routing/bgp.py
@@ -95,16 +95,16 @@ def test(self, number: Optional[int] = None) -> None:
         if not number:
             self.result.is_skipped("VerifyBGPIPv4UnicastCount could not run because number was not supplied")
             return
-        vrf = self.template_params[0]["vrf"]
-        command_output = cast(Dict[str, Dict[Any, Any]], self.instance_commands[0].output)
 
-        peers = command_output["vrfs"][vrf]["peers"]
-        state_issue = _check_bgp_vrfs(command_output["vrfs"])
+        self.result.is_success()
+
+        for index, command in enumerate(self.instance_commands):
+            vrf = cast(Dict[str, str], command.template_params).get("vrf")
+            command_output = cast(Dict[str, Dict[Any, Any]], self.instance_commands[index].output)
+
+            peers = command_output["vrfs"][vrf]["peers"]
+            state_issue = _check_bgp_vrfs(command_output["vrfs"])
 
-        if not state_issue and len(peers) == number:
-            self.result.is_success()
-        else:
-            self.result.is_failure()
             if len(peers) != number:
                 self.result.is_failure(f"Expecting {number} BGP peer in vrf {vrf} and got {len(peers)}")
             if state_issue: