From 1fd6a618b60d7168fd8f37585d5d39d22d775afd Mon Sep 17 00:00:00 2001 From: Daniel Baumann Date: Thu, 28 Mar 2024 07:11:39 +0100 Subject: Adding upstream version 0.13.0. Signed-off-by: Daniel Baumann --- tests/lib/fixture.py | 242 +++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 242 insertions(+) create mode 100644 tests/lib/fixture.py (limited to 'tests/lib/fixture.py') diff --git a/tests/lib/fixture.py b/tests/lib/fixture.py new file mode 100644 index 0000000..68e9e57 --- /dev/null +++ b/tests/lib/fixture.py @@ -0,0 +1,242 @@ +# 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""" +from __future__ import annotations + +import logging +import shutil +from pathlib import Path +from typing import Any, Callable, Iterator +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 + +logger = logging.getLogger(__name__) + +DEVICE_HW_MODEL = "pytest" +DEVICE_NAME = "pytest" +COMMAND_OUTPUT = "retrieved" + +MOCK_CLI_JSON: dict[str, aioeapi.EapiCommandError | dict[str, Any]] = { + "show version": { + "modelName": "DCS-7280CR3-32P4-F", + "version": "4.31.1F", + }, + "enable": {}, + "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=[] + ), +} + +MOCK_CLI_TEXT: dict[str, aioeapi.EapiCommandError | str] = { + "show version": "Arista cEOSLab", + "bash timeout 10 ls -1t /mnt/flash/schedule/tech-support": "dummy_tech-support_2023-12-01.1115.log.gz\ndummy_tech-support_2023-12-01.1015.log.gz", + "bash timeout 10 ls -1t /mnt/flash/schedule/tech-support | head -1": "dummy_tech-support_2023-12-01.1115.log.gz", + "show running-config | include aaa authorization exec default": "aaa authorization exec default local", +} + + +@pytest.fixture +def device(request: pytest.FixtureRequest) -> Iterator[AntaDevice]: + """ + Returns an AntaDevice instance with mocked abstract method + """ + + def _collect(command: AntaCommand) -> None: + command.output = COMMAND_OUTPUT + + kwargs = {"name": DEVICE_NAME, "hw_model": DEVICE_HW_MODEL} + + 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 + + +@pytest.fixture +def test_inventory() -> AntaInventory: + """ + Return the test_inventory + """ + env = default_anta_env() + assert env["ANTA_INVENTORY"] and env["ANTA_USERNAME"] and env["ANTA_PASSWORD"] is not None + return AntaInventory.parse( + filename=env["ANTA_INVENTORY"], + username=env["ANTA_USERNAME"], + password=env["ANTA_PASSWORD"], + ) + + +# tests.unit.test_device.py 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"} + + if hasattr(request, "param"): + # Fixture is parametrized indirectly + kwargs.update(request.param) + dev = AsyncEOSDevice(**kwargs) # type: ignore[arg-type] + return dev + + +# tests.units.result_manager fixtures +@pytest.fixture +def test_result_factory(device: AntaDevice) -> Callable[[int], TestResult]: + """ + Return a anta.result_manager.models.TestResult object + """ + + # pylint: disable=redefined-outer-name + + def _create(index: int = 0) -> TestResult: + """ + Actual Factory + """ + return TestResult( + name=device.name, + test=f"VerifyTest{index}", + categories=["test"], + description=f"Verifies Test {index}", + custom_field=None, + ) + + return _create + + +@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 + """ + + # 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 + + return _factory + + +@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 + """ + + # pylint: disable=redefined-outer-name + + def _factory(number: int = 0) -> ResultManager: + """ + Factory for list[TestResult] entry of size entries + """ + result_manager = ResultManager() + result_manager.add_test_results(list_result_factory(number)) + return result_manager + + return _factory + + +# tests.units.cli fixtures +@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""" + env = default_anta_env() + anta_inventory = str(env["ANTA_INVENTORY"]) + temp_inventory = tmp_path / "test_inventory.yml" + shutil.copy(anta_inventory, temp_inventory) + env["ANTA_INVENTORY"] = str(temp_inventory) + return env + + +@pytest.fixture +def click_runner(capsys: CaptureFixture[str]) -> Iterator[CliRunner]: + """ + Convenience fixture to return a click.CliRunner for cli testing + """ + + class AntaCliRunner(CliRunner): + """Override CliRunner to inject specific variables for ANTA""" + + def invoke(self, *args, **kwargs) -> Result: # type: ignore[no-untyped-def] + # Inject default env if not provided + kwargs["env"] = kwargs["env"] if "env" in kwargs else default_anta_env() + # Deterministic terminal width + kwargs["env"]["COLUMNS"] = "165" + + kwargs["auto_envvar_prefix"] = "ANTA" + # Way to fix https://github.com/pallets/click/issues/824 + with capsys.disabled(): + result = super().invoke(*args, **kwargs) + print("--- CLI Output ---") + print(result.output) + 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 + ) -> 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"] + mock_cli: dict[str, Any] + if ofmt == "json": + mock_cli = MOCK_CLI_JSON + elif ofmt == "text": + mock_cli = MOCK_CLI_TEXT + for mock_cmd, output in mock_cli.items(): + if command == mock_cmd: + logger.info(f"Mocking command {mock_cmd}") + if isinstance(output, aioeapi.EapiCommandError): + raise output + return output + message = f"Command '{command}' is not mocked" + logger.critical(message) + raise NotImplementedError(message) + + res: dict[str, Any] | list[dict[str, Any]] + if command is not None: + logger.debug(f"Mock input {command}") + res = get_output(command) + if commands is not None: + logger.debug(f"Mock input {commands}") + res = list(map(get_output, commands)) + logger.debug(f"Mock output {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" + ): + console._color_system = None # pylint: disable=protected-access + yield AntaCliRunner() -- cgit v1.2.3