From 7ff8043ce39d44d4793758e4dc093bec5d444b57 Mon Sep 17 00:00:00 2001 From: Guillaume Mulocher Date: Fri, 30 Aug 2024 16:32:43 +0200 Subject: [PATCH 1/4] refactor(anta): Change TestStatus to be an Enum for coding clarity (#758) Co-authored-by: Carl Baillargeon ' - repo: https://github.com/astral-sh/ruff-pre-commit - rev: v0.6.2 + rev: v0.6.3 hooks: - id: ruff name: Run Ruff linter diff --git a/anta/cli/nrfu/__init__.py b/anta/cli/nrfu/__init__.py index 6263e845a..d573b49c7 100644 --- a/anta/cli/nrfu/__init__.py +++ b/anta/cli/nrfu/__init__.py @@ -5,14 +5,14 @@ from __future__ import annotations -from typing import TYPE_CHECKING, get_args +from typing import TYPE_CHECKING import click from anta.cli.nrfu import commands from anta.cli.utils import AliasedGroup, catalog_options, inventory_options -from anta.custom_types import TestStatus from anta.result_manager import ResultManager +from anta.result_manager.models import AntaTestStatus if TYPE_CHECKING: from anta.catalog import AntaCatalog @@ -49,7 +49,7 @@ def parse_args(self, ctx: click.Context, args: list[str]) -> list[str]: return super().parse_args(ctx, args) -HIDE_STATUS: list[str] = list(get_args(TestStatus)) +HIDE_STATUS: list[str] = list(AntaTestStatus) HIDE_STATUS.remove("unset") diff --git a/anta/custom_types.py b/anta/custom_types.py index 153fd7011..322fa4aca 100644 --- a/anta/custom_types.py +++ b/anta/custom_types.py @@ -112,9 +112,6 @@ def validate_regex(value: str) -> str: return value -# ANTA framework -TestStatus = Literal["unset", "success", "failure", "error", "skipped"] - # AntaTest.Input types AAAAuthMethod = Annotated[str, AfterValidator(aaa_group_prefix)] Vlan = Annotated[int, Field(ge=0, le=4094)] diff --git a/anta/reporter/__init__.py b/anta/reporter/__init__.py index c4e4f7bcf..e74aaec5f 100644 --- a/anta/reporter/__init__.py +++ b/anta/reporter/__init__.py @@ -18,9 +18,8 @@ if TYPE_CHECKING: import pathlib - from anta.custom_types import TestStatus from anta.result_manager import ResultManager - from anta.result_manager.models import TestResult + from anta.result_manager.models import AntaTestStatus, TestResult logger = logging.getLogger(__name__) @@ -80,19 +79,19 @@ def _build_headers(self, headers: list[str], table: Table) -> Table: table.add_column(header, justify="left") return table - def _color_result(self, status: TestStatus) -> str: - """Return a colored string based on the status value. + def _color_result(self, status: AntaTestStatus) -> str: + """Return a colored string based on an AntaTestStatus. Parameters ---------- - status (TestStatus): status value to color. + status: AntaTestStatus enum to color. Returns ------- - str: the colored string + The colored string. """ - color = RICH_COLOR_THEME.get(status, "") + color = RICH_COLOR_THEME.get(str(status), "") return f"[{color}]{status}" if color != "" else str(status) def report_all(self, manager: ResultManager, title: str = "All tests results") -> Table: diff --git a/anta/reporter/md_reporter.py b/anta/reporter/md_reporter.py index 0cc5b03e2..7b97fb176 100644 --- a/anta/reporter/md_reporter.py +++ b/anta/reporter/md_reporter.py @@ -12,6 +12,7 @@ from anta.constants import MD_REPORT_TOC from anta.logger import anta_log_exception +from anta.result_manager.models import AntaTestStatus if TYPE_CHECKING: from collections.abc import Generator @@ -203,10 +204,10 @@ def generate_rows(self) -> Generator[str, None, None]: """Generate the rows of the summary totals table.""" yield ( f"| {self.results.get_total_results()} " - f"| {self.results.get_total_results({'success'})} " - f"| {self.results.get_total_results({'skipped'})} " - f"| {self.results.get_total_results({'failure'})} " - f"| {self.results.get_total_results({'error'})} |\n" + f"| {self.results.get_total_results({AntaTestStatus.SUCCESS})} " + f"| {self.results.get_total_results({AntaTestStatus.SKIPPED})} " + f"| {self.results.get_total_results({AntaTestStatus.FAILURE})} " + f"| {self.results.get_total_results({AntaTestStatus.ERROR})} |\n" ) def generate_section(self) -> None: diff --git a/anta/result_manager/__init__.py b/anta/result_manager/__init__.py index 1900a28b1..95da45684 100644 --- a/anta/result_manager/__init__.py +++ b/anta/result_manager/__init__.py @@ -9,13 +9,9 @@ from collections import defaultdict from functools import cached_property from itertools import chain -from typing import get_args - -from pydantic import TypeAdapter from anta.constants import ACRONYM_CATEGORIES -from anta.custom_types import TestStatus -from anta.result_manager.models import TestResult +from anta.result_manager.models import AntaTestStatus, TestResult from .models import CategoryStats, DeviceStats, TestStats @@ -95,7 +91,7 @@ def __init__(self) -> None: error_status is set to True. """ self._result_entries: list[TestResult] = [] - self.status: TestStatus = "unset" + self.status: AntaTestStatus = AntaTestStatus.UNSET self.error_status = False self.device_stats: defaultdict[str, DeviceStats] = defaultdict(DeviceStats) @@ -116,7 +112,7 @@ def results(self, value: list[TestResult]) -> None: """Set the list of TestResult.""" # When setting the results, we need to reset the state of the current instance self._result_entries = [] - self.status = "unset" + self.status = AntaTestStatus.UNSET self.error_status = False # Also reset the stats attributes @@ -138,26 +134,24 @@ def sorted_category_stats(self) -> dict[str, CategoryStats]: return dict(sorted(self.category_stats.items())) @cached_property - def results_by_status(self) -> dict[TestStatus, list[TestResult]]: + def results_by_status(self) -> dict[AntaTestStatus, list[TestResult]]: """A cached property that returns the results grouped by status.""" - return {status: [result for result in self._result_entries if result.result == status] for status in get_args(TestStatus)} + return {status: [result for result in self._result_entries if result.result == status] for status in AntaTestStatus} - def _update_status(self, test_status: TestStatus) -> None: + def _update_status(self, test_status: AntaTestStatus) -> None: """Update the status of the ResultManager instance based on the test status. Parameters ---------- - test_status: TestStatus to update the ResultManager status. + test_status: AntaTestStatus to update the ResultManager status. """ - result_validator: TypeAdapter[TestStatus] = TypeAdapter(TestStatus) - result_validator.validate_python(test_status) if test_status == "error": self.error_status = True return if self.status == "unset" or self.status == "skipped" and test_status in {"success", "failure"}: self.status = test_status elif self.status == "success" and test_status == "failure": - self.status = "failure" + self.status = AntaTestStatus.FAILURE def _update_stats(self, result: TestResult) -> None: """Update the statistics based on the test result. @@ -209,14 +203,14 @@ def add(self, result: TestResult) -> None: # Every time a new result is added, we need to clear the cached property self.__dict__.pop("results_by_status", None) - def get_results(self, status: set[TestStatus] | None = None, sort_by: list[str] | None = None) -> list[TestResult]: + def get_results(self, status: set[AntaTestStatus] | None = None, sort_by: list[str] | None = None) -> list[TestResult]: """Get the results, optionally filtered by status and sorted by TestResult fields. If no status is provided, all results are returned. Parameters ---------- - status: Optional set of TestStatus literals to filter the results. + status: Optional set of AntaTestStatus enum members to filter the results. sort_by: Optional list of TestResult fields to sort the results. Returns @@ -235,14 +229,14 @@ def get_results(self, status: set[TestStatus] | None = None, sort_by: list[str] return results - def get_total_results(self, status: set[TestStatus] | None = None) -> int: + def get_total_results(self, status: set[AntaTestStatus] | None = None) -> int: """Get the total number of results, optionally filtered by status. If no status is provided, the total number of results is returned. Parameters ---------- - status: Optional set of TestStatus literals to filter the results. + status: Optional set of AntaTestStatus enum members to filter the results. Returns ------- @@ -259,18 +253,18 @@ def get_status(self, *, ignore_error: bool = False) -> str: """Return the current status including error_status if ignore_error is False.""" return "error" if self.error_status and not ignore_error else self.status - def filter(self, hide: set[TestStatus]) -> ResultManager: + def filter(self, hide: set[AntaTestStatus]) -> ResultManager: """Get a filtered ResultManager based on test status. Parameters ---------- - hide: set of TestStatus literals to select tests to hide based on their status. + hide: Set of AntaTestStatus enum members to select tests to hide based on their status. Returns ------- A filtered `ResultManager`. """ - possible_statuses = set(get_args(TestStatus)) + possible_statuses = set(AntaTestStatus) manager = ResultManager() manager.results = self.get_results(possible_statuses - hide) return manager diff --git a/anta/result_manager/models.py b/anta/result_manager/models.py index 6abce0233..2bb2aed2e 100644 --- a/anta/result_manager/models.py +++ b/anta/result_manager/models.py @@ -6,10 +6,26 @@ from __future__ import annotations from dataclasses import dataclass, field +from enum import Enum from pydantic import BaseModel -from anta.custom_types import TestStatus + +class AntaTestStatus(str, Enum): + """Test status Enum for the TestResult. + + NOTE: This could be updated to StrEnum when Python 3.11 is the minimum supported version in ANTA. + """ + + UNSET = "unset" + SUCCESS = "success" + FAILURE = "failure" + ERROR = "error" + SKIPPED = "skipped" + + def __str__(self) -> str: + """Override the __str__ method to return the value of the Enum, mimicking the behavior of StrEnum.""" + return self.value class TestResult(BaseModel): @@ -17,13 +33,13 @@ class TestResult(BaseModel): Attributes ---------- - name: Device name where the test has run. - test: Test name runs on the device. - categories: List of categories the TestResult belongs to, by default the AntaTest categories. - description: TestResult description, by default the AntaTest description. - result: Result of the test. Can be one of "unset", "success", "failure", "error" or "skipped". - messages: Message to report after the test if any. - custom_field: Custom field to store a string for flexibility in integrating with ANTA + name: Name of the device where the test was run. + test: Name of the test run on the device. + categories: List of categories the TestResult belongs to. Defaults to the AntaTest categories. + description: Description of the TestResult. Defaults to the AntaTest description. + result: Result of the test. Must be one of the AntaTestStatus Enum values: unset, success, failure, error or skipped. + messages: Messages to report after the test, if any. + custom_field: Custom field to store a string for flexibility in integrating with ANTA. """ @@ -31,7 +47,7 @@ class TestResult(BaseModel): test: str categories: list[str] description: str - result: TestStatus = "unset" + result: AntaTestStatus = AntaTestStatus.UNSET messages: list[str] = [] custom_field: str | None = None @@ -43,7 +59,7 @@ def is_success(self, message: str | None = None) -> None: message: Optional message related to the test """ - self._set_status("success", message) + self._set_status(AntaTestStatus.SUCCESS, message) def is_failure(self, message: str | None = None) -> None: """Set status to failure. @@ -53,7 +69,7 @@ def is_failure(self, message: str | None = None) -> None: message: Optional message related to the test """ - self._set_status("failure", message) + self._set_status(AntaTestStatus.FAILURE, message) def is_skipped(self, message: str | None = None) -> None: """Set status to skipped. @@ -63,7 +79,7 @@ def is_skipped(self, message: str | None = None) -> None: message: Optional message related to the test """ - self._set_status("skipped", message) + self._set_status(AntaTestStatus.SKIPPED, message) def is_error(self, message: str | None = None) -> None: """Set status to error. @@ -73,9 +89,9 @@ def is_error(self, message: str | None = None) -> None: message: Optional message related to the test """ - self._set_status("error", message) + self._set_status(AntaTestStatus.ERROR, message) - def _set_status(self, status: TestStatus, message: str | None = None) -> None: + def _set_status(self, status: AntaTestStatus, message: str | None = None) -> None: """Set status and insert optional message. Parameters diff --git a/asynceapi/aio_portcheck.py b/asynceapi/aio_portcheck.py index fd8e7aee2..79f4562fa 100644 --- a/asynceapi/aio_portcheck.py +++ b/asynceapi/aio_portcheck.py @@ -33,7 +33,7 @@ # ----------------------------------------------------------------------------- -async def port_check_url(url: URL, timeout: int = 5) -> bool: # noqa: ASYNC109 +async def port_check_url(url: URL, timeout: int = 5) -> bool: """ Open the port designated by the URL given the timeout in seconds. diff --git a/tests/units/reporter/test__init__.py b/tests/units/reporter/test__init__.py index 2fc62ce92..f0e44b41a 100644 --- a/tests/units/reporter/test__init__.py +++ b/tests/units/reporter/test__init__.py @@ -13,9 +13,9 @@ from anta import RICH_COLOR_PALETTE from anta.reporter import ReportJinja, ReportTable +from anta.result_manager.models import AntaTestStatus if TYPE_CHECKING: - from anta.custom_types import TestStatus from anta.result_manager import ResultManager @@ -73,15 +73,14 @@ def test__build_headers(self, headers: list[str]) -> None: @pytest.mark.parametrize( ("status", "expected_status"), [ - pytest.param("unknown", "unknown", id="unknown status"), - pytest.param("unset", "[grey74]unset", id="unset status"), - pytest.param("skipped", "[bold orange4]skipped", id="skipped status"), - pytest.param("failure", "[bold red]failure", id="failure status"), - pytest.param("error", "[indian_red]error", id="error status"), - pytest.param("success", "[green4]success", id="success status"), + pytest.param(AntaTestStatus.UNSET, "[grey74]unset", id="unset status"), + pytest.param(AntaTestStatus.SKIPPED, "[bold orange4]skipped", id="skipped status"), + pytest.param(AntaTestStatus.FAILURE, "[bold red]failure", id="failure status"), + pytest.param(AntaTestStatus.ERROR, "[indian_red]error", id="error status"), + pytest.param(AntaTestStatus.SUCCESS, "[green4]success", id="success status"), ], ) - def test__color_result(self, status: TestStatus, expected_status: str) -> None: + def test__color_result(self, status: AntaTestStatus, expected_status: str) -> None: """Test _build_headers.""" # pylint: disable=protected-access report = ReportTable() @@ -140,7 +139,7 @@ def test_report_summary_tests( new_results = [result.model_copy() for result in manager.results] for result in new_results: result.name = "test_device" - result.result = "failure" + result.result = AntaTestStatus.FAILURE report = ReportTable() kwargs = {"tests": [test] if test is not None else None, "title": title} @@ -175,7 +174,7 @@ def test_report_summary_devices( new_results = [result.model_copy() for result in manager.results] for result in new_results: result.name = dev or "test_device" - result.result = "failure" + result.result = AntaTestStatus.FAILURE manager.results = new_results report = ReportTable() diff --git a/tests/units/result_manager/test__init__.py b/tests/units/result_manager/test__init__.py index 66a6cfb1d..802d4a4e3 100644 --- a/tests/units/result_manager/test__init__.py +++ b/tests/units/result_manager/test__init__.py @@ -13,9 +13,9 @@ import pytest from anta.result_manager import ResultManager, models +from anta.result_manager.models import AntaTestStatus if TYPE_CHECKING: - from anta.custom_types import TestStatus from anta.result_manager.models import TestResult @@ -56,7 +56,7 @@ def test_json(self, list_result_factory: Callable[[int], list[TestResult]]) -> N success_list = list_result_factory(3) for test in success_list: - test.result = "success" + test.result = AntaTestStatus.SUCCESS result_manager.results = success_list json_res = result_manager.json @@ -141,29 +141,27 @@ def test_sorted_category_stats(self, list_result_factory: Callable[[int], list[T nullcontext(), id="failure, add success", ), - pytest.param( - "unset", "unknown", None, pytest.raises(ValueError, match="Input should be 'unset', 'success', 'failure', 'error' or 'skipped'"), id="wrong status" - ), + pytest.param("unset", "unknown", None, pytest.raises(ValueError, match="'unknown' is not a valid AntaTestStatus"), id="wrong status"), ], ) def test_add( self, test_result_factory: Callable[[], TestResult], - starting_status: TestStatus, - test_status: TestStatus, + starting_status: str, + test_status: str, expected_status: str, expected_raise: AbstractContextManager[Exception], ) -> None: # pylint: disable=too-many-arguments """Test ResultManager_update_status.""" result_manager = ResultManager() - result_manager.status = starting_status + result_manager.status = AntaTestStatus(starting_status) assert result_manager.error_status is False assert len(result_manager) == 0 test = test_result_factory() - test.result = test_status with expected_raise: + test.result = AntaTestStatus(test_status) result_manager.add(test) if test_status == "error": assert result_manager.error_status is True @@ -199,12 +197,12 @@ def test_add_clear_cache(self, result_manager: ResultManager, test_result_factor def test_get_results(self, result_manager: ResultManager) -> None: """Test ResultManager.get_results.""" # Check for single status - success_results = result_manager.get_results(status={"success"}) + success_results = result_manager.get_results(status={AntaTestStatus.SUCCESS}) assert len(success_results) == 7 assert all(r.result == "success" for r in success_results) # Check for multiple statuses - failure_results = result_manager.get_results(status={"failure", "error"}) + failure_results = result_manager.get_results(status={AntaTestStatus.FAILURE, AntaTestStatus.ERROR}) assert len(failure_results) == 21 assert all(r.result in {"failure", "error"} for r in failure_results) @@ -226,7 +224,7 @@ def test_get_results_sort_by(self, result_manager: ResultManager) -> None: assert all_results[-1].name == "DC1-SPINE1" # Check multiple statuses with sort_by categories - success_skipped_results = result_manager.get_results(status={"success", "skipped"}, sort_by=["categories"]) + success_skipped_results = result_manager.get_results(status={AntaTestStatus.SUCCESS, AntaTestStatus.SKIPPED}, sort_by=["categories"]) assert len(success_skipped_results) == 9 assert success_skipped_results[0].categories == ["Interfaces"] assert success_skipped_results[-1].categories == ["VXLAN"] @@ -246,15 +244,15 @@ def test_get_total_results(self, result_manager: ResultManager) -> None: assert result_manager.get_total_results() == 30 # Test single status - assert result_manager.get_total_results(status={"success"}) == 7 - assert result_manager.get_total_results(status={"failure"}) == 19 - assert result_manager.get_total_results(status={"error"}) == 2 - assert result_manager.get_total_results(status={"skipped"}) == 2 + assert result_manager.get_total_results(status={AntaTestStatus.SUCCESS}) == 7 + assert result_manager.get_total_results(status={AntaTestStatus.FAILURE}) == 19 + assert result_manager.get_total_results(status={AntaTestStatus.ERROR}) == 2 + assert result_manager.get_total_results(status={AntaTestStatus.SKIPPED}) == 2 # Test multiple statuses - assert result_manager.get_total_results(status={"success", "failure"}) == 26 - assert result_manager.get_total_results(status={"success", "failure", "error"}) == 28 - assert result_manager.get_total_results(status={"success", "failure", "error", "skipped"}) == 30 + assert result_manager.get_total_results(status={AntaTestStatus.SUCCESS, AntaTestStatus.FAILURE}) == 26 + assert result_manager.get_total_results(status={AntaTestStatus.SUCCESS, AntaTestStatus.FAILURE, AntaTestStatus.ERROR}) == 28 + assert result_manager.get_total_results(status={AntaTestStatus.SUCCESS, AntaTestStatus.FAILURE, AntaTestStatus.ERROR, AntaTestStatus.SKIPPED}) == 30 @pytest.mark.parametrize( ("status", "error_status", "ignore_error", "expected_status"), @@ -266,7 +264,7 @@ def test_get_total_results(self, result_manager: ResultManager) -> None: ) def test_get_status( self, - status: TestStatus, + status: AntaTestStatus, error_status: bool, ignore_error: bool, expected_status: str, @@ -284,28 +282,28 @@ def test_filter(self, test_result_factory: Callable[[], TestResult], list_result success_list = list_result_factory(3) for test in success_list: - test.result = "success" + test.result = AntaTestStatus.SUCCESS result_manager.results = success_list test = test_result_factory() - test.result = "failure" + test.result = AntaTestStatus.FAILURE result_manager.add(test) test = test_result_factory() - test.result = "error" + test.result = AntaTestStatus.ERROR result_manager.add(test) test = test_result_factory() - test.result = "skipped" + test.result = AntaTestStatus.SKIPPED result_manager.add(test) assert len(result_manager) == 6 - assert len(result_manager.filter({"failure"})) == 5 - assert len(result_manager.filter({"error"})) == 5 - assert len(result_manager.filter({"skipped"})) == 5 - assert len(result_manager.filter({"failure", "error"})) == 4 - assert len(result_manager.filter({"failure", "error", "skipped"})) == 3 - assert len(result_manager.filter({"success", "failure", "error", "skipped"})) == 0 + assert len(result_manager.filter({AntaTestStatus.FAILURE})) == 5 + assert len(result_manager.filter({AntaTestStatus.ERROR})) == 5 + assert len(result_manager.filter({AntaTestStatus.SKIPPED})) == 5 + assert len(result_manager.filter({AntaTestStatus.FAILURE, AntaTestStatus.ERROR})) == 4 + assert len(result_manager.filter({AntaTestStatus.FAILURE, AntaTestStatus.ERROR, AntaTestStatus.SKIPPED})) == 3 + assert len(result_manager.filter({AntaTestStatus.SUCCESS, AntaTestStatus.FAILURE, AntaTestStatus.ERROR, AntaTestStatus.SKIPPED})) == 0 def test_get_by_tests(self, test_result_factory: Callable[[], TestResult], result_manager_factory: Callable[[int], ResultManager]) -> None: """Test ResultManager.get_by_tests.""" diff --git a/tests/units/result_manager/test_models.py b/tests/units/result_manager/test_models.py index 2276153f8..bc44ccfd8 100644 --- a/tests/units/result_manager/test_models.py +++ b/tests/units/result_manager/test_models.py @@ -9,6 +9,8 @@ import pytest +from anta.result_manager.models import AntaTestStatus + # Import as Result to avoid pytest collection from tests.data.json_data import TEST_RESULT_SET_STATUS from tests.lib.fixture import DEVICE_NAME @@ -45,7 +47,7 @@ def test__is_status_foo(self, test_result_factory: Callable[[int], Result], data assert data["message"] in testresult.messages # no helper for unset, testing _set_status if data["target"] == "unset": - testresult._set_status("unset", data["message"]) # pylint: disable=W0212 + testresult._set_status(AntaTestStatus.UNSET, data["message"]) # pylint: disable=W0212 assert testresult.result == data["target"] assert data["message"] in testresult.messages From 145b7c479b952ae02907972bda11821946072b1d Mon Sep 17 00:00:00 2001 From: Carl Baillargeon Date: Fri, 30 Aug 2024 11:00:13 -0400 Subject: [PATCH 2/4] fix(anta): Added support for dict commands in EapiCommandError (#803) --- asynceapi/device.py | 3 +- pyproject.toml | 4 +- tests/units/asynceapi/__init__.py | 4 ++ tests/units/asynceapi/conftest.py | 20 +++++++ tests/units/asynceapi/test_data.py | 88 ++++++++++++++++++++++++++++ tests/units/asynceapi/test_device.py | 88 ++++++++++++++++++++++++++++ 6 files changed, 205 insertions(+), 2 deletions(-) create mode 100644 tests/units/asynceapi/__init__.py create mode 100644 tests/units/asynceapi/conftest.py create mode 100644 tests/units/asynceapi/test_data.py create mode 100644 tests/units/asynceapi/test_device.py diff --git a/asynceapi/device.py b/asynceapi/device.py index ca206d3e4..394abe40d 100644 --- a/asynceapi/device.py +++ b/asynceapi/device.py @@ -271,10 +271,11 @@ async def jsonrpc_exec(self, jsonrpc: dict[str, Any]) -> list[dict[str, Any] | s len_data = len(cmd_data) err_at = len_data - 1 err_msg = err_data["message"] + failed_cmd = commands[err_at] raise EapiCommandError( passed=[get_output(cmd_data[cmd_i]) for cmd_i, cmd in enumerate(commands[:err_at])], - failed=commands[err_at]["cmd"], + failed=failed_cmd["cmd"] if isinstance(failed_cmd, dict) else failed_cmd, errors=cmd_data[err_at]["errors"], errmsg=err_msg, not_exec=commands[err_at + 1 :], diff --git a/pyproject.toml b/pyproject.toml index e64ee80df..85973774b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -69,6 +69,7 @@ dev = [ "pytest-cov>=4.1.0", "pytest-dependency", "pytest-html>=3.2.0", + "pytest-httpx>=0.30.0", "pytest-metadata>=3.0.0", "pytest>=7.4.0", "ruff>=0.5.4,<0.7.0", @@ -181,7 +182,8 @@ filterwarnings = [ [tool.coverage.run] branch = true -source = ["anta"] +# https://community.sonarsource.com/t/python-coverage-analysis-warning/62629/7 +include = ["anta/*", "asynceapi/*"] parallel = true relative_files = true diff --git a/tests/units/asynceapi/__init__.py b/tests/units/asynceapi/__init__.py new file mode 100644 index 000000000..d4282a31b --- /dev/null +++ b/tests/units/asynceapi/__init__.py @@ -0,0 +1,4 @@ +# 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. +"""Unit tests for the asynceapi client package used by ANTA.""" diff --git a/tests/units/asynceapi/conftest.py b/tests/units/asynceapi/conftest.py new file mode 100644 index 000000000..812d5b9cd --- /dev/null +++ b/tests/units/asynceapi/conftest.py @@ -0,0 +1,20 @@ +# 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. +"""Fixtures for the asynceapi client package.""" + +import pytest + +from asynceapi import Device + + +@pytest.fixture +def asynceapi_device() -> Device: + """Return an asynceapi Device instance.""" + return Device( + host="localhost", + username="admin", + password="admin", + proto="https", + port=443, + ) diff --git a/tests/units/asynceapi/test_data.py b/tests/units/asynceapi/test_data.py new file mode 100644 index 000000000..908d6084b --- /dev/null +++ b/tests/units/asynceapi/test_data.py @@ -0,0 +1,88 @@ +# 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. +"""Unit tests data for the asynceapi client package.""" + +SUCCESS_EAPI_RESPONSE = { + "jsonrpc": "2.0", + "id": "EapiExplorer-1", + "result": [ + { + "mfgName": "Arista", + "modelName": "cEOSLab", + "hardwareRevision": "", + "serialNumber": "5E9D49D20F09DA471333DD835835FD1A", + "systemMacAddress": "00:1c:73:2e:7b:a3", + "hwMacAddress": "00:00:00:00:00:00", + "configMacAddress": "00:00:00:00:00:00", + "version": "4.31.1F-34554157.4311F (engineering build)", + "architecture": "i686", + "internalVersion": "4.31.1F-34554157.4311F", + "internalBuildId": "47114ca4-ae9f-4f32-8c1f-2864db93b7e8", + "imageFormatVersion": "1.0", + "imageOptimization": "None", + "cEosToolsVersion": "(unknown)", + "kernelVersion": "6.5.0-44-generic", + "bootupTimestamp": 1723429239.9352903, + "uptime": 1300202.749528885, + "memTotal": 65832112, + "memFree": 41610316, + "isIntlVersion": False, + }, + { + "utcTime": 1724729442.6863558, + "timezone": "EST", + "localTime": { + "year": 2024, + "month": 8, + "dayOfMonth": 26, + "hour": 22, + "min": 30, + "sec": 42, + "dayOfWeek": 0, + "dayOfYear": 239, + "daylightSavingsAdjust": 0, + }, + "clockSource": {"local": True}, + }, + ], +} +"""Successful eAPI JSON response.""" + +ERROR_EAPI_RESPONSE = { + "jsonrpc": "2.0", + "id": "EapiExplorer-1", + "error": { + "code": 1002, + "message": "CLI command 2 of 3 'bad command' failed: invalid command", + "data": [ + { + "mfgName": "Arista", + "modelName": "cEOSLab", + "hardwareRevision": "", + "serialNumber": "5E9D49D20F09DA471333DD835835FD1A", + "systemMacAddress": "00:1c:73:2e:7b:a3", + "hwMacAddress": "00:00:00:00:00:00", + "configMacAddress": "00:00:00:00:00:00", + "version": "4.31.1F-34554157.4311F (engineering build)", + "architecture": "i686", + "internalVersion": "4.31.1F-34554157.4311F", + "internalBuildId": "47114ca4-ae9f-4f32-8c1f-2864db93b7e8", + "imageFormatVersion": "1.0", + "imageOptimization": "None", + "cEosToolsVersion": "(unknown)", + "kernelVersion": "6.5.0-44-generic", + "bootupTimestamp": 1723429239.9352903, + "uptime": 1300027.2297976017, + "memTotal": 65832112, + "memFree": 41595080, + "isIntlVersion": False, + }, + {"errors": ["Invalid input (at token 1: 'bad')"]}, + ], + }, +} +"""Error eAPI JSON response.""" + +JSONRPC_REQUEST_TEMPLATE = {"jsonrpc": "2.0", "method": "runCmds", "params": {"version": 1, "cmds": [], "format": "json"}, "id": "EapiExplorer-1"} +"""Template for JSON-RPC eAPI request. `cmds` must be filled by the parametrize decorator.""" diff --git a/tests/units/asynceapi/test_device.py b/tests/units/asynceapi/test_device.py new file mode 100644 index 000000000..8a140ee3b --- /dev/null +++ b/tests/units/asynceapi/test_device.py @@ -0,0 +1,88 @@ +# 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. +"""Unit tests the asynceapi.device module.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any + +import pytest +from httpx import HTTPStatusError + +from asynceapi import Device, EapiCommandError + +from .test_data import ERROR_EAPI_RESPONSE, JSONRPC_REQUEST_TEMPLATE, SUCCESS_EAPI_RESPONSE + +if TYPE_CHECKING: + from pytest_httpx import HTTPXMock + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "cmds", + [ + (["show version", "show clock"]), + ([{"cmd": "show version"}, {"cmd": "show clock"}]), + ([{"cmd": "show version"}, "show clock"]), + ], + ids=["simple_commands", "complex_commands", "mixed_commands"], +) +async def test_jsonrpc_exec_success( + asynceapi_device: Device, + httpx_mock: HTTPXMock, + cmds: list[str | dict[str, Any]], +) -> None: + """Test the Device.jsonrpc_exec method with a successful response. Simple and complex commands are tested.""" + jsonrpc_request: dict[str, Any] = JSONRPC_REQUEST_TEMPLATE.copy() + jsonrpc_request["params"]["cmds"] = cmds + + httpx_mock.add_response(json=SUCCESS_EAPI_RESPONSE) + + result = await asynceapi_device.jsonrpc_exec(jsonrpc=jsonrpc_request) + + assert result == SUCCESS_EAPI_RESPONSE["result"] + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "cmds", + [ + (["show version", "bad command", "show clock"]), + ([{"cmd": "show version"}, {"cmd": "bad command"}, {"cmd": "show clock"}]), + ([{"cmd": "show version"}, {"cmd": "bad command"}, "show clock"]), + ], + ids=["simple_commands", "complex_commands", "mixed_commands"], +) +async def test_jsonrpc_exec_eapi_command_error( + asynceapi_device: Device, + httpx_mock: HTTPXMock, + cmds: list[str | dict[str, Any]], +) -> None: + """Test the Device.jsonrpc_exec method with an error response. Simple and complex commands are tested.""" + jsonrpc_request: dict[str, Any] = JSONRPC_REQUEST_TEMPLATE.copy() + jsonrpc_request["params"]["cmds"] = cmds + + error_eapi_response: dict[str, Any] = ERROR_EAPI_RESPONSE.copy() + httpx_mock.add_response(json=error_eapi_response) + + with pytest.raises(EapiCommandError) as exc_info: + await asynceapi_device.jsonrpc_exec(jsonrpc=jsonrpc_request) + + assert exc_info.value.passed == [error_eapi_response["error"]["data"][0]] + assert exc_info.value.failed == "bad command" + assert exc_info.value.errors == ["Invalid input (at token 1: 'bad')"] + assert exc_info.value.errmsg == "CLI command 2 of 3 'bad command' failed: invalid command" + assert exc_info.value.not_exec == [jsonrpc_request["params"]["cmds"][2]] + + +@pytest.mark.asyncio +async def test_jsonrpc_exec_http_status_error(asynceapi_device: Device, httpx_mock: HTTPXMock) -> None: + """Test the Device.jsonrpc_exec method with an HTTPStatusError.""" + jsonrpc_request: dict[str, Any] = JSONRPC_REQUEST_TEMPLATE.copy() + jsonrpc_request["params"]["cmds"] = ["show version"] + + httpx_mock.add_response(status_code=500, text="Internal Server Error") + + with pytest.raises(HTTPStatusError): + await asynceapi_device.jsonrpc_exec(jsonrpc=jsonrpc_request) From 722b3e1828c472d0a62ff7acdd1ebd82ad00a6e6 Mon Sep 17 00:00:00 2001 From: Thomas Grimonet Date: Fri, 30 Aug 2024 17:06:36 +0200 Subject: [PATCH 3/4] bump(anta): Upgrade asyncssh to 2.16.0 to suppress deprecation warning from Cryptography (#777) * bump: Force cryptography to lower than 43.0.0 * fix: Suppress deprecation warning from Cryptography in asyncssh * Bump: asyncssh to 2.16.0 --------- Co-authored-by: Guillaume Mulocher --- pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index 85973774b..7202d4839 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -20,7 +20,7 @@ description = "Arista Network Test Automation (ANTA) Framework" license = { file = "LICENSE" } dependencies = [ "aiocache>=0.12.2", - "asyncssh>=2.13.2", + "asyncssh>=2.16", "cvprac>=1.3.1", "eval-type-backport>=0.1.3", # Support newer typing features in older Python versions (required until Python 3.9 support is removed) "Jinja2>=3.1.2", From aa1fde8360d0a9786e42eb35a48c9d006abb4a6e Mon Sep 17 00:00:00 2001 From: Guillaume Mulocher Date: Fri, 30 Aug 2024 17:09:16 +0200 Subject: [PATCH 4/4] feat(anta.cli): Remove --tags from debug commands (#727) --- anta/cli/debug/commands.py | 7 ++++-- anta/cli/debug/utils.py | 5 ++--- anta/cli/utils.py | 46 +++++++++++++++++++++++++++----------- docs/cli/debug.md | 9 +++----- 4 files changed, 43 insertions(+), 24 deletions(-) diff --git a/anta/cli/debug/commands.py b/anta/cli/debug/commands.py index 14f168ba4..1304758a4 100644 --- a/anta/cli/debug/commands.py +++ b/anta/cli/debug/commands.py @@ -72,13 +72,16 @@ def run_template( revision: int, ) -> None: # pylint: disable=too-many-arguments + # Using \b for click + # ruff: noqa: D301 """Run arbitrary templated command to an ANTA device. Takes a list of arguments (keys followed by a value) to build a dictionary used as template parameters. - Example: + \b + Example ------- - anta debug run-template -d leaf1a -t 'show vlan {vlan_id}' vlan_id 1 + anta debug run-template -d leaf1a -t 'show vlan {vlan_id}' vlan_id 1 """ template_params = dict(zip(params[::2], params[1::2])) diff --git a/anta/cli/debug/utils.py b/anta/cli/debug/utils.py index 04a7a38b1..4e20c5a74 100644 --- a/anta/cli/debug/utils.py +++ b/anta/cli/debug/utils.py @@ -11,7 +11,7 @@ import click -from anta.cli.utils import ExitCode, inventory_options +from anta.cli.utils import ExitCode, core_options if TYPE_CHECKING: from anta.inventory import AntaInventory @@ -22,7 +22,7 @@ def debug_options(f: Callable[..., Any]) -> Callable[..., Any]: """Click common options required to execute a command on a specific device.""" - @inventory_options + @core_options @click.option( "--ofmt", type=click.Choice(["json", "text"]), @@ -44,7 +44,6 @@ def wrapper( ctx: click.Context, *args: tuple[Any], inventory: AntaInventory, - tags: set[str] | None, device: str, **kwargs: Any, ) -> Any: diff --git a/anta/cli/utils.py b/anta/cli/utils.py index 6d31e55ae..2f6e7d302 100644 --- a/anta/cli/utils.py +++ b/anta/cli/utils.py @@ -112,7 +112,7 @@ def resolve_command(self, ctx: click.Context, args: Any) -> Any: return cmd.name, cmd, args -def inventory_options(f: Callable[..., Any]) -> Callable[..., Any]: +def core_options(f: Callable[..., Any]) -> Callable[..., Any]: """Click common options when requiring an inventory to interact with devices.""" @click.option( @@ -190,22 +190,12 @@ def inventory_options(f: Callable[..., Any]) -> Callable[..., Any]: required=True, type=click.Path(file_okay=True, dir_okay=False, exists=True, readable=True, path_type=Path), ) - @click.option( - "--tags", - help="List of tags using comma as separator: tag1,tag2,tag3.", - show_envvar=True, - envvar="ANTA_TAGS", - type=str, - required=False, - callback=parse_tags, - ) @click.pass_context @functools.wraps(f) def wrapper( ctx: click.Context, *args: tuple[Any], inventory: Path, - tags: set[str] | None, username: str, password: str | None, enable_password: str | None, @@ -219,7 +209,7 @@ def wrapper( # pylint: disable=too-many-arguments # If help is invoke somewhere, do not parse inventory if ctx.obj.get("_anta_help"): - return f(*args, inventory=None, tags=tags, **kwargs) + return f(*args, inventory=None, **kwargs) if prompt: # User asked for a password prompt if password is None: @@ -255,7 +245,37 @@ def wrapper( ) except (TypeError, ValueError, YAMLError, OSError, InventoryIncorrectSchemaError, InventoryRootKeyError): ctx.exit(ExitCode.USAGE_ERROR) - return f(*args, inventory=i, tags=tags, **kwargs) + return f(*args, inventory=i, **kwargs) + + return wrapper + + +def inventory_options(f: Callable[..., Any]) -> Callable[..., Any]: + """Click common options when requiring an inventory to interact with devices.""" + + @core_options + @click.option( + "--tags", + help="List of tags using comma as separator: tag1,tag2,tag3.", + show_envvar=True, + envvar="ANTA_TAGS", + type=str, + required=False, + callback=parse_tags, + ) + @click.pass_context + @functools.wraps(f) + def wrapper( + ctx: click.Context, + *args: tuple[Any], + tags: set[str] | None, + **kwargs: dict[str, Any], + ) -> Any: + # pylint: disable=too-many-arguments + # If help is invoke somewhere, do not parse inventory + if ctx.obj.get("_anta_help"): + return f(*args, tags=tags, **kwargs) + return f(*args, tags=tags, **kwargs) return wrapper diff --git a/docs/cli/debug.md b/docs/cli/debug.md index db5f4961d..376dffb14 100644 --- a/docs/cli/debug.md +++ b/docs/cli/debug.md @@ -52,8 +52,6 @@ Options: ANTA_DISABLE_CACHE] -i, --inventory FILE Path to the inventory YAML file. [env var: ANTA_INVENTORY; required] - --tags TEXT List of tags using comma as separator: - tag1,tag2,tag3. [env var: ANTA_TAGS] --ofmt [json|text] EOS eAPI format to use. can be text or json -v, --version [1|latest] EOS eAPI version -r, --revision INTEGER eAPI command revision @@ -97,8 +95,9 @@ Usage: anta debug run-template [OPTIONS] PARAMS... Takes a list of arguments (keys followed by a value) to build a dictionary used as template parameters. - Example: ------- anta debug run-template -d leaf1a -t 'show vlan {vlan_id}' - vlan_id 1 + Example + ------- + anta debug run-template -d leaf1a -t 'show vlan {vlan_id}' vlan_id 1 Options: -u, --username TEXT Username to connect to EOS [env var: @@ -125,8 +124,6 @@ Options: ANTA_DISABLE_CACHE] -i, --inventory FILE Path to the inventory YAML file. [env var: ANTA_INVENTORY; required] - --tags TEXT List of tags using comma as separator: - tag1,tag2,tag3. [env var: ANTA_TAGS] --ofmt [json|text] EOS eAPI format to use. can be text or json -v, --version [1|latest] EOS eAPI version -r, --revision INTEGER eAPI command revision