diff options
author | Daniel Baumann <daniel.baumann@progress-linux.org> | 2024-04-14 08:36:44 +0000 |
---|---|---|
committer | Daniel Baumann <daniel.baumann@progress-linux.org> | 2024-04-14 08:36:50 +0000 |
commit | 7763cc454d686d51bf2e0ccc1f2ccf7d62a0d625 (patch) | |
tree | f36d2006dd01bd01a069956741d831d9d5633377 /tests/lib | |
parent | Adding debian version 0.13.0-1. (diff) | |
download | anta-7763cc454d686d51bf2e0ccc1f2ccf7d62a0d625.tar.xz anta-7763cc454d686d51bf2e0ccc1f2ccf7d62a0d625.zip |
Merging upstream version 0.14.0.
Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>
Diffstat (limited to 'tests/lib')
-rw-r--r-- | tests/lib/__init__.py | 1 | ||||
-rw-r--r-- | tests/lib/anta.py | 14 | ||||
-rw-r--r-- | tests/lib/fixture.py | 160 | ||||
-rw-r--r-- | tests/lib/utils.py | 22 |
4 files changed, 96 insertions, 101 deletions
diff --git a/tests/lib/__init__.py b/tests/lib/__init__.py index e772bee..cd54f3a 100644 --- a/tests/lib/__init__.py +++ b/tests/lib/__init__.py @@ -1,3 +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. +"""Library for ANTA unit tests.""" diff --git a/tests/lib/anta.py b/tests/lib/anta.py index b97d91d..cabb27b 100644 --- a/tests/lib/anta.py +++ b/tests/lib/anta.py @@ -1,20 +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. -""" -generic test funciton used to generate unit tests for each AntaTest -""" +"""generic test function used to generate unit tests for each AntaTest.""" + from __future__ import annotations import asyncio -from typing import Any +from typing import TYPE_CHECKING, Any -from anta.device import AntaDevice +if TYPE_CHECKING: + from anta.device import AntaDevice def test(device: AntaDevice, data: dict[str, Any]) -> None: - """ - Generic test function for AntaTest subclass. + """Generic test function for AntaTest subclass. + See `tests/units/anta_tests/README.md` for more information on how to use it. """ # Instantiate the AntaTest subclass diff --git a/tests/lib/fixture.py b/tests/lib/fixture.py index 68e9e57..43fb60a 100644 --- a/tests/lib/fixture.py +++ b/tests/lib/fixture.py @@ -1,28 +1,32 @@ # 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. -"""Fixture for Anta Testing""" +"""Fixture for Anta Testing.""" + from __future__ import annotations import logging import shutil -from pathlib import Path -from typing import Any, Callable, Iterator +from typing import TYPE_CHECKING, Any, Callable from unittest.mock import patch import pytest from click.testing import CliRunner, Result -from pytest import CaptureFixture from anta import aioeapi from anta.cli.console import console from anta.device import AntaDevice, AsyncEOSDevice from anta.inventory import AntaInventory -from anta.models import AntaCommand from anta.result_manager import ResultManager from anta.result_manager.models import TestResult from tests.lib.utils import default_anta_env +if TYPE_CHECKING: + from collections.abc import Iterator + from pathlib import Path + + from anta.models import AntaCommand + logger = logging.getLogger(__name__) DEVICE_HW_MODEL = "pytest" @@ -38,7 +42,11 @@ MOCK_CLI_JSON: dict[str, aioeapi.EapiCommandError | dict[str, Any]] = { "clear counters": {}, "clear hardware counter drop": {}, "undefined": aioeapi.EapiCommandError( - passed=[], failed="show version", errors=["Authorization denied for command 'show version'"], errmsg="Invalid command", not_exec=[] + passed=[], + failed="show version", + errors=["Authorization denied for command 'show version'"], + errmsg="Invalid command", + not_exec=[], ), } @@ -50,11 +58,9 @@ MOCK_CLI_TEXT: dict[str, aioeapi.EapiCommandError | str] = { } -@pytest.fixture +@pytest.fixture() def device(request: pytest.FixtureRequest) -> Iterator[AntaDevice]: - """ - Returns an AntaDevice instance with mocked abstract method - """ + """Return an AntaDevice instance with mocked abstract method.""" def _collect(command: AntaCommand) -> None: command.output = COMMAND_OUTPUT @@ -64,22 +70,21 @@ def device(request: pytest.FixtureRequest) -> Iterator[AntaDevice]: if hasattr(request, "param"): # Fixture is parametrized indirectly kwargs.update(request.param) - with patch.object(AntaDevice, "__abstractmethods__", set()): - with patch("anta.device.AntaDevice._collect", side_effect=_collect): - # AntaDevice constructor does not have hw_model argument - hw_model = kwargs.pop("hw_model") - dev = AntaDevice(**kwargs) # type: ignore[abstract, arg-type] # pylint: disable=abstract-class-instantiated, unexpected-keyword-arg - dev.hw_model = hw_model - yield dev + with patch.object(AntaDevice, "__abstractmethods__", set()), patch("anta.device.AntaDevice._collect", side_effect=_collect): + # AntaDevice constructor does not have hw_model argument + hw_model = kwargs.pop("hw_model") + dev = AntaDevice(**kwargs) # type: ignore[abstract, arg-type] # pylint: disable=abstract-class-instantiated, unexpected-keyword-arg + dev.hw_model = hw_model + yield dev -@pytest.fixture +@pytest.fixture() def test_inventory() -> AntaInventory: - """ - Return the test_inventory - """ + """Return the test_inventory.""" env = default_anta_env() - assert env["ANTA_INVENTORY"] and env["ANTA_USERNAME"] and env["ANTA_PASSWORD"] is not None + assert env["ANTA_INVENTORY"] + assert env["ANTA_USERNAME"] + assert env["ANTA_PASSWORD"] is not None return AntaInventory.parse( filename=env["ANTA_INVENTORY"], username=env["ANTA_USERNAME"], @@ -88,34 +93,30 @@ def test_inventory() -> AntaInventory: # tests.unit.test_device.py fixture -@pytest.fixture +@pytest.fixture() def async_device(request: pytest.FixtureRequest) -> AsyncEOSDevice: - """ - Returns an AsyncEOSDevice instance - """ - - kwargs = {"name": DEVICE_NAME, "host": "42.42.42.42", "username": "anta", "password": "anta"} + """Return an AsyncEOSDevice instance.""" + kwargs = { + "name": DEVICE_NAME, + "host": "42.42.42.42", + "username": "anta", + "password": "anta", + } if hasattr(request, "param"): # Fixture is parametrized indirectly kwargs.update(request.param) - dev = AsyncEOSDevice(**kwargs) # type: ignore[arg-type] - return dev + return AsyncEOSDevice(**kwargs) # type: ignore[arg-type] # tests.units.result_manager fixtures -@pytest.fixture +@pytest.fixture() def test_result_factory(device: AntaDevice) -> Callable[[int], TestResult]: - """ - Return a anta.result_manager.models.TestResult object - """ - + """Return a anta.result_manager.models.TestResult object.""" # pylint: disable=redefined-outer-name def _create(index: int = 0) -> TestResult: - """ - Actual Factory - """ + """Actual Factory.""" return TestResult( name=device.name, test=f"VerifyTest{index}", @@ -127,50 +128,39 @@ def test_result_factory(device: AntaDevice) -> Callable[[int], TestResult]: return _create -@pytest.fixture +@pytest.fixture() def list_result_factory(test_result_factory: Callable[[int], TestResult]) -> Callable[[int], list[TestResult]]: - """ - Return a list[TestResult] with 'size' TestResult instanciated using the test_result_factory fixture - """ - + """Return a list[TestResult] with 'size' TestResult instantiated using the test_result_factory fixture.""" # pylint: disable=redefined-outer-name def _factory(size: int = 0) -> list[TestResult]: - """ - Factory for list[TestResult] entry of size entries - """ - result: list[TestResult] = [] - for i in range(size): - result.append(test_result_factory(i)) - return result + """Create a factory for list[TestResult] entry of size entries.""" + return [test_result_factory(i) for i in range(size)] return _factory -@pytest.fixture +@pytest.fixture() def result_manager_factory(list_result_factory: Callable[[int], list[TestResult]]) -> Callable[[int], ResultManager]: - """ - Return a ResultManager factory that takes as input a number of tests - """ - + """Return a ResultManager factory that takes as input a number of tests.""" # pylint: disable=redefined-outer-name def _factory(number: int = 0) -> ResultManager: - """ - Factory for list[TestResult] entry of size entries - """ + """Create a factory for list[TestResult] entry of size entries.""" result_manager = ResultManager() - result_manager.add_test_results(list_result_factory(number)) + result_manager.results = list_result_factory(number) return result_manager return _factory # tests.units.cli fixtures -@pytest.fixture +@pytest.fixture() def temp_env(tmp_path: Path) -> dict[str, str | None]: - """Fixture that create a temporary ANTA inventory that can be overriden - and returns the corresponding environment variables""" + """Fixture that create a temporary ANTA inventory. + + The inventory can be overridden and returns the corresponding environment variables. + """ env = default_anta_env() anta_inventory = str(env["ANTA_INVENTORY"]) temp_inventory = tmp_path / "test_inventory.yml" @@ -179,16 +169,19 @@ def temp_env(tmp_path: Path) -> dict[str, str | None]: return env -@pytest.fixture -def click_runner(capsys: CaptureFixture[str]) -> Iterator[CliRunner]: - """ - Convenience fixture to return a click.CliRunner for cli testing - """ +@pytest.fixture() +# Disabling C901 - too complex as we like our runner like this +def click_runner(capsys: pytest.CaptureFixture[str]) -> Iterator[CliRunner]: # noqa: C901 + """Return a click.CliRunner for cli testing.""" class AntaCliRunner(CliRunner): - """Override CliRunner to inject specific variables for ANTA""" + """Override CliRunner to inject specific variables for ANTA.""" - def invoke(self, *args, **kwargs) -> Result: # type: ignore[no-untyped-def] + def invoke( + self, + *args: Any, # noqa: ANN401 + **kwargs: Any, # noqa: ANN401 + ) -> Result: # Inject default env if not provided kwargs["env"] = kwargs["env"] if "env" in kwargs else default_anta_env() # Deterministic terminal width @@ -198,14 +191,18 @@ def click_runner(capsys: CaptureFixture[str]) -> Iterator[CliRunner]: # Way to fix https://github.com/pallets/click/issues/824 with capsys.disabled(): result = super().invoke(*args, **kwargs) - print("--- CLI Output ---") - print(result.output) + # disabling T201 as we want to print here + print("--- CLI Output ---") # noqa: T201 + print(result.output) # noqa: T201 return result def cli( - command: str | None = None, commands: list[dict[str, Any]] | None = None, ofmt: str = "json", version: int | str | None = "latest", **kwargs: Any + command: str | None = None, + commands: list[dict[str, Any]] | None = None, + ofmt: str = "json", + _version: int | str | None = "latest", + **_kwargs: Any, # noqa: ANN401 ) -> dict[str, Any] | list[dict[str, Any]]: - # pylint: disable=unused-argument def get_output(command: str | dict[str, Any]) -> dict[str, Any]: if isinstance(command, dict): command = command["cmd"] @@ -216,7 +213,7 @@ def click_runner(capsys: CaptureFixture[str]) -> Iterator[CliRunner]: mock_cli = MOCK_CLI_TEXT for mock_cmd, output in mock_cli.items(): if command == mock_cmd: - logger.info(f"Mocking command {mock_cmd}") + logger.info("Mocking command %s", mock_cmd) if isinstance(output, aioeapi.EapiCommandError): raise output return output @@ -226,17 +223,22 @@ def click_runner(capsys: CaptureFixture[str]) -> Iterator[CliRunner]: res: dict[str, Any] | list[dict[str, Any]] if command is not None: - logger.debug(f"Mock input {command}") + logger.debug("Mock input %s", command) res = get_output(command) if commands is not None: - logger.debug(f"Mock input {commands}") + logger.debug("Mock input %s", commands) res = list(map(get_output, commands)) - logger.debug(f"Mock output {res}") + logger.debug("Mock output %s", res) return res # Patch aioeapi methods used by AsyncEOSDevice. See tests/units/test_device.py - with patch("aioeapi.device.Device.check_connection", return_value=True), patch("aioeapi.device.Device.cli", side_effect=cli), patch("asyncssh.connect"), patch( - "asyncssh.scp" + with ( + patch("aioeapi.device.Device.check_connection", return_value=True), + patch("aioeapi.device.Device.cli", side_effect=cli), + patch("asyncssh.connect"), + patch( + "asyncssh.scp", + ), ): console._color_system = None # pylint: disable=protected-access yield AntaCliRunner() diff --git a/tests/lib/utils.py b/tests/lib/utils.py index 460e014..1255936 100644 --- a/tests/lib/utils.py +++ b/tests/lib/utils.py @@ -1,9 +1,8 @@ # 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. -""" -tests.lib.utils -""" +"""tests.lib.utils.""" + from __future__ import annotations from pathlib import Path @@ -11,22 +10,17 @@ from typing import Any def generate_test_ids_dict(val: dict[str, Any], key: str = "name") -> str: - """ - generate_test_ids Helper to generate test ID for parametrize - """ + """generate_test_ids Helper to generate test ID for parametrize.""" return val.get(key, "unamed_test") def generate_test_ids_list(val: list[dict[str, Any]], key: str = "name") -> list[str]: - """ - generate_test_ids Helper to generate test ID for parametrize - """ - return [entry[key] if key in entry.keys() else "unamed_test" for entry in val] + """generate_test_ids Helper to generate test ID for parametrize.""" + return [entry.get(key, "unamed_test") for entry in val] def generate_test_ids(data: list[dict[str, Any]]) -> list[str]: - """ - build id for a unit test of an AntaTest subclass + """Build id for a unit test of an AntaTest subclass. { "name": "meaniful test name", @@ -38,9 +32,7 @@ def generate_test_ids(data: list[dict[str, Any]]) -> list[str]: def default_anta_env() -> dict[str, str | None]: - """ - Return a default_anta_environement which can be passed to a cliRunner.invoke method - """ + """Return a default_anta_environement which can be passed to a cliRunner.invoke method.""" return { "ANTA_USERNAME": "anta", "ANTA_PASSWORD": "formica", |