summaryrefslogtreecommitdiffstats
path: root/tests/lib
diff options
context:
space:
mode:
Diffstat (limited to 'tests/lib')
-rw-r--r--tests/lib/__init__.py3
-rw-r--r--tests/lib/anta.py34
-rw-r--r--tests/lib/fixture.py242
-rw-r--r--tests/lib/utils.py49
4 files changed, 328 insertions, 0 deletions
diff --git a/tests/lib/__init__.py b/tests/lib/__init__.py
new file mode 100644
index 0000000..e772bee
--- /dev/null
+++ b/tests/lib/__init__.py
@@ -0,0 +1,3 @@
+# 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.
diff --git a/tests/lib/anta.py b/tests/lib/anta.py
new file mode 100644
index 0000000..b97d91d
--- /dev/null
+++ b/tests/lib/anta.py
@@ -0,0 +1,34 @@
+# 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
+"""
+from __future__ import annotations
+
+import asyncio
+from typing import Any
+
+from anta.device import AntaDevice
+
+
+def test(device: AntaDevice, data: dict[str, Any]) -> None:
+ """
+ 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
+ test_instance = data["test"](device, inputs=data["inputs"], eos_data=data["eos_data"])
+ # Run the test() method
+ asyncio.run(test_instance.test())
+ # Assert expected result
+ assert test_instance.result.result == data["expected"]["result"], test_instance.result.messages
+ if "messages" in data["expected"]:
+ # We expect messages in test result
+ assert len(test_instance.result.messages) == len(data["expected"]["messages"])
+ # Test will pass if the expected message is included in the test result message
+ for message, expected in zip(test_instance.result.messages, data["expected"]["messages"]): # NOTE: zip(strict=True) has been added in Python 3.10
+ assert expected in message
+ else:
+ # Test result should not have messages
+ assert test_instance.result.messages == []
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()
diff --git a/tests/lib/utils.py b/tests/lib/utils.py
new file mode 100644
index 0000000..460e014
--- /dev/null
+++ b/tests/lib/utils.py
@@ -0,0 +1,49 @@
+# 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
+"""
+from __future__ import annotations
+
+from pathlib import Path
+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
+ """
+ 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]
+
+
+def generate_test_ids(data: list[dict[str, Any]]) -> list[str]:
+ """
+ build id for a unit test of an AntaTest subclass
+
+ {
+ "name": "meaniful test name",
+ "test": <AntaTest instance>,
+ ...
+ }
+ """
+ return [f"{val['test'].__module__}.{val['test'].__name__}-{val['name']}" for val in data]
+
+
+def default_anta_env() -> dict[str, str | None]:
+ """
+ Return a default_anta_environement which can be passed to a cliRunner.invoke method
+ """
+ return {
+ "ANTA_USERNAME": "anta",
+ "ANTA_PASSWORD": "formica",
+ "ANTA_INVENTORY": str(Path(__file__).parent.parent / "data" / "test_inventory.yml"),
+ "ANTA_CATALOG": str(Path(__file__).parent.parent / "data" / "test_catalog.yml"),
+ }