aristanetworks · gmuloc · May 5, 2023 · May 4, 2023 · May 5, 2023
diff --git a/anta/cli/utils.py b/anta/cli/utils.py
@@ -10,6 +10,9 @@
 from rich.logging import RichHandler
 
 logger = logging.getLogger(__name__)
+# For logs triggered before setup_logging is called
+FORMAT = "%(message)s"
+logging.basicConfig(format=FORMAT, datefmt="[%X]", handlers=[RichHandler()])
 
 
 def setup_logging(level: str = "info") -> None:
@@ -25,9 +28,9 @@ def setup_logging(level: str = "info") -> None:
         level (str, optional): level name to configure. Defaults to 'critical'.
     """
     loglevel = getattr(logging, level.upper())
+    logging.getLogger("anta").setLevel(loglevel)
+    logging.getLogger("aioeapi").setLevel(loglevel)
 
-    FORMAT = "%(message)s"
-    logging.basicConfig(level=loglevel, format=FORMAT, datefmt="[%X]", handlers=[RichHandler()])
     logging.getLogger("anta.inventory").setLevel(loglevel)
     logging.getLogger("anta.result_manager").setLevel(loglevel)
 

diff --git a/anta/inventory/models.py b/anta/inventory/models.py
@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import logging
+import sys
 import traceback
 from typing import Any, Dict, Iterator, List, Optional, Union
 
@@ -23,6 +24,42 @@
 
 RFC_1123_REGEX = r"^(([a-zA-Z0-9]|[a-zA-Z0-9][a-zA-Z0-9\-]*[a-zA-Z0-9])\.)*([A-Za-z0-9]|[A-Za-z0-9][A-Za-z0-9\-]*[A-Za-z0-9])$"
 
+if sys.version_info < (3, 10):
+    # @gmuloc - TODO - drop this when anta drops 3.8/3.9 support
+    # For Python < 3.10, it is not possible to install a version of aio-eapi newer than 0.3.0
+    # which sadly hardcodes version to 1 in its call to eAPI
+    # This little piece of nasty hack patches the aio-eapi function to support using a different
+    # version of the eAPI.
+    # Hic Sunt Draconis.
+    # Are we proud of this? No.
+    def patched_jsoncrpc_command(self: Device, commands: List[str], ofmt: str, **kwargs: Dict[Any, Any]) -> Dict[str, Any]:
+        """
+        Used to create the JSON-RPC command dictionary object
+        """
+        version = kwargs.get("version", "latest")
+
+        cmd = {
+            "jsonrpc": "2.0",
+            "method": "runCmds",
+            "params": {
+                "version": version,
+                "cmds": commands,
+                "format": ofmt or self.EAPI_DEFAULT_OFMT,
+            },
+            "id": str(kwargs.get("req_id") or id(self)),
+        }
+        if "autoComplete" in kwargs:
+            cmd["params"]["autoComplete"] = kwargs["autoComplete"]  # type: ignore
+
+        if "expandAliases" in kwargs:
+            cmd["params"]["expandAliases"] = kwargs["expandAliases"]  # type: ignore
+
+        return cmd
+
+    python_version = ".".join(map(str, sys.version_info[:3]))
+    logger.warning(f"Using Python {python_version} < 3.10 - patching aioeapi.Device.jsoncrpc_command to support 'latest' version")
+    Device.jsoncrpc_command = patched_jsoncrpc_command
+
 
 class AntaInventoryHost(BaseModel):
     """
@@ -189,6 +226,7 @@ async def collect(self, command: AntaTestCommand) -> Any:
             response = await self.session.cli(
                 commands=[enable_cmd, command.command],
                 ofmt=command.ofmt,
+                version=command.version,
             )
             # remove first dict related to enable command
             # only applicable to json output

diff --git a/anta/models.py b/anta/models.py
@@ -9,7 +9,7 @@
 from abc import ABC, abstractmethod
 from copy import deepcopy
 from functools import wraps
-from typing import TYPE_CHECKING, Any, Callable, ClassVar, Coroutine, Dict, Optional, TypeVar, Union
+from typing import TYPE_CHECKING, Any, Callable, ClassVar, Coroutine, Dict, Literal, Optional, TypeVar, Union
 
 from pydantic import BaseModel
 
@@ -27,13 +27,13 @@ class AntaTestCommand(BaseModel):
 
     Attributes:
         command(str): Test command
-        version(str): eAPI version - default is latest
+        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
     """
 
     command: str
-    version: str = "latest"
+    version: Union[int, Literal["latest"]] = "latest"
     ofmt: str = "json"
     output: Optional[Union[Dict[str, Any], str]]
     is_dynamic: bool = False
@@ -44,13 +44,13 @@ class AntaTestTemplate(BaseModel):
 
     Attributes:
         command(str): Test command
-        version(str): eAPI version - default is latest
+        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: str
-    version: str = "latest"
+    version: Union[int, Literal["latest"]] = "latest"
     ofmt: str = "json"
 
 

diff --git a/anta/tests/routing/generic.py b/anta/tests/routing/generic.py
@@ -2,108 +2,108 @@
 Generic routing test functions
 """
 import logging
+from typing import Any, Dict, Optional, cast
 
-from anta.inventory.models import InventoryDevice
-from anta.result_manager.models import TestResult
-from anta.tests import anta_test
+from anta.models import AntaTest, AntaTestCommand
 
 logger = logging.getLogger(__name__)
 
 
-@anta_test
-async def verify_routing_protocol_model(device: InventoryDevice, result: TestResult, model: str = "multi-agent") -> TestResult:
+class VerifyRoutingProtocolModel(AntaTest):
     """
     Verifies the configured routing protocol model is the one we expect.
     And if there is no mismatch between the configured and operating routing protocol model.
 
-    Args:
-        device (InventoryDevice): InventoryDevice instance containing all devices information.
         model(str): Expected routing protocol model (multi-agent or ribd). Default is multi-agent
-
-    Returns:
-        TestResult instance with
-        * result = "unset" if the test has not been executed
-        * result = "skipped" if the test `model` parameter is missing
-        * result = "success" if routing model is well configured
-        * result = "failure" otherwise.
-        * result = "error" if any exception is caught
     """
-    if not model:
-        result.is_skipped("verify_routing_protocol_model was not run as no model was given")
-        return result
 
-    response = await device.session.cli(command={"cmd": "show ip route summary", "revision": 3}, ofmt="json")
-    logger.debug(f"query result is: {response}")
-    configured_model = response["protoModelStatus"]["configuredProtoModel"]
-    operating_model = response["protoModelStatus"]["operatingProtoModel"]
-    if configured_model == operating_model == model:
-        result.is_success()
-    else:
-        result.is_failure(f"routing model is misconfigured: configured:{configured_model} - " f"operating:{operating_model} - expected:{model} ")
+    name = "VerifyRoutingProtocolModel"
+    description = (
+        "Verifies the configured routing protocol model is the expected one and if there is no mismatch between the configured and operating routing protocol model."
+    )
+    categories = ["routing", "generic"]
+    # "revision": 3
+    commands = [AntaTestCommand(command="show ip route summary")]
+
+    @AntaTest.anta_test
+    def test(self, model: Optional[str] = "multi-agent") -> None:
+        """Run VerifyRoutingProtocolModel validation"""
+
+        if not model:
+            self.result.is_skipped("VerifyRoutingProtocolModel was not run as no model was given")
+            return
+        command_output = cast(Dict[str, Dict[Any, Any]], self.instance_commands[0].output)
 
-    return result
+        configured_model = command_output["protoModelStatus"]["configuredProtoModel"]
+        operating_model = command_output["protoModelStatus"]["operatingProtoModel"]
+        if configured_model == operating_model == model:
+            self.result.is_success()
+        else:
+            self.result.is_failure(f"routing model is misconfigured: configured: {configured_model} - operating: {operating_model} - expected: {model}")
 
 
-@anta_test
-async def verify_routing_table_size(device: InventoryDevice, result: TestResult, minimum: int, maximum: int) -> TestResult:
+class VerifyRoutingTableSize(AntaTest):
     """
     Verifies the size of the IP routing table (default VRF).
     Should be between the two provided thresholds.
 
     Args:
-        device (InventoryDevice): InventoryDevice instance containing all devices information.
         minimum(int): Expected minimum routing table (default VRF) size.
         maximum(int): Expected maximum routing table (default VRF) size.
-
-    Returns:
-        TestResult instance with
-        * result = "unset" if the test has not been executed
-        * result = "skipped" if the test `minimum` or `maximum` parameters are missing
-        * result = "success" if routing-table size is correct
-        * result = "failure" otherwise.
-        * result = "error" if any exception is caught
     """
-    if not minimum or not maximum:
-        result.is_skipped("verify_routing_table_size was not run as no minimum or maximum were given")
-        return result
-    response = await device.session.cli(command={"cmd": "show ip route summary", "revision": 3}, ofmt="json")
-    logger.debug(f"query result is: {response}")
-    total_routes = int(response["vrfs"]["default"]["totalRoutes"])
-    if minimum <= total_routes <= maximum:
-        result.is_success()
-    else:
-        result.is_failure(f"routing-table has {total_routes} routes and not between min ({minimum}) and maximum ({maximum})")
-
-    return result
-
-
-@anta_test
-async def verify_bfd(device: InventoryDevice, result: TestResult) -> TestResult:
+
+    name = "VerifyRoutingTableSize"
+    description = "Verifies the size of the IP routing table (default VRF). Should be between the two provided thresholds."
+    categories = ["routing", "generic"]
+    # "revision": 3
+    commands = [AntaTestCommand(command="show ip route summary")]
+
+    @AntaTest.anta_test
+    def test(self, minimum: Optional[int] = None, maximum: Optional[int] = None) -> None:
+        """Run VerifyRoutingTableSize validation"""
+
+        if not minimum or not maximum:
+            self.result.is_skipped(f"VerifyRoutingTableSize was not run as either minimum {minimum} or maximum {maximum} was not provided")
+            return
+        if not isinstance(minimum, int) or not isinstance(maximum, int):
+            self.result.is_error(f"VerifyRoutingTableSize was not run as either minimum {minimum} or maximum {maximum} is not a valid value (integer)")
+            return
+        if maximum < minimum:
+            self.result.is_error(f"VerifyRoutingTableSize was not run as minimum {minimum} is greate than maximum {maximum}.")
+            return
+
+        command_output = cast(Dict[str, Dict[Any, Any]], self.instance_commands[0].output)
+        total_routes = int(command_output["vrfs"]["default"]["totalRoutes"])
+        if minimum <= total_routes <= maximum:
+            self.result.is_success()
+        else:
+            self.result.is_failure(f"routing-table has {total_routes} routes and not between min ({minimum}) and maximum ({maximum})")
+
+
+class VerifyBFD(AntaTest):
     """
     Verifies there is no BFD peer in down state (all VRF, IPv4 neighbors).
-
-    Args:
-        device (InventoryDevice): InventoryDevice instance containing all devices information.
-
-    Returns:
-        TestResult instance with
-        * result = "unset" if the test has not been executed
-        * result = "success" if routing-table size is OK
-        * result = "failure" otherwise.
-        * result = "error" if any exception is caught
     """
-    response = await device.session.cli(command="show bfd peers", ofmt="json")
-    logger.debug(f"query result is: {response}")
-    has_failed: bool = False
-    for vrf in response["vrfs"]:
-        for neighbor in response["vrfs"][vrf]["ipv4Neighbors"]:
-            for interface in response["vrfs"][vrf]["ipv4Neighbors"][neighbor]["peerStats"]:
-                if response["vrfs"][vrf]["ipv4Neighbors"][neighbor]["peerStats"][interface]["status"] != "up":
-                    intf_state = response["vrfs"][vrf]["ipv4Neighbors"][neighbor]["peerStats"][interface]["status"]
-                    intf_name = response["vrfs"][vrf]["ipv4Neighbors"][neighbor]["peerStats"][interface]
-                    has_failed = True
-                    result.is_failure(f"bfd state on interface {intf_name} is {intf_state} (expected up)")
-    if has_failed is False:
-        result.is_success()
-
-    return result
+
+    name = "VerifyBFD"
+    description = "Verifies there is no BFD peer in down state (all VRF, IPv4 neighbors)."
+    categories = ["routing", "generic"]
+    # revision 1 as later revision introduce additional nesting for type
+    commands = [AntaTestCommand(command="show bfd peers", version=1)]
+
+    @AntaTest.anta_test
+    def test(self) -> None:
+        """Run VerifyBFD validation"""
+
+        command_output = cast(Dict[str, Dict[Any, Any]], self.instance_commands[0].output)
+
+        self.result.is_success()
+
+        for _, vrf_data in command_output["vrfs"].items():
+            for _, neighbor_data in vrf_data["ipv4Neighbors"].items():
+                for peer, peer_data in neighbor_data["peerStats"].items():
+                    if (peer_status := peer_data["status"]) != "up":
+                        failure_message = f"bfd state for peer '{peer}' is {peer_status} (expected up)."
+                        if (peer_l3intf := peer_data.get("l3intf")) is not None and peer_l3intf != "":
+                            failure_message += f" Interface: {peer_l3intf}."
+                        self.result.is_failure(failure_message)
diff --git a/tests/units/anta_tests/mlag/data.py b/tests/units/anta_tests/mlag/data.py
@@ -172,7 +172,8 @@
         ],
         "side_effect": [],
         "expected_result": "failure",
-        "expected_messages": ["MLAG config-sanity returned Global inconsistancies: {'bridging': {'globalParameters': {'admin-state vlan 33': {'localValue': 'active'}, 'mac-learning vlan 33': {'localValue': 'True'}}}}"]  # noqa: E501
+        "expected_messages": ["MLAG config-sanity returned Global inconsistancies: {'bridging': {'globalParameters':"
+                              " {'admin-state vlan 33': {'localValue': 'active'}, 'mac-learning vlan 33': {'localValue': 'True'}}}}"]
     },
     {
         "name": "failure",
@@ -195,6 +196,7 @@
         ],
         "side_effect": [],
         "expected_result": "failure",
-        "expected_messages": ["MLAG config-sanity returned Interface inconsistancies: {'trunk-native-vlan mlag30': {'interface': {'Port-Channel30': {'localValue': '123', 'peerValue': '3700'}}}}"]  # noqa: E501
+        "expected_messages": ["MLAG config-sanity returned Interface inconsistancies: {'trunk-native-vlan mlag30': "
+                              "{'interface': {'Port-Channel30': {'localValue': '123', 'peerValue': '3700'}}}}"]
     },
 ]
diff --git a/tests/units/anta_tests/routing/generic/__init__.py b/tests/units/anta_tests/routing/generic/__init__.py