From 1f033f433cac808fa60c9232b089075a3a5e5a1e Mon Sep 17 00:00:00 2001 From: Guillaume Mulocher Date: Mon, 6 Jan 2025 19:56:57 +0100 Subject: [PATCH 1/8] test: idempotency of setup logger (#981) --------- Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com> --- anta/logger.py | 58 +++++++++++++++++++++++++++++--------- tests/units/test_logger.py | 45 ++++++++++++++++++++++++++++- 2 files changed, 88 insertions(+), 15 deletions(-) diff --git a/anta/logger.py b/anta/logger.py index 167b821b1..e6d04287b 100644 --- a/anta/logger.py +++ b/anta/logger.py @@ -9,15 +9,13 @@ import traceback from datetime import timedelta from enum import Enum -from typing import TYPE_CHECKING, Literal +from pathlib import Path +from typing import Literal from rich.logging import RichHandler from anta import __DEBUG__ -if TYPE_CHECKING: - from pathlib import Path - logger = logging.getLogger(__name__) @@ -69,27 +67,59 @@ def setup_logging(level: LogLevel = Log.INFO, file: Path | None = None) -> None: # httpx as well logging.getLogger("httpx").setLevel(logging.WARNING) - # Add RichHandler for stdout - rich_handler = RichHandler(markup=True, rich_tracebacks=True, tracebacks_show_locals=False) - # Show Python module in stdout at DEBUG level - fmt_string = "[grey58]\\[%(name)s][/grey58] %(message)s" if loglevel == logging.DEBUG else "%(message)s" - formatter = logging.Formatter(fmt=fmt_string, datefmt="[%X]") - rich_handler.setFormatter(formatter) - root.addHandler(rich_handler) - # Add FileHandler if file is provided - if file: + # Add RichHandler for stdout if not already present + _maybe_add_rich_handler(loglevel, root) + + # Add FileHandler if file is provided and same File Handler is not already present + if file and not _get_file_handler(root, file): file_handler = logging.FileHandler(file) formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s") file_handler.setFormatter(formatter) root.addHandler(file_handler) # If level is DEBUG and file is provided, do not send DEBUG level to stdout - if loglevel == logging.DEBUG: + if loglevel == logging.DEBUG and (rich_handler := _get_rich_handler(root)) is not None: rich_handler.setLevel(logging.INFO) if __DEBUG__: logger.debug("ANTA Debug Mode enabled") +def _get_file_handler(logger_instance: logging.Logger, file: Path) -> logging.FileHandler | None: + """Return the FileHandler if present.""" + return ( + next( + ( + handler + for handler in logger_instance.handlers + if isinstance(handler, logging.FileHandler) and str(Path(handler.baseFilename).resolve()) == str(file.resolve()) + ), + None, + ) + if logger_instance.hasHandlers() + else None + ) + + +def _get_rich_handler(logger_instance: logging.Logger) -> logging.Handler | None: + """Return the ANTA Rich Handler.""" + return next((handler for handler in logger_instance.handlers if handler.get_name() == "ANTA_RICH_HANDLER"), None) if logger_instance.hasHandlers() else None + + +def _maybe_add_rich_handler(loglevel: int, logger_instance: logging.Logger) -> None: + """Add RichHandler for stdout if not already present.""" + if _get_rich_handler(logger_instance) is not None: + # Nothing to do. + return + + anta_rich_handler = RichHandler(markup=True, rich_tracebacks=True, tracebacks_show_locals=False) + anta_rich_handler.set_name("ANTA_RICH_HANDLER") + # Show Python module in stdout at DEBUG level + fmt_string = "[grey58]\\[%(name)s][/grey58] %(message)s" if loglevel == logging.DEBUG else "%(message)s" + formatter = logging.Formatter(fmt=fmt_string, datefmt="[%X]") + anta_rich_handler.setFormatter(formatter) + logger_instance.addHandler(anta_rich_handler) + + def format_td(seconds: float, digits: int = 3) -> str: """Return a formatted string from a float number representing seconds and a number of digits.""" isec, fsec = divmod(round(seconds * 10**digits), 10**digits) diff --git a/tests/units/test_logger.py b/tests/units/test_logger.py index d26932001..a8f0bc794 100644 --- a/tests/units/test_logger.py +++ b/tests/units/test_logger.py @@ -6,11 +6,54 @@ from __future__ import annotations import logging +from pathlib import Path from unittest.mock import patch import pytest -from anta.logger import anta_log_exception, exc_to_str, tb_to_str +from anta.logger import Log, LogLevel, _get_file_handler, _get_rich_handler, anta_log_exception, exc_to_str, setup_logging, tb_to_str + + +@pytest.mark.parametrize( + ("level", "path", "debug_value"), + [ + pytest.param(Log.INFO, None, False, id="INFO no file"), + pytest.param(Log.DEBUG, None, False, id="DEBUG no file"), + pytest.param(Log.INFO, Path("/tmp/file.log"), False, id="INFO file"), + pytest.param(Log.DEBUG, Path("/tmp/file.log"), False, id="DEBUG file"), + pytest.param(Log.INFO, None, True, id="INFO no file __DEBUG__ set"), + pytest.param(Log.DEBUG, None, True, id="INFO no file __DEBUG__ set"), + ], +) +def test_setup_logging(level: LogLevel, path: Path | None, debug_value: bool) -> None: + """Test setup_logging.""" + # Clean up any logger on root + root = logging.getLogger() + if root.hasHandlers(): + root.handlers = [] + + with patch("anta.logger.__DEBUG__", new=debug_value): + setup_logging(level, path) + + rich_handler = _get_rich_handler(root) + assert rich_handler is not None + + # When __DEBUG__ is True, the log level is overwritten to DEBUG + if debug_value: + assert root.level == logging.DEBUG + if path is not None: + assert rich_handler.level == logging.INFO + + if path is not None: + assert _get_file_handler(root, path) is not None + expected_handlers = 2 + else: + expected_handlers = 1 + assert len(root.handlers) == expected_handlers + + # Check idempotency + setup_logging(level, path) + assert len(root.handlers) == expected_handlers @pytest.mark.parametrize( From b45e87afe01a53a18e3e0c0cb89a09bdd2aad9ca Mon Sep 17 00:00:00 2001 From: "pre-commit-ci[bot]" <66853113+pre-commit-ci[bot]@users.noreply.github.com> Date: Mon, 6 Jan 2025 20:37:41 +0100 Subject: [PATCH 2/8] ci: pre-commit autoupdate (#992) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit updates: - [github.com/astral-sh/ruff-pre-commit: v0.8.4 → v0.8.6](https://github.com/astral-sh/ruff-pre-commit/compare/v0.8.4...v0.8.6) - [github.com/pre-commit/mirrors-mypy: v1.14.0 → v1.14.1](https://github.com/pre-commit/mirrors-mypy/compare/v1.14.0...v1.14.1) Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com> Co-authored-by: Matthieu Tâche --- .pre-commit-config.yaml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 36d721b5e..f12dda006 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -46,7 +46,7 @@ repos: - '' - repo: https://github.com/astral-sh/ruff-pre-commit - rev: v0.8.4 + rev: v0.8.6 hooks: - id: ruff name: Run Ruff linter @@ -85,7 +85,7 @@ repos: types: [text] - repo: https://github.com/pre-commit/mirrors-mypy - rev: v1.14.0 + rev: v1.14.1 hooks: - id: mypy name: Check typing with mypy From d21e44354d0c4a84c90ebce70444a87ff04a060c Mon Sep 17 00:00:00 2001 From: vitthalmagadum <122079046+vitthalmagadum@users.noreply.github.com> Date: Wed, 8 Jan 2025 18:52:41 +0530 Subject: [PATCH 3/8] refactor(anta): Refactor the VerifyBFDPeersIntervals test to add detection time (#858) --- anta/input_models/bfd.py | 4 + anta/tests/bfd.py | 9 +++ examples/tests.yaml | 1 + tests/units/anta_tests/test_bfd.py | 119 ++++++++++++++++++++++++++++- 4 files changed, 131 insertions(+), 2 deletions(-) diff --git a/anta/input_models/bfd.py b/anta/input_models/bfd.py index 2d057d670..06838d0e7 100644 --- a/anta/input_models/bfd.py +++ b/anta/input_models/bfd.py @@ -31,6 +31,10 @@ class BFDPeer(BaseModel): """Multiplier of BFD peer. Required field in the `VerifyBFDPeersIntervals` test.""" protocols: list[BfdProtocol] | None = None """List of protocols to be verified. Required field in the `VerifyBFDPeersRegProtocols` test.""" + detection_time: int | None = None + """Detection time of BFD peer in milliseconds. Defines how long to wait without receiving BFD packets before declaring the peer session as down. + + Optional field in the `VerifyBFDPeersIntervals` test.""" def __str__(self) -> str: """Return a human-readable string representation of the BFDPeer for reporting.""" diff --git a/anta/tests/bfd.py b/anta/tests/bfd.py index a27a786cf..861a6a2e4 100644 --- a/anta/tests/bfd.py +++ b/anta/tests/bfd.py @@ -99,15 +99,18 @@ class VerifyBFDPeersIntervals(AntaTest): 1. Confirms that the specified VRF is configured. 2. Verifies that the peer exists in the BFD configuration. 3. Confirms that BFD peer is correctly configured with the `Transmit interval, Receive interval and Multiplier`. + 4. Verifies that BFD peer is correctly configured with the `Detection time`, if provided. Expected Results ---------------- * Success: If all of the following conditions are met: - All specified peers are found in the BFD configuration within the specified VRF. - All BFD peers are correctly configured with the `Transmit interval, Receive interval and Multiplier`. + - If provided, the `Detection time` is correctly configured. * Failure: If any of the following occur: - A specified peer is not found in the BFD configuration within the specified VRF. - Any BFD peer not correctly configured with the `Transmit interval, Receive interval and Multiplier`. + - Any BFD peer is not correctly configured with `Detection time`, if provided. Examples -------- @@ -125,6 +128,7 @@ class VerifyBFDPeersIntervals(AntaTest): tx_interval: 1200 rx_interval: 1200 multiplier: 3 + detection_time: 3600 ``` """ @@ -151,6 +155,7 @@ def test(self) -> None: tx_interval = bfd_peer.tx_interval rx_interval = bfd_peer.rx_interval multiplier = bfd_peer.multiplier + detect_time = bfd_peer.detection_time # Check if BFD peer configured bfd_output = get_value( @@ -166,6 +171,7 @@ def test(self) -> None: bfd_details = bfd_output.get("peerStatsDetail", {}) op_tx_interval = bfd_details.get("operTxInterval") // 1000 op_rx_interval = bfd_details.get("operRxInterval") // 1000 + op_detection_time = bfd_details.get("detectTime") // 1000 detect_multiplier = bfd_details.get("detectMult") if op_tx_interval != tx_interval: @@ -177,6 +183,9 @@ def test(self) -> None: if detect_multiplier != multiplier: self.result.is_failure(f"{bfd_peer} - Incorrect Multiplier - Expected: {multiplier} Actual: {detect_multiplier}") + if detect_time and op_detection_time != detect_time: + self.result.is_failure(f"{bfd_peer} - Incorrect Detection Time - Expected: {detect_time} Actual: {op_detection_time}") + class VerifyBFDPeersHealth(AntaTest): """Verifies the health of IPv4 BFD peers across all VRFs. diff --git a/examples/tests.yaml b/examples/tests.yaml index a4bc1fabf..77b534a74 100644 --- a/examples/tests.yaml +++ b/examples/tests.yaml @@ -87,6 +87,7 @@ anta.tests.bfd: tx_interval: 1200 rx_interval: 1200 multiplier: 3 + detection_time: 3600 - VerifyBFDPeersRegProtocols: # Verifies the registered routing protocol of IPv4 BFD peer sessions. bfd_peers: diff --git a/tests/units/anta_tests/test_bfd.py b/tests/units/anta_tests/test_bfd.py index af1329f94..8b234222f 100644 --- a/tests/units/anta_tests/test_bfd.py +++ b/tests/units/anta_tests/test_bfd.py @@ -27,6 +27,7 @@ "operTxInterval": 1200000, "operRxInterval": 1200000, "detectMult": 3, + "detectTime": 3600000, } } } @@ -42,6 +43,7 @@ "operTxInterval": 1200000, "operRxInterval": 1200000, "detectMult": 3, + "detectTime": 3600000, } } } @@ -59,6 +61,55 @@ }, "expected": {"result": "success"}, }, + { + "name": "success-detection-time", + "test": VerifyBFDPeersIntervals, + "eos_data": [ + { + "vrfs": { + "default": { + "ipv4Neighbors": { + "192.0.255.7": { + "peerStats": { + "": { + "peerStatsDetail": { + "operTxInterval": 1200000, + "operRxInterval": 1200000, + "detectMult": 3, + "detectTime": 3600000, + } + } + } + } + } + }, + "MGMT": { + "ipv4Neighbors": { + "192.0.255.70": { + "peerStats": { + "": { + "peerStatsDetail": { + "operTxInterval": 1200000, + "operRxInterval": 1200000, + "detectMult": 3, + "detectTime": 3600000, + } + } + } + } + } + }, + } + } + ], + "inputs": { + "bfd_peers": [ + {"peer_address": "192.0.255.7", "vrf": "default", "tx_interval": 1200, "rx_interval": 1200, "multiplier": 3, "detection_time": 3600}, + {"peer_address": "192.0.255.70", "vrf": "MGMT", "tx_interval": 1200, "rx_interval": 1200, "multiplier": 3, "detection_time": 3600}, + ] + }, + "expected": {"result": "success"}, + }, { "name": "failure-no-peer", "test": VerifyBFDPeersIntervals, @@ -74,6 +125,7 @@ "operTxInterval": 1200000, "operRxInterval": 1200000, "detectMult": 3, + "detectTime": 3600000, } } } @@ -89,6 +141,7 @@ "operTxInterval": 1200000, "operRxInterval": 1200000, "detectMult": 3, + "detectTime": 3600000, } } } @@ -100,8 +153,8 @@ ], "inputs": { "bfd_peers": [ - {"peer_address": "192.0.255.7", "vrf": "CS", "tx_interval": 1200, "rx_interval": 1200, "multiplier": 3}, - {"peer_address": "192.0.255.70", "vrf": "MGMT", "tx_interval": 1200, "rx_interval": 1200, "multiplier": 3}, + {"peer_address": "192.0.255.7", "vrf": "CS", "tx_interval": 1200, "rx_interval": 1200, "multiplier": 3, "detection_time": 3600}, + {"peer_address": "192.0.255.70", "vrf": "MGMT", "tx_interval": 1200, "rx_interval": 1200, "multiplier": 3, "detection_time": 3600}, ] }, "expected": { @@ -127,6 +180,7 @@ "operTxInterval": 1300000, "operRxInterval": 1200000, "detectMult": 4, + "detectTime": 4000000, } } } @@ -142,6 +196,7 @@ "operTxInterval": 120000, "operRxInterval": 120000, "detectMult": 5, + "detectTime": 4000000, } } } @@ -168,6 +223,66 @@ ], }, }, + { + "name": "failure-incorrect-timers-with-detection-time", + "test": VerifyBFDPeersIntervals, + "eos_data": [ + { + "vrfs": { + "default": { + "ipv4Neighbors": { + "192.0.255.7": { + "peerStats": { + "": { + "peerStatsDetail": { + "operTxInterval": 1300000, + "operRxInterval": 1200000, + "detectMult": 4, + "detectTime": 4000000, + } + } + } + } + } + }, + "MGMT": { + "ipv4Neighbors": { + "192.0.255.70": { + "peerStats": { + "": { + "peerStatsDetail": { + "operTxInterval": 120000, + "operRxInterval": 120000, + "detectMult": 5, + "detectTime": 4000000, + } + } + } + } + } + }, + } + } + ], + "inputs": { + "bfd_peers": [ + {"peer_address": "192.0.255.7", "vrf": "default", "tx_interval": 1200, "rx_interval": 1200, "multiplier": 3, "detection_time": 3600}, + {"peer_address": "192.0.255.70", "vrf": "MGMT", "tx_interval": 1200, "rx_interval": 1200, "multiplier": 3, "detection_time": 3600}, + ] + }, + "expected": { + "result": "failure", + "messages": [ + "Peer: 192.0.255.7 VRF: default - Incorrect Transmit interval - Expected: 1200 Actual: 1300", + "Peer: 192.0.255.7 VRF: default - Incorrect Multiplier - Expected: 3 Actual: 4", + "Peer: 192.0.255.7 VRF: default - Incorrect Detection Time - Expected: 3600 Actual: 4000", + "Peer: 192.0.255.70 VRF: MGMT - Incorrect Transmit interval - Expected: 1200 Actual: 120", + "Peer: 192.0.255.70 VRF: MGMT - Incorrect Receive interval - Expected: 1200 Actual: 120", + "Peer: 192.0.255.70 VRF: MGMT - Incorrect Multiplier - Expected: 3 Actual: 5", + "Peer: 192.0.255.70 VRF: MGMT - Incorrect Detection Time - Expected: 3600 Actual: 4000", + ], + }, + }, { "name": "success", "test": VerifyBFDSpecificPeers, From 68c664df189cce09da62731ffab39618f122dfae Mon Sep 17 00:00:00 2001 From: Guillaume Mulocher Date: Wed, 8 Jan 2025 16:06:04 +0100 Subject: [PATCH 4/8] feat(anta.cli): Better error message when invalid inventory or catalogs are parsed (#1000) * Feat(anta.cli): Better error message when invalid inventory or catalogs are parsed * test: Add tests --- anta/cli/utils.py | 7 +++++-- tests/data/invalid_inventory.yml | 5 +++++ tests/units/cli/nrfu/test__init__.py | 20 ++++++++++++++++++++ 3 files changed, 30 insertions(+), 2 deletions(-) create mode 100644 tests/data/invalid_inventory.yml diff --git a/anta/cli/utils.py b/anta/cli/utils.py index e740f8c56..508424dd0 100644 --- a/anta/cli/utils.py +++ b/anta/cli/utils.py @@ -17,6 +17,7 @@ from anta.catalog import AntaCatalog from anta.inventory import AntaInventory from anta.inventory.exceptions import InventoryIncorrectSchemaError, InventoryRootKeyError +from anta.logger import anta_log_exception if TYPE_CHECKING: from click import Option @@ -242,7 +243,8 @@ def wrapper( insecure=insecure, disable_cache=disable_cache, ) - except (TypeError, ValueError, YAMLError, OSError, InventoryIncorrectSchemaError, InventoryRootKeyError): + except (TypeError, ValueError, YAMLError, OSError, InventoryIncorrectSchemaError, InventoryRootKeyError) as e: + anta_log_exception(e, f"Failed to parse the inventory: {inventory}", logger) ctx.exit(ExitCode.USAGE_ERROR) return f(*args, inventory=i, **kwargs) @@ -319,7 +321,8 @@ def wrapper( try: file_format = catalog_format.lower() c = AntaCatalog.parse(catalog, file_format=file_format) # type: ignore[arg-type] - except (TypeError, ValueError, YAMLError, OSError): + except (TypeError, ValueError, YAMLError, OSError) as e: + anta_log_exception(e, f"Failed to parse the catalog: {catalog}", logger) ctx.exit(ExitCode.USAGE_ERROR) return f(*args, catalog=c, **kwargs) diff --git a/tests/data/invalid_inventory.yml b/tests/data/invalid_inventory.yml new file mode 100644 index 000000000..5199a15fa --- /dev/null +++ b/tests/data/invalid_inventory.yml @@ -0,0 +1,5 @@ +--- +anta_inventory: + - host: 172.20.20.101 + name: DC1-SPINE1 + tags: ["SPINE", "DC1"] diff --git a/tests/units/cli/nrfu/test__init__.py b/tests/units/cli/nrfu/test__init__.py index f81c67e9e..bbc892839 100644 --- a/tests/units/cli/nrfu/test__init__.py +++ b/tests/units/cli/nrfu/test__init__.py @@ -5,14 +5,18 @@ from __future__ import annotations +from pathlib import Path from typing import TYPE_CHECKING from anta.cli import anta from anta.cli.utils import ExitCode if TYPE_CHECKING: + import pytest from click.testing import CliRunner +DATA_DIR: Path = Path(__file__).parents[3].resolve() / "data" + # TODO: write unit tests for ignore-status and ignore-error @@ -123,3 +127,19 @@ def test_hide(click_runner: CliRunner) -> None: """Test the `--hide` option of the `anta nrfu` command.""" result = click_runner.invoke(anta, ["nrfu", "--hide", "success", "text"]) assert "SUCCESS" not in result.output + + +def test_invalid_inventory(caplog: pytest.LogCaptureFixture, click_runner: CliRunner) -> None: + """Test invalid inventory.""" + result = click_runner.invoke(anta, ["nrfu", "--inventory", str(DATA_DIR / "invalid_inventory.yml")]) + assert "CRITICAL" in caplog.text + assert "Failed to parse the inventory" in caplog.text + assert result.exit_code == ExitCode.USAGE_ERROR + + +def test_invalid_catalog(caplog: pytest.LogCaptureFixture, click_runner: CliRunner) -> None: + """Test invalid catalog.""" + result = click_runner.invoke(anta, ["nrfu", "--catalog", str(DATA_DIR / "test_catalog_not_a_list.yml")]) + assert "CRITICAL" in caplog.text + assert "Failed to parse the catalog" in caplog.text + assert result.exit_code == ExitCode.USAGE_ERROR From 44d34f37240c3e9a297557098bfcbf4eb491e09a Mon Sep 17 00:00:00 2001 From: "dependabot[bot]" <49699333+dependabot[bot]@users.noreply.github.com> Date: Fri, 10 Jan 2025 09:39:19 +0100 Subject: [PATCH 5/8] chore: update ruff requirement from <0.9.0,>=0.5.4 to >=0.5.4,<0.10.0 (#1002) * chore: update ruff requirement from <0.9.0,>=0.5.4 to >=0.5.4,<0.10.0 Updates the requirements on [ruff](https://github.com/astral-sh/ruff) to permit the latest version. - [Release notes](https://github.com/astral-sh/ruff/releases) - [Changelog](https://github.com/astral-sh/ruff/blob/main/CHANGELOG.md) - [Commits](https://github.com/astral-sh/ruff/compare/0.5.4...0.9.0) --- updated-dependencies: - dependency-name: ruff dependency-type: direct:production ... Signed-off-by: dependabot[bot] * Update pre-commit as well * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * ci: Ignore shadowing logging module with anta.tests.logging --------- Signed-off-by: dependabot[bot] Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> Co-authored-by: Guillaume Mulocher Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com> --- .pre-commit-config.yaml | 2 +- anta/catalog.py | 8 +++----- anta/runner.py | 5 ++--- anta/tests/system.py | 2 +- pyproject.toml | 5 ++++- tests/benchmark/test_anta.py | 2 +- tests/units/test_runner.py | 2 +- 7 files changed, 13 insertions(+), 13 deletions(-) diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index f12dda006..1961ec0c4 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -46,7 +46,7 @@ repos: - '' - repo: https://github.com/astral-sh/ruff-pre-commit - rev: v0.8.6 + rev: v0.9.0 hooks: - id: ruff name: Run Ruff linter diff --git a/anta/catalog.py b/anta/catalog.py index 65031cedc..ca0eeb176 100644 --- a/anta/catalog.py +++ b/anta/catalog.py @@ -182,7 +182,7 @@ def flatten_modules(data: dict[str, Any], package: str | None = None) -> dict[Mo except Exception as e: # A test module is potentially user-defined code. # We need to catch everything if we want to have meaningful logs - module_str = f"{module_name[1:] if module_name.startswith('.') else module_name}{f' from package {package}' if package else ''}" + module_str = f"{module_name.removeprefix('.')}{f' from package {package}' if package else ''}" message = f"Module named {module_str} cannot be imported. Verify that the module exists and there is no Python syntax issues." anta_log_exception(e, message, logger) raise ValueError(message) from e @@ -223,16 +223,14 @@ def check_tests(cls: type[AntaCatalogFile], data: Any) -> Any: # noqa: ANN401 raise ValueError(msg) # noqa: TRY004 pydantic catches ValueError or AssertionError, no TypeError if len(test_definition) != 1: msg = ( - f"Syntax error when parsing: {test_definition}\n" - "It must be a dictionary with a single entry. Check the indentation in the test catalog." + f"Syntax error when parsing: {test_definition}\nIt must be a dictionary with a single entry. Check the indentation in the test catalog." ) raise ValueError(msg) for test_name, test_inputs in test_definition.copy().items(): test: type[AntaTest] | None = getattr(module, test_name, None) if test is None: msg = ( - f"{test_name} is not defined in Python module {module.__name__}" - f"{f' (from {module.__file__})' if module.__file__ is not None else ''}" + f"{test_name} is not defined in Python module {module.__name__}{f' (from {module.__file__})' if module.__file__ is not None else ''}" ) raise ValueError(msg) test_definitions.append(AntaTestDefinition(test=test, inputs=test_inputs)) diff --git a/anta/runner.py b/anta/runner.py index 93c433722..2b3c8b70e 100644 --- a/anta/runner.py +++ b/anta/runner.py @@ -115,7 +115,7 @@ async def setup_inventory(inventory: AntaInventory, tags: set[str] | None, devic # If there are no devices in the inventory after filtering, exit if not selected_inventory.devices: - msg = f'No reachable device {f"matching the tags {tags} " if tags else ""}was found.{f" Selected devices: {devices} " if devices is not None else ""}' + msg = f"No reachable device {f'matching the tags {tags} ' if tags else ''}was found.{f' Selected devices: {devices} ' if devices is not None else ''}" logger.warning(msg) return None @@ -170,8 +170,7 @@ def prepare_tests( if total_test_count == 0: msg = ( - f"There are no tests{f' matching the tags {tags} ' if tags else ' '}to run in the current " - "test catalog and device inventory, please verify your inputs." + f"There are no tests{f' matching the tags {tags} ' if tags else ' '}to run in the current test catalog and device inventory, please verify your inputs." ) logger.warning(msg) return None diff --git a/anta/tests/system.py b/anta/tests/system.py index 85235f2fc..11cf8398c 100644 --- a/anta/tests/system.py +++ b/anta/tests/system.py @@ -224,7 +224,7 @@ def test(self) -> None: if memory_usage > MEMORY_THRESHOLD: self.result.is_success() else: - self.result.is_failure(f"Device has reported a high memory usage: {(1 - memory_usage)*100:.2f}%") + self.result.is_failure(f"Device has reported a high memory usage: {(1 - memory_usage) * 100:.2f}%") class VerifyFileSystemUtilization(AntaTest): diff --git a/pyproject.toml b/pyproject.toml index 1e85b01f0..09fee8c3e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -75,7 +75,7 @@ dev = [ "pytest-metadata>=3.0.0", "pytest>=7.4.0", "respx>=0.22.0", - "ruff>=0.5.4,<0.9.0", + "ruff>=0.5.4,<0.10.0", "tox>=4.10.0,<5.0.0", "types-PyYAML", "types-pyOpenSSL", @@ -428,6 +428,9 @@ runtime-evaluated-base-classes = ["pydantic.BaseModel", "anta.models.AntaTest.In "C901", # TODO: test function is too complex, needs a refactor "PLR0912" # Too many branches (15/12) (too-many-branches), needs a refactor ] +"anta/tests/logging.py" = [ + "A005", # TODO: Module `logging` shadows a Python standard-library module +] "anta/decorators.py" = [ "ANN401", # Ok to use Any type hint in our decorators ] diff --git a/tests/benchmark/test_anta.py b/tests/benchmark/test_anta.py index 91d3b3ff2..1daf7f369 100644 --- a/tests/benchmark/test_anta.py +++ b/tests/benchmark/test_anta.py @@ -47,7 +47,7 @@ def _() -> None: if len(results.results) != len(inventory) * len(catalog.tests): pytest.fail(f"Expected {len(inventory) * len(catalog.tests)} tests but got {len(results.results)}", pytrace=False) - bench_info = "\n--- ANTA NRFU Dry-Run Benchmark Information ---\n" f"Test count: {len(results.results)}\n" "-----------------------------------------------" + bench_info = f"\n--- ANTA NRFU Dry-Run Benchmark Information ---\nTest count: {len(results.results)}\n-----------------------------------------------" logger.info(bench_info) diff --git a/tests/units/test_runner.py b/tests/units/test_runner.py index b3ac1b56a..1b9c40c88 100644 --- a/tests/units/test_runner.py +++ b/tests/units/test_runner.py @@ -67,7 +67,7 @@ async def test_no_selected_device(caplog: pytest.LogCaptureFixture, inventory: A caplog.set_level(logging.WARNING) manager = ResultManager() await main(manager, inventory, FAKE_CATALOG, tags=tags, devices=devices) - msg = f'No reachable device {f"matching the tags {tags} " if tags else ""}was found.{f" Selected devices: {devices} " if devices is not None else ""}' + msg = f"No reachable device {f'matching the tags {tags} ' if tags else ''}was found.{f' Selected devices: {devices} ' if devices is not None else ''}" assert msg in caplog.messages From 5389bb01d206f42381ec1043e43e12ac6bb883c5 Mon Sep 17 00:00:00 2001 From: vitthalmagadum <122079046+vitthalmagadum@users.noreply.github.com> Date: Tue, 14 Jan 2025 09:58:22 +0530 Subject: [PATCH 6/8] feat(anta): Added the test case to verify syslog logging is enabled (#859) --- anta/tests/logging.py | 29 ++++++++++++++++++ examples/tests.yaml | 2 ++ tests/units/anta_tests/test_logging.py | 42 ++++++++++++++++++++++++++ 3 files changed, 73 insertions(+) diff --git a/anta/tests/logging.py b/anta/tests/logging.py index d32701836..4a1c594e8 100644 --- a/anta/tests/logging.py +++ b/anta/tests/logging.py @@ -43,6 +43,35 @@ def _get_logging_states(logger: logging.Logger, command_output: str) -> str: return log_states +class VerifySyslogLogging(AntaTest): + """Verifies if syslog logging is enabled. + + Expected Results + ---------------- + * Success: The test will pass if syslog logging is enabled. + * Failure: The test will fail if syslog logging is disabled. + + Examples + -------- + ```yaml + anta.tests.logging: + - VerifySyslogLogging: + ``` + """ + + categories: ClassVar[list[str]] = ["logging"] + commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show logging", ofmt="text")] + + @AntaTest.anta_test + def test(self) -> None: + """Main test function for VerifySyslogLogging.""" + self.result.is_success() + log_output = self.instance_commands[0].text_output + + if "Syslog logging: enabled" not in _get_logging_states(self.logger, log_output): + self.result.is_failure("Syslog logging is disabled.") + + class VerifyLoggingPersistent(AntaTest): """Verifies if logging persistent is enabled and logs are saved in flash. diff --git a/examples/tests.yaml b/examples/tests.yaml index 77b534a74..0b573616c 100644 --- a/examples/tests.yaml +++ b/examples/tests.yaml @@ -307,6 +307,8 @@ anta.tests.logging: vrf: default - VerifyLoggingTimestamp: # Verifies if logs are generated with the appropriate timestamp. + - VerifySyslogLogging: + # Verifies if syslog logging is enabled. anta.tests.mlag: - VerifyMlagConfigSanity: # Verifies there are no MLAG config-sanity inconsistencies. diff --git a/tests/units/anta_tests/test_logging.py b/tests/units/anta_tests/test_logging.py index 6aeac4a21..0c1408817 100644 --- a/tests/units/anta_tests/test_logging.py +++ b/tests/units/anta_tests/test_logging.py @@ -16,6 +16,7 @@ VerifyLoggingPersistent, VerifyLoggingSourceIntf, VerifyLoggingTimestamp, + VerifySyslogLogging, ) from tests.units.anta_tests import test @@ -277,4 +278,45 @@ "inputs": None, "expected": {"result": "failure", "messages": ["Device has reported syslog messages with a severity of ERRORS or higher"]}, }, + { + "name": "success", + "test": VerifySyslogLogging, + "eos_data": [ + """Syslog logging: enabled + Buffer logging: level debugging + + External configuration: + active: + inactive: + + Facility Severity Effective Severity + -------------------- ------------- ------------------ + aaa debugging debugging + accounting debugging debugging""", + ], + "inputs": None, + "expected": {"result": "success"}, + }, + { + "name": "failure", + "test": VerifySyslogLogging, + "eos_data": [ + """Syslog logging: disabled + Buffer logging: level debugging + Console logging: level errors + Persistent logging: disabled + Monitor logging: level errors + + External configuration: + active: + inactive: + + Facility Severity Effective Severity + -------------------- ------------- ------------------ + aaa debugging debugging + accounting debugging debugging""", + ], + "inputs": None, + "expected": {"result": "failure", "messages": ["Syslog logging is disabled."]}, + }, ] From 5fda9651edc700df1ab7ef194de4df7e9b808424 Mon Sep 17 00:00:00 2001 From: "pre-commit-ci[bot]" <66853113+pre-commit-ci[bot]@users.noreply.github.com> Date: Tue, 14 Jan 2025 13:58:09 +0100 Subject: [PATCH 7/8] ci: pre-commit autoupdate (#1005) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit updates: - [github.com/astral-sh/ruff-pre-commit: v0.9.0 → v0.9.1](https://github.com/astral-sh/ruff-pre-commit/compare/v0.9.0...v0.9.1) Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com> Co-authored-by: Guillaume Mulocher --- .pre-commit-config.yaml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 1961ec0c4..42b0ce1cb 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -46,7 +46,7 @@ repos: - '' - repo: https://github.com/astral-sh/ruff-pre-commit - rev: v0.9.0 + rev: v0.9.1 hooks: - id: ruff name: Run Ruff linter From 8e5de9a6cceb6e56812518d7b33add341b50b5a8 Mon Sep 17 00:00:00 2001 From: vitthalmagadum <122079046+vitthalmagadum@users.noreply.github.com> Date: Tue, 14 Jan 2025 18:59:16 +0530 Subject: [PATCH 8/8] feat(anta): Added the test case to verify SNMP user (#877) * Added TC for SNMP user * Updated input model refactoring changes * updated documentation apis * added unit tests for the input models * addressed review comments: updated docstrings, input model * updated field validator * Addressed review comments: updated input model docstrings * Remove unnecessary TypeVar --------- Co-authored-by: VitthalMagadum Co-authored-by: Carl Baillargeon --- anta/custom_types.py | 4 +- anta/input_models/snmp.py | 35 ++++++ anta/tests/snmp.py | 79 ++++++++++++ docs/api/tests.snmp.md | 16 +++ examples/tests.yaml | 8 ++ tests/units/anta_tests/test_snmp.py | 165 ++++++++++++++++++++++++++ tests/units/input_models/test_snmp.py | 44 +++++++ 7 files changed, 350 insertions(+), 1 deletion(-) create mode 100644 anta/input_models/snmp.py create mode 100644 tests/units/input_models/test_snmp.py diff --git a/anta/custom_types.py b/anta/custom_types.py index bd6a7b8d2..4763be495 100644 --- a/anta/custom_types.py +++ b/anta/custom_types.py @@ -208,7 +208,6 @@ def validate_regex(value: str) -> str: SnmpErrorCounter = Literal[ "inVersionErrs", "inBadCommunityNames", "inBadCommunityUses", "inParseErrs", "outTooBigErrs", "outNoSuchNameErrs", "outBadValueErrs", "outGeneralErrs" ] - IPv4RouteType = Literal[ "connected", "static", @@ -238,3 +237,6 @@ def validate_regex(value: str) -> str: "Route Cache Route", "CBF Leaked Route", ] +SnmpVersion = Literal["v1", "v2c", "v3"] +SnmpHashingAlgorithm = Literal["MD5", "SHA", "SHA-224", "SHA-256", "SHA-384", "SHA-512"] +SnmpEncryptionAlgorithm = Literal["AES-128", "AES-192", "AES-256", "DES"] diff --git a/anta/input_models/snmp.py b/anta/input_models/snmp.py new file mode 100644 index 000000000..680d617c0 --- /dev/null +++ b/anta/input_models/snmp.py @@ -0,0 +1,35 @@ +# Copyright (c) 2023-2025 Arista Networks, Inc. +# Use of this source code is governed by the Apache License 2.0 +# that can be found in the LICENSE file. +"""Module containing input models for SNMP tests.""" + +from __future__ import annotations + +from pydantic import BaseModel, ConfigDict + +from anta.custom_types import SnmpEncryptionAlgorithm, SnmpHashingAlgorithm, SnmpVersion + + +class SnmpUser(BaseModel): + """Model for a SNMP User.""" + + model_config = ConfigDict(extra="forbid") + username: str + """SNMP user name.""" + group_name: str + """SNMP group for the user.""" + version: SnmpVersion + """SNMP protocol version.""" + auth_type: SnmpHashingAlgorithm | None = None + """User authentication algorithm. Can be provided in the `VerifySnmpUser` test.""" + priv_type: SnmpEncryptionAlgorithm | None = None + """User privacy algorithm. Can be provided in the `VerifySnmpUser` test.""" + + def __str__(self) -> str: + """Return a human-readable string representation of the SnmpUser for reporting. + + Examples + -------- + - User: Test Group: Test_Group Version: v2c + """ + return f"User: {self.username} Group: {self.group_name} Version: {self.version}" diff --git a/anta/tests/snmp.py b/anta/tests/snmp.py index 36b97ba2f..ef7410430 100644 --- a/anta/tests/snmp.py +++ b/anta/tests/snmp.py @@ -9,7 +9,10 @@ from typing import TYPE_CHECKING, ClassVar, get_args +from pydantic import field_validator + from anta.custom_types import PositiveInteger, SnmpErrorCounter, SnmpPdu +from anta.input_models.snmp import SnmpUser from anta.models import AntaCommand, AntaTest from anta.tools import get_value @@ -339,3 +342,79 @@ def test(self) -> None: self.result.is_success() else: self.result.is_failure(f"The following SNMP error counters are not found or have non-zero error counters:\n{error_counters_not_ok}") + + +class VerifySnmpUser(AntaTest): + """Verifies the SNMP user configurations. + + This test performs the following checks for each specified user: + + 1. User exists in SNMP configuration. + 2. Group assignment is correct. + 3. For SNMPv3 users only: + - Authentication type matches (if specified) + - Privacy type matches (if specified) + + Expected Results + ---------------- + * Success: If all of the following conditions are met: + - All users exist with correct group assignments. + - SNMPv3 authentication and privacy types match specified values. + * Failure: If any of the following occur: + - User not found in SNMP configuration. + - Incorrect group assignment. + - For SNMPv3: Mismatched authentication or privacy types. + + Examples + -------- + ```yaml + anta.tests.snmp: + - VerifySnmpUser: + snmp_users: + - username: test + group_name: test_group + version: v3 + auth_type: MD5 + priv_type: AES-128 + ``` + """ + + categories: ClassVar[list[str]] = ["snmp"] + commands: ClassVar[list[AntaCommand | AntaTemplate]] = [AntaCommand(command="show snmp user", revision=1)] + + class Input(AntaTest.Input): + """Input model for the VerifySnmpUser test.""" + + snmp_users: list[SnmpUser] + """List of SNMP users.""" + + @field_validator("snmp_users") + @classmethod + def validate_snmp_users(cls, snmp_users: list[SnmpUser]) -> list[SnmpUser]: + """Validate that 'auth_type' or 'priv_type' field is provided in each SNMPv3 user.""" + for user in snmp_users: + if user.version == "v3" and not (user.auth_type or user.priv_type): + msg = f"{user} 'auth_type' or 'priv_type' field is required with 'version: v3'" + raise ValueError(msg) + return snmp_users + + @AntaTest.anta_test + def test(self) -> None: + """Main test function for VerifySnmpUser.""" + self.result.is_success() + + for user in self.inputs.snmp_users: + # Verify SNMP user details. + if not (user_details := get_value(self.instance_commands[0].json_output, f"usersByVersion.{user.version}.users.{user.username}")): + self.result.is_failure(f"{user} - Not found") + continue + + if user.group_name != (act_group := user_details.get("groupName", "Not Found")): + self.result.is_failure(f"{user} - Incorrect user group - Actual: {act_group}") + + if user.version == "v3": + if user.auth_type and (act_auth_type := get_value(user_details, "v3Params.authType", "Not Found")) != user.auth_type: + self.result.is_failure(f"{user} - Incorrect authentication type - Expected: {user.auth_type} Actual: {act_auth_type}") + + if user.priv_type and (act_encryption := get_value(user_details, "v3Params.privType", "Not Found")) != user.priv_type: + self.result.is_failure(f"{user} - Incorrect privacy type - Expected: {user.priv_type} Actual: {act_encryption}") diff --git a/docs/api/tests.snmp.md b/docs/api/tests.snmp.md index 85c147e2a..eec97202f 100644 --- a/docs/api/tests.snmp.md +++ b/docs/api/tests.snmp.md @@ -7,7 +7,10 @@ anta_title: ANTA catalog for SNMP tests ~ that can be found in the LICENSE file. --> +# Tests + ::: anta.tests.snmp + options: show_root_heading: false show_root_toc_entry: false @@ -18,3 +21,16 @@ anta_title: ANTA catalog for SNMP tests filters: - "!test" - "!render" + +# Input models + +::: anta.input_models.snmp + + options: + show_root_heading: false + show_root_toc_entry: false + show_bases: false + merge_init_into_class: false + anta_hide_test_module_description: true + show_labels: true + filters: ["!^__str__"] diff --git a/examples/tests.yaml b/examples/tests.yaml index 0b573616c..c2e00f009 100644 --- a/examples/tests.yaml +++ b/examples/tests.yaml @@ -746,6 +746,14 @@ anta.tests.snmp: - VerifySnmpStatus: # Verifies if the SNMP agent is enabled. vrf: default + - VerifySnmpUser: + # Verifies the SNMP user configurations. + snmp_users: + - username: test + group_name: test_group + version: v3 + auth_type: MD5 + priv_type: AES-128 anta.tests.software: - VerifyEOSExtensions: # Verifies that all EOS extensions installed on the device are enabled for boot persistence. diff --git a/tests/units/anta_tests/test_snmp.py b/tests/units/anta_tests/test_snmp.py index 195ef298e..d2eb6b87f 100644 --- a/tests/units/anta_tests/test_snmp.py +++ b/tests/units/anta_tests/test_snmp.py @@ -15,6 +15,7 @@ VerifySnmpLocation, VerifySnmpPDUCounters, VerifySnmpStatus, + VerifySnmpUser, ) from tests.units.anta_tests import test @@ -319,4 +320,168 @@ ], }, }, + { + "name": "success", + "test": VerifySnmpUser, + "eos_data": [ + { + "usersByVersion": { + "v1": { + "users": { + "Test1": { + "groupName": "TestGroup1", + }, + } + }, + "v2c": { + "users": { + "Test2": { + "groupName": "TestGroup2", + }, + } + }, + "v3": { + "users": { + "Test3": { + "groupName": "TestGroup3", + "v3Params": {"authType": "SHA-384", "privType": "AES-128"}, + }, + "Test4": {"groupName": "TestGroup3", "v3Params": {"authType": "SHA-512", "privType": "AES-192"}}, + } + }, + } + } + ], + "inputs": { + "snmp_users": [ + {"username": "Test1", "group_name": "TestGroup1", "version": "v1"}, + {"username": "Test2", "group_name": "TestGroup2", "version": "v2c"}, + {"username": "Test3", "group_name": "TestGroup3", "version": "v3", "auth_type": "SHA-384", "priv_type": "AES-128"}, + {"username": "Test4", "group_name": "TestGroup3", "version": "v3", "auth_type": "SHA-512", "priv_type": "AES-192"}, + ] + }, + "expected": {"result": "success"}, + }, + { + "name": "failure-not-configured", + "test": VerifySnmpUser, + "eos_data": [ + { + "usersByVersion": { + "v3": { + "users": { + "Test3": { + "groupName": "TestGroup3", + "v3Params": {"authType": "SHA-384", "privType": "AES-128"}, + }, + } + }, + } + } + ], + "inputs": { + "snmp_users": [ + {"username": "Test1", "group_name": "TestGroup1", "version": "v1"}, + {"username": "Test2", "group_name": "TestGroup2", "version": "v2c"}, + {"username": "Test3", "group_name": "TestGroup3", "version": "v3", "auth_type": "SHA-384", "priv_type": "AES-128"}, + {"username": "Test4", "group_name": "TestGroup3", "version": "v3", "auth_type": "SHA-512", "priv_type": "AES-192"}, + ] + }, + "expected": { + "result": "failure", + "messages": [ + "User: Test1 Group: TestGroup1 Version: v1 - Not found", + "User: Test2 Group: TestGroup2 Version: v2c - Not found", + "User: Test4 Group: TestGroup3 Version: v3 - Not found", + ], + }, + }, + { + "name": "failure-incorrect-group", + "test": VerifySnmpUser, + "eos_data": [ + { + "usersByVersion": { + "v1": { + "users": { + "Test1": { + "groupName": "TestGroup2", + }, + } + }, + "v2c": { + "users": { + "Test2": { + "groupName": "TestGroup1", + }, + } + }, + "v3": {}, + } + } + ], + "inputs": { + "snmp_users": [ + {"username": "Test1", "group_name": "TestGroup1", "version": "v1"}, + {"username": "Test2", "group_name": "TestGroup2", "version": "v2c"}, + ] + }, + "expected": { + "result": "failure", + "messages": [ + "User: Test1 Group: TestGroup1 Version: v1 - Incorrect user group - Actual: TestGroup2", + "User: Test2 Group: TestGroup2 Version: v2c - Incorrect user group - Actual: TestGroup1", + ], + }, + }, + { + "name": "failure-incorrect-auth-encryption", + "test": VerifySnmpUser, + "eos_data": [ + { + "usersByVersion": { + "v1": { + "users": { + "Test1": { + "groupName": "TestGroup1", + }, + } + }, + "v2c": { + "users": { + "Test2": { + "groupName": "TestGroup2", + }, + } + }, + "v3": { + "users": { + "Test3": { + "groupName": "TestGroup3", + "v3Params": {"authType": "SHA-512", "privType": "AES-192"}, + }, + "Test4": {"groupName": "TestGroup4", "v3Params": {"authType": "SHA-384", "privType": "AES-128"}}, + } + }, + } + } + ], + "inputs": { + "snmp_users": [ + {"username": "Test1", "group_name": "TestGroup1", "version": "v1"}, + {"username": "Test2", "group_name": "TestGroup2", "version": "v2c"}, + {"username": "Test3", "group_name": "TestGroup3", "version": "v3", "auth_type": "SHA-384", "priv_type": "AES-128"}, + {"username": "Test4", "group_name": "TestGroup4", "version": "v3", "auth_type": "SHA-512", "priv_type": "AES-192"}, + ] + }, + "expected": { + "result": "failure", + "messages": [ + "User: Test3 Group: TestGroup3 Version: v3 - Incorrect authentication type - Expected: SHA-384 Actual: SHA-512", + "User: Test3 Group: TestGroup3 Version: v3 - Incorrect privacy type - Expected: AES-128 Actual: AES-192", + "User: Test4 Group: TestGroup4 Version: v3 - Incorrect authentication type - Expected: SHA-512 Actual: SHA-384", + "User: Test4 Group: TestGroup4 Version: v3 - Incorrect privacy type - Expected: AES-192 Actual: AES-128", + ], + }, + }, ] diff --git a/tests/units/input_models/test_snmp.py b/tests/units/input_models/test_snmp.py new file mode 100644 index 000000000..94551ca76 --- /dev/null +++ b/tests/units/input_models/test_snmp.py @@ -0,0 +1,44 @@ +# Copyright (c) 2023-2025 Arista Networks, Inc. +# Use of this source code is governed by the Apache License 2.0 +# that can be found in the LICENSE file. +"""Tests for anta.input_models.snmp.py.""" + +# pylint: disable=C0302 +from __future__ import annotations + +from typing import TYPE_CHECKING + +import pytest +from pydantic import ValidationError + +from anta.tests.snmp import VerifySnmpUser + +if TYPE_CHECKING: + from anta.input_models.snmp import SnmpUser + + +class TestVerifySnmpUserInput: + """Test anta.tests.snmp.VerifySnmpUser.Input.""" + + @pytest.mark.parametrize( + ("snmp_users"), + [ + pytest.param([{"username": "test", "group_name": "abc", "version": "v1", "auth_type": None, "priv_type": None}], id="valid-v1"), + pytest.param([{"username": "test", "group_name": "abc", "version": "v2c", "auth_type": None, "priv_type": None}], id="valid-v2c"), + pytest.param([{"username": "test", "group_name": "abc", "version": "v3", "auth_type": "SHA", "priv_type": "AES-128"}], id="valid-v3"), + ], + ) + def test_valid(self, snmp_users: list[SnmpUser]) -> None: + """Test VerifySnmpUser.Input valid inputs.""" + VerifySnmpUser.Input(snmp_users=snmp_users) + + @pytest.mark.parametrize( + ("snmp_users"), + [ + pytest.param([{"username": "test", "group_name": "abc", "version": "v3", "auth_type": None, "priv_type": None}], id="invalid-v3"), + ], + ) + def test_invalid(self, snmp_users: list[SnmpUser]) -> None: + """Test VerifySnmpUser.Input invalid inputs.""" + with pytest.raises(ValidationError): + VerifySnmpUser.Input(snmp_users=snmp_users)