From 641d0d615623d4818993e1967fc96af1eefc4605 Mon Sep 17 00:00:00 2001 From: Daniel Baumann Date: Sun, 14 Apr 2024 10:35:51 +0200 Subject: Adding upstream version 0.14.0. Signed-off-by: Daniel Baumann --- tests/lib/fixture.py | 160 ++++++++++++++++++++++++++------------------------- 1 file changed, 81 insertions(+), 79 deletions(-) (limited to 'tests/lib/fixture.py') 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() -- cgit v1.2.3