fix(anta): Fix anta exec clear-counters by adding supported platforms

anta/__init__.py

@@ -45,4 +45,4 @@ class RICH_COLOR_PALETTE:
  "unset": RICH_COLOR_PALETTE.UNSET,
 }
 
-GITHUB_SUGGESTION = "Please reach out to the maintainer team or open an issue on Github: https://github.com/arista-netdevops-community/anta."
+GITHUB_SUGGESTION = "Please reach out to the maintainer team or open an issue on GitHub: https://github.com/arista-netdevops-community/anta."

anta/cli/exec/utils.py

@@ -20,6 +20,7 @@
 
 from anta.device import AntaDevice, AsyncEOSDevice
 from anta.models import AntaCommand
+from anta.platform_utils import SUPPORT_HARDWARE_COUNTERS_SERIES
 
 if TYPE_CHECKING:
  from anta.inventory import AntaInventory
@@ -30,17 +31,36 @@
 
 
 async def clear_counters_utils(anta_inventory: AntaInventory, tags: list[str] | None = None) -> None:
- """Clear counters."""
+ """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.
+
+ """
+
+ async def clear(device: AntaDevice) -> None:
+ 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
 
- async def clear(dev: AntaDevice) -> None:
  commands = [AntaCommand(command="clear counters")]
- if dev.hw_model not in ["cEOSLab", "vEOS-lab"]:
+
+ if device.hw_series is None:
+ logger.debug("Could not clear hardware counters on device %s because its platform series is not set and the command might be unsupported.", device.name)
+ elif device.hw_series in SUPPORT_HARDWARE_COUNTERS_SERIES:
  commands.append(AntaCommand(command="clear hardware counter drop"))
- await dev.collect_commands(commands=commands)
+
+ await device.collect_commands(commands=commands)
+
  for command in commands:
  if not command.collected:
- logger.error("Could not clear counters on device %s: %s", dev.name, command.errors)
- logger.info("Cleared counters on %s (%s)", dev.name, dev.hw_model)
+ logger.error("Could not '%s' on device %s: %s", command.command, device.name, command.errors)
+ else:
+ logger.info("Successfully '%s' on device %s (%s)", command.command, device.name, device.hw_model)
 
  logger.info("Connecting to devices...")
  await anta_inventory.connect_inventory()

anta/decorators.py

@@ -6,7 +6,7 @@
 from __future__ import annotations
 
 from functools import wraps
-from typing import TYPE_CHECKING, Any, Callable, TypeVar, cast
+from typing import TYPE_CHECKING, Any, Callable, Literal, TypeVar, cast
 
 from anta.models import AntaTest, logger
 
@@ -58,21 +58,96 @@ async def wrapper(*args: Any, **kwargs: Any) -> Any:
  return decorator
 
 
-def skip_on_platforms(platforms: list[str]) -> Callable[[F], F]:
+def platform_series_filter(series: list[str], action: Literal["run", "skip"]) -> Callable[[F], F]:
+ """Return a decorator to run or skip 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 either be run or skip depending on the action.
+
+ Args:
+ ----
+ series (list[str]): List of platform series on which the test should be run.
+ action (Literal["run", "skip"]): The action to take if the device's hardware model is part of the provided series. It can be either "run" or "skip".
+
+ 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
+ @platform_series_filter(series=["7800R3", "7500R3", "7500R", "7280R3"], action="run")
+ @AntaTest.anta_test
+ def test(self) -> None:
+ pass
+ """
+ if action not in ["run", "skip"]:
+ msg = f"Improper way of using the platform series filter decorator function. Action must be 'run' or 'skip', not '{action}'."
+ raise ValueError(msg)
+
+ 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 finds the series of the device the test is run on, checks if it is part of the specified series
+ and either runs or skips the test based on the action.
+ """
+ anta_test = args[0]
+
+ if anta_test.result.result != "unset":
+ AntaTest.update_progress()
+ return anta_test.result
+
+ if anta_test.device.hw_series is None:
+ msg = f"Platform series filter is ignored for test {anta_test.__class__.__name__} since the device's platform series is not set."
+ logger.warning(msg)
+
+ elif (action == "run" and anta_test.device.hw_series not in series) or (action == "skip" and anta_test.device.hw_series 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
+
+ return await function(*args, **kwargs)
+
+ return cast(F, wrapper)
+
+ return decorator
+
+
+def platform_filter(platforms: list[str], action: Literal["run", "skip"]) -> Callable[[F], F]:
  """Return a decorator to skip 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 in the list of platforms specified, the test will be skipped.
+ the test is run on. If the model is in the list of platforms specified, the test will either be run or skip depending on the action.
 
  Args:
  ----
  platforms (list[str]): List of hardware models on which the test should be skipped.
+ action (Literal["run", "skip"]): The action to take if the device's hardware model is part of the provided platforms. It can be either "run" or "skip".
 
  Returns
  -------
  Callable[[F], F]: A decorator that can be used to wrap test functions.
 
  """
+ if action not in ["run", "skip"]:
+ msg = f"Improper way of using the platform filter decorator function. Action must be 'run' or 'skip', not '{action}'."
+ raise ValueError(msg)
 
  def decorator(function: F) -> F:
  """Actual decorator that either runs the test or skips it based on the device's hardware model.
@@ -91,16 +166,18 @@ def decorator(function: F) -> F:
  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.
+ This wrapper checks if the hardware model of the device the test is run on is part of the specified platforms
+ and either runs or skips the test based on the action.
  """
  anta_test = args[0]
 
  if anta_test.result.result != "unset":
  AntaTest.update_progress()
  return anta_test.result
 
- if anta_test.device.hw_model in platforms:
+ platform = anta_test.device.hw_model
+
+ if (action == "run" and platform not in platforms) or (action == "skip" and platform in platforms):
  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

anta/device.py

@@ -17,8 +17,9 @@
 from asyncssh import SSHClientConnection, SSHClientConnectionOptions
 from httpx import ConnectError, HTTPError
 
-from anta import __DEBUG__, aioeapi
+from anta import __DEBUG__, GITHUB_SUGGESTION, aioeapi
 from anta.logger import exc_to_str
+from anta.platform_utils import find_series_by_modules, find_series_by_platform
 
 if TYPE_CHECKING:
  from collections.abc import Iterator
@@ -29,7 +30,7 @@
 logger = logging.getLogger(__name__)
 
 
-class AntaDevice(ABC):
+class AntaDevice(ABC): # pylint: disable=R0902
  """Abstract class representing a device in ANTA.
 
  An implementation of this class must override the abstract coroutines `_collect()` and
@@ -41,6 +42,7 @@ class AntaDevice(ABC):
  is_online: True if the device IP is reachable and a port can be open
  established: True if remote command execution succeeds
  hw_model: Hardware model of the device
+ hw_series: Hardware series of the device
  tags: List of tags for this device
  cache: In-memory cache from aiocache library for this device (None if cache is disabled)
  cache_locks: Dictionary mapping keys to asyncio locks to guarantee exclusive access to the cache if not disabled
@@ -59,6 +61,7 @@ def __init__(self, name: str, tags: list[str] | None = None, *, disable_cache: b
  """
  self.name: str = name
  self.hw_model: str | None = None
+ self.hw_series: str | None = None
  self.tags: list[str] = tags if tags is not None else []
  # A device always has its own name as tag
  self.tags.append(self.name)
@@ -207,7 +210,7 @@ async def copy(self, sources: list[Path], destination: Path, direction: Literal[
  raise NotImplementedError(msg)
 
 
-class AsyncEOSDevice(AntaDevice):
+class AsyncEOSDevice(AntaDevice): # pylint: disable=R0902
  """Implementation of AntaDevice for EOS using aio-eapi.
 
  Attributes
@@ -344,6 +347,8 @@ async def _collect(self, command: AntaCommand) -> None:
  command.errors = e.errors
  if self.supports(command):
  logger.error("Command '%s' failed on %s", command.command, self.name)
+ else:
+ logger.warning("Command '%s' is not supported on %s. %s", command.command, self.hw_model, GITHUB_SUGGESTION)
  except (HTTPError, ConnectError) as e:
  command.errors = [str(e)]
  logger.error("Cannot connect to device %s", self.name)
@@ -359,25 +364,31 @@ async def refresh(self) -> None:
  - is_online: When a device IP is reachable and a port can be open
  - established: When a command execution succeeds
  - hw_model: The hardware model of the device
+ - hw_series: The hardware series of the device
  """
  logger.debug("Refreshing device %s", self.name)
  self.is_online = await self._session.check_connection()
  if self.is_online:
- show_version = "show version"
- hw_model_key = "modelName"
  try:
- response = await self._session.cli(command=show_version)
- except aioeapi.EapiCommandError as e:
- logger.warning("Cannot get hardware information from device %s: %s", self.name, e.errmsg)
-
+ show_version = await self._session.cli(command="show version")
+ if (series := find_series_by_platform(show_version["modelName"])) is None:
+ logger.debug("Device %s is potentially a chassis, trying to get the platform series from the modules", self.name)
+ show_module = await self._session.cli(command="show module")
+ series = find_series_by_modules(show_module["modules"])
+
+ except (aioeapi.EapiCommandError, KeyError) as e:
+ error_msg = "Cannot parse 'show version' or 'show module'" if isinstance(e, KeyError) else e.errmsg
+ logger.warning("Cannot get hardware information from device %s: %s", self.name, error_msg)
  except (HTTPError, ConnectError) as e:
  logger.warning("Cannot get hardware information from device %s: %s", self.name, exc_to_str(e))
 
  else:
- if hw_model_key in response:
- self.hw_model = response[hw_model_key]
+ self.hw_model = show_version["modelName"]
+ if series is None:
+ logger.warning("Cannot find the platform series of device %s (%s). %s", self.name, self.hw_model, GITHUB_SUGGESTION)
  else:
- logger.warning("Cannot get hardware information from device %s: cannot parse '%s'", self.name, show_version)
+ self.hw_series = series
+ logger.debug("Device platform series %s found for device %s", self.hw_series, self.name)
 
  else:
  logger.warning("Could not connect to device %s: cannot open eAPI port", self.name)

anta/models.py

@@ -522,15 +522,13 @@ def format_td(seconds: float, digits: int = 3) -> str:
  return self.result
 
  if cmds := self.failed_commands:
- self.logger.debug(self.device.supports)
- unsupported_commands = [f"Skipped because {c.command} is not supported on {self.device.hw_model}" for c in cmds if not self.device.supports(c)]
- self.logger.debug(unsupported_commands)
+ unsupported_commands = [f"Skipped because '{c.command}' is not supported on {self.device.hw_model}" for c in cmds if not self.device.supports(c)]
  if unsupported_commands:
- msg = f"Test {self.name} has been skipped because it is not supported on {self.device.hw_model}: {GITHUB_SUGGESTION}"
- self.logger.warning(msg)
+ self.logger.debug(unsupported_commands)
  self.result.is_skipped("\n".join(unsupported_commands))
- return self.result
- self.result.is_error(message="\n".join([f"{c.command} has failed: {', '.join(c.errors)}" for c in cmds]))
+ else:
+ self.result.is_error(message="\n".join([f"{c.command} has failed: {', '.join(c.errors)}" for c in cmds]))
+
  return self.result
 
  try: