diff --git a/anta/__init__.py b/anta/__init__.py index 58dcd2cf7..3eb0bb081 100644 --- a/anta/__init__.py +++ b/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." diff --git a/anta/cli/exec/utils.py b/anta/cli/exec/utils.py index 0537ee00d..f9035561b 100644 --- a/anta/cli/exec/utils.py +++ b/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() diff --git a/anta/decorators.py b/anta/decorators.py index 37f4fe2ff..b8a41184d 100644 --- a/anta/decorators.py +++ b/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,8 +166,8 @@ 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] @@ -100,7 +175,9 @@ async def wrapper(*args: Any, **kwargs: Any) -> TestResult: 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 diff --git a/anta/device.py b/anta/device.py index 193c16120..ec71943cd 100644 --- a/anta/device.py +++ b/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) diff --git a/anta/models.py b/anta/models.py index d213a70c4..9f99df875 100644 --- a/anta/models.py +++ b/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: diff --git a/anta/platform_utils.py b/anta/platform_utils.py new file mode 100644 index 000000000..bdb3d7038 --- /dev/null +++ b/anta/platform_utils.py @@ -0,0 +1,367 @@ +# 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. + +"""Arista platform utilities.""" + +from __future__ import annotations + +import logging +import re +from typing import Any + +from anta import GITHUB_SUGGESTION + +logger = logging.getLogger(__name__) + +SUPPORT_HARDWARE_COUNTERS_SERIES: list[str] = ["7800R3", "7500R3", "7500R", "7280R3", "7280R2", "7280R"] + +TRIDENT_SERIES: list[str] = ["7300X3", "7300X", "7358X4", "7050X4", "7050X3", "7050X", "7010TX", "720X", "720D", "722XPM", "710P"] + +VIRTUAL_PLATFORMS: list[str] = ["cEOSLab", "vEOS-lab", "cEOSCloudLab"] + +HARDWARE_PLATFORMS: list[dict[str, Any]] = [ + # Data center chassis/modular series. These series support different modules; linecards, supervisors, fabrics, etc. + # The modules of the system determine the series it belongs too, e.g. the 7800R3-36P-LC linecard belongs to the 7800R3 series. + { + "series": "7800R3", + "families": [ + "7800R3", + "7800R3K", + "7800R3A", + "7800R3AK", + ], + }, + { + "series": "7500R3", + "families": [ + "7500R3", + "7500R3K", + ], + }, + { + "series": "7500R", + "families": [ + "7500R", + "7500R2", + "7500R2A", + "7500R2AK", + "7500R2AM", + "7500R2M", + "7500RM", + ], + }, + { + "series": "7500E", + "families": [ + "7500E", + ], + }, + { + "series": "7300X3", + "families": [ + "7300X3", + ], + }, + { + "series": "7300X", + "families": [ + "7300X", + ], + }, + { + "series": "7388X5", + "families": [ + "7388", + "7388X5", + ], + }, + { + "series": "7368X4", + "families": [ + "7368", + "7368X4", + ], + }, + { + "series": "7358X4", + "families": [ + "7358", + "7358X4", + ], + }, + { + "series": "7280R3", + "families": [ + "7289", + "7289R3A", + "7289R3AK", + "7289R3AM", + ], + }, + # Data center fixed series. + { + "series": "7280R3", + "families": [ + "7280DR3A", + "7280DR3AM", + "7280DR3AK", + "7280CR3A", + "7280CR3AM", + "7280CR3AK", + "7280PR3", + "7280PR3K", + "7280DR3", + "7280DR3K", + "7280CR3MK", + "7280CR3", + "7280CR3K", + "7280SR3A", + "7280SR3AM", + "7280SR3AK", + "7280SR3M", + "7280SR3MK", + "7280SR3", + "7280SR3K", + "7280SR3E", + "7280TR3", + ], + }, + { + "series": "7280R2", + "families": [ + "7280CR2", + "7280CR2A", + "7280CR2K", + "7280CR2M", + "7280SR2", + "7280SR2A", + "7280SR2K", + ], + }, + { + "series": "7280R", + "families": [ + "7280CR", + "7280QR", + "7280QRA", + "7280SE", + "7280SR", + "7280SRA", + "7280SRAM", + "7280SRM", + "7280TR", + "7280TRA", + ], + }, + { + "series": "7170", + "families": [ + "7170", + "7170B", + ], + }, + { + "series": "7130", + "families": [ + "7130LBR", + "7130B", + "7132LB", + "7135LB", + "7130", + ], + }, + { + "series": "7060X", + "families": [ + "7060DX5", + "7060PX5", + "7060PX4", + "7060DX4", + "7060SX2", + "7060CX2", + "7060CX", + ], + }, + { + "series": "7260X", + "families": [ + "7260QX", + "7260CX3", + "7260CX", + ], + }, + { + "series": "7050X4", + "families": [ + "7050PX4", + "7050DX4", + "7050SDX4", + "7050SPX4", + "7050CX4", + "7050CX4M", + ], + }, + { + "series": "7050X3", + "families": [ + "7050TX3", + "7050SX3", + "7050CX3", + "7050CX3M", + ], + }, + { + "series": "7050X", + "families": [ + "7050QX", + ], + }, + { + "series": "7020TR", + "families": [ + "7020SR", + "7020TR", + "7020TRA", + ], + }, + { + "series": "7010TX", + "families": [ + "7010TX", + ], + }, + # Campus chassis/modular series. + { + "series": "750", + "families": [ + "750", + "750X", + ], + }, + # Campus fixed series. + { + "series": "720X", + "families": [ + "720XP", + ], + }, + { + "series": "720D", + "families": [ + "720DP", + "720DT", + "720DF", + ], + }, + { + "series": "722XPM", + "families": [ + "722XPM", + ], + }, + { + "series": "710P", + "families": [ + "710P", + ], + }, +] + + +def check_if_virtual_platform(platform: str) -> bool: + """Check if the platform is a virtual platform. + + Arguments: + ---------- + platform (str): The platform model to check. + + Returns + ------- + bool: True if the platform is a virtual platform, otherwise False. + + Examples + -------- + ```python + >>> from anta.platform_utils import check_if_virtual_platform + >>> check_if_virtual_platform(platform="vEOS-lab") + True + >>> check_if_virtual_platform(platform="DCS-7280CR3A-72-F") + False + ``` + """ + return platform in VIRTUAL_PLATFORMS + + +def find_series_by_modules(modules: dict[str, dict[str, Any]]) -> str | None: + """TODO: Document this function.""" + regex_pattern = r"\b(\d{3,}\w*)\b" + for module_data in modules.values(): + module_name = str(module_data.get("modelName")) + + # We extract the family from the module name, e.g. CCS-750X-48THP-LC -> 750X + match = re.search(regex_pattern, module_name) + if match: + platform_family = match.group(1) + else: + logger.warning("Module %s does not match the expected Arista product name pattern. %s", module_name, GITHUB_SUGGESTION) + continue + + for series in HARDWARE_PLATFORMS: + for family in series["families"]: + if family == platform_family: + return series["series"] + + # If no series is found, we need to add a new family to the HARDWARE_PLATFORMS dictionary + logger.warning("Module %s series was not found in the ANTA hardware platforms database. %s", module_name, GITHUB_SUGGESTION) + + return None + + +def find_series_by_platform(platform: str) -> str | None: + """Get the series of a platform based on the HARDWARE_PLATFORMS dictionary of this module. + + The function extract the family from the platform name and then searches for the series. + + If the platform is a virtual platform, the function will return the platform name as is. + + Arguments: + ---------- + platform (str): The platform model to find the series for. + + Returns + ------- + str | None: The series of the platform if found, otherwise None. + + Examples + -------- + ```python + >>> from anta.platform_utils import find_series_by_platform + >>> find_series_by_platform(platform="DCS-7280CR3A-72-F") + '7280R3' + >>> find_series_by_platform(platform="cEOSLab") + 'cEOSLab' + ``` + """ + # If the platform is a virtual platform, we return the platform name as is. + if check_if_virtual_platform(platform): + return platform + + # We extract the family from the platform name, e.g. DCS-7280CR3A-72-F -> 7280CR3A + regex_pattern = r"\b(\d{3,}\w*)\b" + match = re.search(regex_pattern, platform) + + if match: + platform_family = match.group(1) + else: + logger.warning("Platform %s does not match the expected Arista product name pattern. %s", platform, GITHUB_SUGGESTION) + return None + + for series in HARDWARE_PLATFORMS: + for family in series["families"]: + if family == platform_family: + return series["series"] + + # If no series is found, we need to add a new family to the HARDWARE_PLATFORMS dictionary + logger.warning("Platform %s series was not found in the ANTA hardware platforms database. %s", platform, GITHUB_SUGGESTION) + return None diff --git a/anta/tests/field_notices.py b/anta/tests/field_notices.py index 34ea1959e..1711d5244 100644 --- a/anta/tests/field_notices.py +++ b/anta/tests/field_notices.py @@ -7,8 +7,9 @@ from typing import TYPE_CHECKING, ClassVar -from anta.decorators import skip_on_platforms +from anta.decorators import platform_filter from anta.models import AntaCommand, AntaTest +from anta.platform_utils import VIRTUAL_PLATFORMS if TYPE_CHECKING: from anta.models import AntaTemplate @@ -39,7 +40,7 @@ class VerifyFieldNotice44Resolution(AntaTest): categories: ClassVar[list[str]] = ["field notices"] commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show version detail", revision=1)] - @skip_on_platforms(["cEOSLab", "vEOS-lab", "cEOSCloudLab"]) + @platform_filter(platforms=VIRTUAL_PLATFORMS, action="skip") @AntaTest.anta_test def test(self) -> None: """Main test function for VerifyFieldNotice44Resolution.""" @@ -143,7 +144,7 @@ class VerifyFieldNotice72Resolution(AntaTest): categories: ClassVar[list[str]] = ["field notices"] commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show version detail", revision=1)] - @skip_on_platforms(["cEOSLab", "vEOS-lab", "cEOSCloudLab"]) + @platform_filter(platforms=VIRTUAL_PLATFORMS, action="skip") @AntaTest.anta_test def test(self) -> None: """Main test function for VerifyFieldNotice72Resolution.""" diff --git a/anta/tests/hardware.py b/anta/tests/hardware.py index 569c180d7..cf5bfaf14 100644 --- a/anta/tests/hardware.py +++ b/anta/tests/hardware.py @@ -9,8 +9,9 @@ from typing import TYPE_CHECKING, ClassVar -from anta.decorators import skip_on_platforms +from anta.decorators import platform_filter, platform_series_filter from anta.models import AntaCommand, AntaTest +from anta.platform_utils import SUPPORT_HARDWARE_COUNTERS_SERIES, VIRTUAL_PLATFORMS if TYPE_CHECKING: from anta.models import AntaTemplate @@ -47,7 +48,7 @@ class Input(AntaTest.Input): manufacturers: list[str] """List of approved transceivers manufacturers.""" - @skip_on_platforms(["cEOSLab", "vEOS-lab", "cEOSCloudLab"]) + @platform_filter(platforms=VIRTUAL_PLATFORMS, action="skip") @AntaTest.anta_test def test(self) -> None: """Main test function for VerifyTransceiversManufacturers.""" @@ -82,7 +83,7 @@ class VerifyTemperature(AntaTest): categories: ClassVar[list[str]] = ["hardware"] commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show system environment temperature", revision=1)] - @skip_on_platforms(["cEOSLab", "vEOS-lab", "cEOSCloudLab"]) + @platform_filter(platforms=VIRTUAL_PLATFORMS, action="skip") @AntaTest.anta_test def test(self) -> None: """Main test function for VerifyTemperature.""" @@ -115,7 +116,7 @@ class VerifyTransceiversTemperature(AntaTest): categories: ClassVar[list[str]] = ["hardware"] commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show system environment temperature transceiver", revision=1)] - @skip_on_platforms(["cEOSLab", "vEOS-lab", "cEOSCloudLab"]) + @platform_filter(platforms=VIRTUAL_PLATFORMS, action="skip") @AntaTest.anta_test def test(self) -> None: """Main test function for VerifyTransceiversTemperature.""" @@ -156,7 +157,7 @@ class VerifyEnvironmentSystemCooling(AntaTest): categories: ClassVar[list[str]] = ["hardware"] commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show system environment cooling", revision=1)] - @skip_on_platforms(["cEOSLab", "vEOS-lab", "cEOSCloudLab"]) + @platform_filter(platforms=VIRTUAL_PLATFORMS, action="skip") @AntaTest.anta_test def test(self) -> None: """Main test function for VerifyEnvironmentSystemCooling.""" @@ -196,7 +197,7 @@ class Input(AntaTest.Input): states: list[str] """List of accepted states of fan status.""" - @skip_on_platforms(["cEOSLab", "vEOS-lab", "cEOSCloudLab"]) + @platform_filter(platforms=VIRTUAL_PLATFORMS, action="skip") @AntaTest.anta_test def test(self) -> None: """Main test function for VerifyEnvironmentCooling.""" @@ -243,7 +244,7 @@ class Input(AntaTest.Input): states: list[str] """List of accepted states list of power supplies status.""" - @skip_on_platforms(["cEOSLab", "vEOS-lab", "cEOSCloudLab"]) + @platform_filter(platforms=VIRTUAL_PLATFORMS, action="skip") @AntaTest.anta_test def test(self) -> None: """Main test function for VerifyEnvironmentPower.""" @@ -259,7 +260,9 @@ def test(self) -> None: class VerifyAdverseDrops(AntaTest): - """Verifies there are no adverse drops on DCS-7280 and DCS-7500 family switches (Arad/Jericho chips). + """Verifies there are no adverse drops on the Arad/Jericho chips switches. + + The following series will be tested: DCS-7800R3, DCS-7500R3, DCS-7500R, DCS-7280R3, DCS-7280R2, DCS-7280R. Expected Results ---------------- @@ -275,11 +278,11 @@ class VerifyAdverseDrops(AntaTest): """ name = "VerifyAdverseDrops" - description = "Verifies there are no adverse drops on DCS-7280 and DCS-7500 family switches." + description = "Verifies there are no adverse drops on the Arad/Jericho chips switches." categories: ClassVar[list[str]] = ["hardware"] commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show hardware counter drop", revision=1)] - @skip_on_platforms(["cEOSLab", "vEOS-lab", "cEOSCloudLab"]) + @platform_series_filter(series=SUPPORT_HARDWARE_COUNTERS_SERIES, action="run") @AntaTest.anta_test def test(self) -> None: """Main test function for VerifyAdverseDrops.""" diff --git a/anta/tests/interfaces.py b/anta/tests/interfaces.py index 738fc1e78..481791748 100644 --- a/anta/tests/interfaces.py +++ b/anta/tests/interfaces.py @@ -15,8 +15,9 @@ from pydantic_extra_types.mac_address import MacAddress from anta.custom_types import Interface, Percent, PositiveInteger -from anta.decorators import skip_on_platforms +from anta.decorators import platform_filter from anta.models import AntaCommand, AntaTemplate, AntaTest +from anta.platform_utils import VIRTUAL_PLATFORMS from anta.tools.get_item import get_item from anta.tools.get_value import get_value @@ -295,7 +296,7 @@ class VerifyStormControlDrops(AntaTest): categories: ClassVar[list[str]] = ["interfaces"] commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show storm-control", revision=1)] - @skip_on_platforms(["cEOSLab", "vEOS-lab", "cEOSCloudLab"]) + @platform_filter(platforms=VIRTUAL_PLATFORMS, action="skip") @AntaTest.anta_test def test(self) -> None: """Main test function for VerifyStormControlDrops.""" diff --git a/anta/tests/lanz.py b/anta/tests/lanz.py index dcdab69db..fd3e0e360 100644 --- a/anta/tests/lanz.py +++ b/anta/tests/lanz.py @@ -7,8 +7,9 @@ from typing import TYPE_CHECKING, ClassVar -from anta.decorators import skip_on_platforms +from anta.decorators import platform_filter from anta.models import AntaCommand, AntaTest +from anta.platform_utils import VIRTUAL_PLATFORMS if TYPE_CHECKING: from anta.models import AntaTemplate @@ -35,7 +36,7 @@ class VerifyLANZ(AntaTest): categories: ClassVar[list[str]] = ["lanz"] commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show queue-monitor length status", revision=1)] - @skip_on_platforms(["cEOSLab", "vEOS-lab", "cEOSCloudLab"]) + @platform_filter(platforms=VIRTUAL_PLATFORMS, action="skip") @AntaTest.anta_test def test(self) -> None: """Main test function for VerifyLANZ.""" diff --git a/anta/tests/profiles.py b/anta/tests/profiles.py index 859c8866c..441de6123 100644 --- a/anta/tests/profiles.py +++ b/anta/tests/profiles.py @@ -9,8 +9,9 @@ from typing import TYPE_CHECKING, ClassVar, Literal -from anta.decorators import skip_on_platforms +from anta.decorators import platform_filter, platform_series_filter from anta.models import AntaCommand, AntaTest +from anta.platform_utils import TRIDENT_SERIES, VIRTUAL_PLATFORMS if TYPE_CHECKING: from anta.models import AntaTemplate @@ -19,6 +20,8 @@ class VerifyUnifiedForwardingTableMode(AntaTest): """Verifies the device is using the expected UFT (Unified Forwarding Table) mode. + The test will run on Trident platform series only. + Expected Results ---------------- * Success: The test will pass if the device is using the expected UFT mode. @@ -44,7 +47,7 @@ class Input(AntaTest.Input): mode: Literal[0, 1, 2, 3, 4, "flexible"] """Expected UFT mode. Valid values are 0, 1, 2, 3, 4, or "flexible".""" - @skip_on_platforms(["cEOSLab", "vEOS-lab", "cEOSCloudLab"]) + @platform_series_filter(series=TRIDENT_SERIES, action="run") @AntaTest.anta_test def test(self) -> None: """Main test function for VerifyUnifiedForwardingTableMode.""" @@ -83,7 +86,7 @@ class Input(AntaTest.Input): profile: str """Expected TCAM profile.""" - @skip_on_platforms(["cEOSLab", "vEOS-lab", "cEOSCloudLab"]) + @platform_filter(platforms=VIRTUAL_PLATFORMS, action="skip") @AntaTest.anta_test def test(self) -> None: """Main test function for VerifyTcamProfile.""" diff --git a/anta/tests/ptp.py b/anta/tests/ptp.py index eabda8835..1fe4e36a9 100644 --- a/anta/tests/ptp.py +++ b/anta/tests/ptp.py @@ -9,8 +9,9 @@ from typing import TYPE_CHECKING, ClassVar -from anta.decorators import skip_on_platforms +from anta.decorators import platform_filter from anta.models import AntaCommand, AntaTest +from anta.platform_utils import VIRTUAL_PLATFORMS if TYPE_CHECKING: from anta.models import AntaTemplate @@ -38,7 +39,7 @@ class VerifyPtpModeStatus(AntaTest): categories: ClassVar[list[str]] = ["ptp"] commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show ptp", revision=2)] - @skip_on_platforms(["cEOSLab", "vEOS-lab", "cEOSCloudLab"]) + @platform_filter(platforms=VIRTUAL_PLATFORMS, action="skip") @AntaTest.anta_test def test(self) -> None: """Main test function for VerifyPtpModeStatus.""" @@ -85,7 +86,7 @@ class Input(AntaTest.Input): categories: ClassVar[list[str]] = ["ptp"] commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show ptp", revision=2)] - @skip_on_platforms(["cEOSLab", "vEOS-lab", "cEOSCloudLab"]) + @platform_filter(platforms=VIRTUAL_PLATFORMS, action="skip") @AntaTest.anta_test def test(self) -> None: """Main test function for VerifyPtpGMStatus.""" @@ -125,7 +126,7 @@ class VerifyPtpLockStatus(AntaTest): categories: ClassVar[list[str]] = ["ptp"] commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show ptp", revision=2)] - @skip_on_platforms(["cEOSLab", "vEOS-lab", "cEOSCloudLab"]) + @platform_filter(platforms=VIRTUAL_PLATFORMS, action="skip") @AntaTest.anta_test def test(self) -> None: """Main test function for VerifyPtpLockStatus.""" @@ -166,7 +167,7 @@ class VerifyPtpOffset(AntaTest): categories: ClassVar[list[str]] = ["ptp"] commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show ptp monitor", revision=1)] - @skip_on_platforms(["cEOSLab", "vEOS-lab", "cEOSCloudLab"]) + @platform_filter(platforms=VIRTUAL_PLATFORMS, action="skip") @AntaTest.anta_test def test(self) -> None: """Main test function for VerifyPtpOffset.""" @@ -211,7 +212,7 @@ class VerifyPtpPortModeStatus(AntaTest): categories: ClassVar[list[str]] = ["ptp"] commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show ptp", revision=2)] - @skip_on_platforms(["cEOSLab", "vEOS-lab", "cEOSCloudLab"]) + @platform_filter(platforms=VIRTUAL_PLATFORMS, action="skip") @AntaTest.anta_test def test(self) -> None: """Main test function for VerifyPtpPortModeStatus.""" diff --git a/docs/advanced_usages/custom-tests.md b/docs/advanced_usages/custom-tests.md index a3ef035d9..45688b682 100644 --- a/docs/advanced_usages/custom-tests.md +++ b/docs/advanced_usages/custom-tests.md @@ -17,7 +17,8 @@ ANTA provides an abstract class [AntaTest](../api/models.md#anta.models.AntaTest ```python from anta.models import AntaTest, AntaCommand -from anta.decorators import skip_on_platforms +from anta.decorators import platform_filters +from anta.platform_utils import VIRTUAL_PLATFORMS class VerifyTemperature(AntaTest): @@ -41,7 +42,7 @@ class VerifyTemperature(AntaTest): categories: ClassVar[list[str]] = ["hardware"] commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show system environment temperature", revision=1)] - @skip_on_platforms(["cEOSLab", "vEOS-lab", "cEOSCloudLab"]) + @platform_filter(platforms=VIRTUAL_PLATFORMS, action="skip") @AntaTest.anta_test def test(self) -> None: """Main test function for VerifyTemperature.""" diff --git a/tests/units/test_models.py b/tests/units/test_models.py index 11775e60b..c59c503bd 100644 --- a/tests/units/test_models.py +++ b/tests/units/test_models.py @@ -11,7 +11,7 @@ import pytest -from anta.decorators import deprecated_test, skip_on_platforms +from anta.decorators import deprecated_test, platform_filter from anta.models import AntaCommand, AntaTemplate, AntaTest from tests.lib.fixture import DEVICE_HW_MODEL from tests.lib.utils import generate_test_ids @@ -196,7 +196,7 @@ class SkipOnPlatformTest(AntaTest): categories: ClassVar[list[str]] = [] commands: ClassVar[list[AntaCommand | AntaTemplate]] = [] - @skip_on_platforms([DEVICE_HW_MODEL]) + @platform_filter([DEVICE_HW_MODEL], action="skip") @AntaTest.anta_test def test(self) -> None: """Test function.""" @@ -211,7 +211,7 @@ class UnSkipOnPlatformTest(AntaTest): categories: ClassVar[list[str]] = [] commands: ClassVar[list[AntaCommand | AntaTemplate]] = [] - @skip_on_platforms(["dummy"]) + @platform_filter(["dummy"], action="skip") @AntaTest.anta_test def test(self) -> None: """Test function.""" @@ -231,7 +231,7 @@ class Input(AntaTest.Input): string: str - @skip_on_platforms([DEVICE_HW_MODEL]) + @platform_filter([DEVICE_HW_MODEL], action="skip") @AntaTest.anta_test def test(self) -> None: """Test function."""