WIP

anta/cli/exec/utils.py

@@ -12,15 +12,15 @@
 import logging
 import re
 from pathlib import Path
-from typing import TYPE_CHECKING, Literal
+from typing import TYPE_CHECKING, Any, Literal
 
 from aioeapi import EapiCommandError
 from click.exceptions import UsageError
 from httpx import ConnectError, HTTPError
 
 from anta.device import AntaDevice, AsyncEOSDevice
 from anta.models import AntaCommand
-from anta.platform_utils import SUPPORT_HARDWARE_COUNTERS_SERIES, SUPPORT_VNI_COUNTERS_SERIES, SUPPORT_VXLAN_COUNTERS_SERIES, get_model_series
+from anta.platform_utils import SUPPORT_HARDWARE_COUNTERS_SERIES, find_series_by_model
 from anta.tools.get_item import get_item
 
 if TYPE_CHECKING:
@@ -32,8 +32,18 @@
 
 
 async def clear_counters_utils(anta_inventory: AntaInventory, tags: list[str] | None = None) -> None:
- """Clear counters."""
- clear_counters_commands = [
+ """Clear counters on the devices in the inventory.
+
+ If the device is part of a series that supports hardware counters, the hardware counters are also cleared.
+
+ Arguments:
+ ----------
+ anta_inventory (AntaInventory): The ANTA inventory object containing the devices to connect to.
+ tags (list[str] | None): A list of tags to filter the devices to connect to.
+
+ """
+ # Commands to clear counters. Can be used later to add more commands if needed.
+ clear_counters_commands: list[dict[str, Any]] = [
  {
  "command": "clear counters",
  "is_cleared": False,
@@ -44,22 +54,17 @@ async def clear_counters_utils(anta_inventory: AntaInventory, tags: list[str] |
  "is_cleared": False,
  "restricted_series": SUPPORT_HARDWARE_COUNTERS_SERIES,
  },
- {
- "command": "clear vxlan counters vtep",
- "is_cleared": False,
- "restricted_series": SUPPORT_VXLAN_COUNTERS_SERIES,
- },
- {
- "command": "clear vxlan counters vni",
- "is_cleared": False,
- "restricted_series": SUPPORT_VNI_COUNTERS_SERIES,
- },
  ]
 
  async def clear(device: AntaDevice) -> None:
- model_series = get_model_series(device.hw_model)
+ if not isinstance(device.hw_model, str):
+ logger.error("Could not clear counters on device %s because its hardware model is not set or invalid.", device.name)
+ return
+
+ model_series = find_series_by_model(device.hw_model)
  commands = [
- AntaCommand(command=command["command"]) for command in clear_counters_commands
+ AntaCommand(command=command["command"])
+ for command in clear_counters_commands
  if not command["restricted_series"] or model_series in command["restricted_series"]
  ]
 

anta/decorators.py

@@ -9,7 +9,7 @@
 from typing import TYPE_CHECKING, Any, Callable, TypeVar, cast
 
 from anta.models import AntaTest, logger
-from anta.platform_utils import get_model_series
+from anta.platform_utils import find_series_by_model
 
 if TYPE_CHECKING:
  from anta.result_manager.models import TestResult
@@ -58,17 +58,59 @@ async def wrapper(*args: Any, **kwargs: Any) -> Any:
 
  return decorator
 
+
 def run_on_platform_series(series: list[str]) -> Callable[[F], F]:
+ """Return a decorator to run a test based on the device's hardware model.
+
+ This decorator factory generates a decorator that will check the hardware model of the device
+ the test is run on. If the model is part of the provided platform series, the test will be run.
+
+ Args:
+ ----
+ series (list[str]): List of platform series on which the test should be run.
+
+ Returns
+ -------
+ Callable[[F], F]: A decorator that can be used to wrap test functions.
+
+ Examples
+ --------
+ The following test will only run if the device's hardware model is part of the 7800R3, 7500R3, 7500R, or 7280R3 series, e.g. DCS-7280SR3-48YC8.
+
+ ```python
+ @run_on_platform_series([["7800R3", "7500R3", "7500R", "7280R3"])
+ @AntaTest.anta_test
+ def test(self) -> None:
+ pass
+ """
+
  def decorator(function: F) -> F:
+ """Actual decorator that either runs the test or skips it based on the device's hardware model.
+
+ Args:
+ ----
+ function (F): The test function to be decorated.
+
+ Returns
+ -------
+ F: The decorated function.
+
+ """
+
  @wraps(function)
  async def wrapper(*args: Any, **kwargs: Any) -> Any:
+ """Check the device's hardware model and conditionally run or skip the test.
+
+ This wrapper inspects the hardware model of the device the test is run on.
+ If the model is NOT in the list of specified platform series, the test is skipped.
+ """
  anta_test = args[0]
 
  if anta_test.result.result != "unset":
  AntaTest.update_progress()
  return anta_test.result
 
- if get_model_series(anta_test.device.hw_model) not in series:
+ if find_series_by_model(anta_test.device.hw_model) not in series:
  anta_test.result.is_skipped(f"{anta_test.__class__.__name__} test is not supported on {anta_test.device.hw_model}.")
  AntaTest.update_progress()
  return anta_test.result
@@ -80,7 +122,6 @@ async def wrapper(*args: Any, **kwargs: Any) -> Any:
  return decorator
 
 
-
 def skip_on_platforms(platforms: list[str]) -> Callable[[F], F]:
  """Return a decorator to skip a test based on the device's hardware model.
 
@@ -115,7 +156,7 @@ async def wrapper(*args: Any, **kwargs: Any) -> TestResult:
  """Check the device's hardware model and conditionally run or skip the test.
 
  This wrapper inspects the hardware model of the device the test is run on.
- If the model is in the list of specified platforms, the test is either skipped.
+ If the model is in the list of specified platforms, the test is skipped.
  """
  anta_test = args[0]
 

anta/device.py

@@ -342,7 +342,7 @@ async def _collect(self, command: AntaCommand) -> None:
  )
  except aioeapi.EapiCommandError as e:
  command.errors = e.errors
- if self.supports(command):
+ if self.supports(command, log=True):
  logger.error("Command '%s' failed on %s", command.command, self.name)
  except (HTTPError, ConnectError) as e:
  command.errors = [str(e)]

anta/platform_utils.py

@@ -12,21 +12,74 @@
 
 logger = logging.getLogger(__name__)
 
-SUPPORT_HARDWARE_COUNTERS_SERIES = ["7500R3", "7500R", "7280R3"]
-SUPPORT_VXLAN_COUNTERS_SERIES = ["7500R3", "7500R", "7280R3", "7050X3"]
-SUPPORT_VNI_COUNTERS_SERIES = ["7050X3"]
+SUPPORT_HARDWARE_COUNTERS_SERIES = ["7800R3", "7500R3", "7500R", "7280R3", "7280R2", "7280R"]
 
+# TODO: For chassis, check if the hardware commands return the linecard model. If so, we can remove the chassis families from the list.
 ARISTA_PLATFORMS = {
- "7050X3": {
- "chipset": "Trident3",
+ "7800R3": {
+ "chipsets": ["Ramon"],
  "families": [
- "7050TX3",
- "7050SX3",
- "7050CX3M",
+ "7800R3", # This is a linecard. The chipset is Jericho2
+ "7800R3A", # This is a linecard. The chipset is Jericho2C
+ "7800R3K", # This is a linecard. The chipset is Jericho2C
+ "7816LR3",
+ "7816R3",
+ "7812R3",
+ "7808R3",
+ "7804R3",
+ ],
+ },
+ "7500R3": {
+ "chipsets": ["Ramon"],
+ "families": [
+ "7500R3", # This is a linecard. The chipset is Jericho
+ "7500R3K", # This is a linecard. The chipset is Jericho2C
+ "7512R3",
+ "7508R3",
+ "7504R3",
+ ],
+ },
+ "7500R": {
+ "chipsets": ["FE3600"],
+ "families": [
+ "7500R2" # This is a linecard. The chipset is Jericho+
+ "7516R",
+ "7512R",
+ "7508R",
+ "7504R",
+ ],
+ },
+ "7300X3": {
+ "chipsets": ["Tomahawk"],
+ "families": [
+ "7300X3", # This is a linecard. The chipset is Trident3X7
+ "7308X3",
+ "7304X3",
+ ],
+ },
+ "7300X": {
+ "chipsets": ["Trident2"],
+ "families": [
+ "7300X" # This is a linecard. The chipset is Trident2
+ "7316X"
+ "7308X",
+ "7304X",
+ ],
+ },
+ "7388X5": {
+ "chipsets": ["Tomahawk4"],
+ "families": [
+ "7388X5",
+ ],
+ },
+ "7368X4": {
+ "chipsets": ["Tomahawk3"],
+ "families": [
+ "7368X4",
  ],
  },
  "7280R3": {
- "chipset": "Jericho2",
+ "chipsets": ["Jericho2", "Jericho2c", "Jericho2c+", "Qumran2C"],
  "families": [
  "7280DR3A",
  "7280R3A",
@@ -39,10 +92,142 @@
  "7280SR3",
  ],
  },
+ "7280R2": {
+ "chipsets": ["Jericho+", "Jericho2"],
+ "families": [
+ "7280CR2",
+ "7280SR2",
+ ],
+ },
+ "7280R": {
+ "chipsets": ["Jericho", "Qumran-MX", "Arad+"],
+ "families": [
+ "7280CR",
+ "7280QR",
+ "7280SE",
+ "7280SR",
+ "7280TR",
+ ],
+ },
+ "7170": {
+ "chipsets": ["Tofino", "Tofino2"],
+ "families": [
+ "7170",
+ "7170B",
+ ],
+ },
+ "7130": {
+ "chipsets": ["Jericho2"],
+ "families": [
+ "7130LBR",
+ "7130B",
+ "7132LB",
+ "7135LB",
+ "7130",
+ ],
+ },
+ "7060X": {
+ "chipsets": ["Tomahawk", "Tomahawk+", "Tomahawk3", "Tomahawk4"],
+ "families": [
+ "7060DX5",
+ "7060PX5",
+ "7060PX4",
+ "7060DX4",
+ "7060SX2",
+ "7060CX2",
+ "7060CX",
+ ],
+ },
+ "7260X": {
+ "chipsets": ["Tomahawk", "Tomahawk2"],
+ "families": [
+ "7260QX",
+ "7260CX3",
+ "7260CX",
+ ],
+ },
+ "7358X4": {
+ "chipsets": ["Trident4X11"],
+ "families": [
+ "7358X4",
+ ],
+ },
+ "7050X4": {
+ "chipsets": ["Trident4X9", "Trident4X11"],
+ "families": [
+ "7050PX4",
+ "7050DX4",
+ "7050SDX4",
+ "7050SPX4",
+ "7050CX4",
+ "7050CX4M",
+ ],
+ },
+ "7050X3": {
+ "chipsets": ["Trident3X5", "Trident3X7"],
+ "families": [
+ "7050TX3",
+ "7050SX3",
+ "7050CX3M",
+ ],
+ },
+ "7050X": {
+ "chipsets": ["Trident2", "Trident2+"],
+ "families": [
+ "7050QX",
+ ],
+ },
+ "7020TR": {
+ "chipsets": ["Qumran-AX"],
+ "families": [
+ "7020SR",
+ "7020TR",
+ "7020TRA",
+ ],
+ },
+ "7010TX": {
+ "chipsets": ["Trident3X2"],
+ "families": [
+ "7010TX",
+ ],
+ },
+ "750": {
+ "chipsets": ["Firelight"],
+ "families": [
+ "750X", # This is a linecard. The chipset is Firelight
+ ],
+ },
+ "720X": {
+ "chipsets": ["Trident3X3"],
+ "families": [
+ "720XP",
+ ],
+ },
+ "720D": {
+ "chipsets": ["Trident3X1, Trident3X2, Trident3X3"],
+ "families": [
+ "720DP",
+ "720DT",
+ "720DF",
+ ],
+ },
+ "722XPM": {
+ "chipsets": ["Trident3X2"],
+ "families": [
+ "722XPM",
+ ],
+ },
+ "710P": {
+ "chipsets": ["Trident3X1"],
+ "families": [
+ "710P",
+ ],
+ },
 }
 
-def get_model_series(model: str) -> str | None:
- """Get an Arista model series based on the model name."""
+
+def find_series_by_model(model: str) -> str | None:
+ """Get the series of a model based on the ARISTA_PLATFORMS dictionary."""
  for series, data in ARISTA_PLATFORMS.items():
  for family in data["families"]:
  if family in model: