summaryrefslogtreecommitdiffstats
path: root/tests/lib
diff options
context:
space:
mode:
Diffstat (limited to 'tests/lib')
-rw-r--r--tests/lib/__init__.py1
-rw-r--r--tests/lib/anta.py14
-rw-r--r--tests/lib/fixture.py160
-rw-r--r--tests/lib/utils.py22
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",
generated by cgit v1.2.3 (git 2.39.1) at 2024-07-03 14:21:43 +0000