summaryrefslogtreecommitdiffstats
path: root/tests/units/test_device.py
diff options
context:
space:
mode:
Diffstat (limited to 'tests/units/test_device.py')
-rw-r--r--tests/units/test_device.py221
1 files changed, 107 insertions, 114 deletions
diff --git a/tests/units/test_device.py b/tests/units/test_device.py
index 845da2b..c901a3d 100644
--- a/tests/units/test_device.py
+++ b/tests/units/test_device.py
@@ -1,20 +1,17 @@
# 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.
-"""
-test anta.device.py
-"""
+"""test anta.device.py."""
from __future__ import annotations
import asyncio
from pathlib import Path
-from typing import Any
+from typing import TYPE_CHECKING, Any
from unittest.mock import patch
import httpx
import pytest
-from _pytest.mark.structures import ParameterSet
from asyncssh import SSHClientConnection, SSHClientConnectionOptions
from rich import print as rprint
@@ -24,6 +21,9 @@ from anta.models import AntaCommand
from tests.lib.fixture import COMMAND_OUTPUT
from tests.lib.utils import generate_test_ids_list
+if TYPE_CHECKING:
+ from _pytest.mark.structures import ParameterSet
+
INIT_DATA: list[dict[str, Any]] = [
{
"name": "no name, no port",
@@ -155,8 +155,8 @@ AIOEAPI_COLLECT_DATA: list[dict[str, Any]] = [
"memTotal": 8099732,
"memFree": 4989568,
"isIntlVersion": False,
- }
- ]
+ },
+ ],
},
},
"expected": {
@@ -211,7 +211,7 @@ AIOEAPI_COLLECT_DATA: list[dict[str, Any]] = [
"memFree": 4989568,
"isIntlVersion": False,
},
- ]
+ ],
},
},
"expected": {
@@ -266,7 +266,7 @@ AIOEAPI_COLLECT_DATA: list[dict[str, Any]] = [
"memFree": 4989568,
"isIntlVersion": False,
},
- ]
+ ],
},
},
"expected": {
@@ -322,7 +322,7 @@ AIOEAPI_COLLECT_DATA: list[dict[str, Any]] = [
"memFree": 4989568,
"isIntlVersion": False,
},
- ]
+ ],
},
},
"expected": {
@@ -356,8 +356,12 @@ AIOEAPI_COLLECT_DATA: list[dict[str, Any]] = [
"command": "show version",
"patch_kwargs": {
"side_effect": 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=[],
+ ),
},
},
"expected": {"output": None, "errors": ["Authorization denied for command 'show version'"]},
@@ -369,7 +373,7 @@ AIOEAPI_COLLECT_DATA: list[dict[str, Any]] = [
"command": "show version",
"patch_kwargs": {"side_effect": httpx.HTTPError(message="404")},
},
- "expected": {"output": None, "errors": ["404"]},
+ "expected": {"output": None, "errors": ["HTTPError: 404"]},
},
{
"name": "httpx.ConnectError",
@@ -378,7 +382,7 @@ AIOEAPI_COLLECT_DATA: list[dict[str, Any]] = [
"command": "show version",
"patch_kwargs": {"side_effect": httpx.ConnectError(message="Cannot open port")},
},
- "expected": {"output": None, "errors": ["Cannot open port"]},
+ "expected": {"output": None, "errors": ["ConnectError: Cannot open port"]},
},
]
AIOEAPI_COPY_DATA: list[dict[str, Any]] = [
@@ -387,7 +391,7 @@ AIOEAPI_COPY_DATA: list[dict[str, Any]] = [
"device": {},
"copy": {
"sources": [Path("/mnt/flash"), Path("/var/log/agents")],
- "destination": Path("."),
+ "destination": Path(),
"direction": "from",
},
},
@@ -396,7 +400,7 @@ AIOEAPI_COPY_DATA: list[dict[str, Any]] = [
"device": {},
"copy": {
"sources": [Path("/mnt/flash"), Path("/var/log/agents")],
- "destination": Path("."),
+ "destination": Path(),
"direction": "to",
},
},
@@ -405,7 +409,7 @@ AIOEAPI_COPY_DATA: list[dict[str, Any]] = [
"device": {},
"copy": {
"sources": [Path("/mnt/flash"), Path("/var/log/agents")],
- "destination": Path("."),
+ "destination": Path(),
"direction": "wrong",
},
},
@@ -417,26 +421,28 @@ REFRESH_DATA: list[dict[str, Any]] = [
"patch_kwargs": (
{"return_value": True},
{
- "return_value": {
- "mfgName": "Arista",
- "modelName": "DCS-7280CR3-32P4-F",
- "hardwareRevision": "11.00",
- "serialNumber": "JPE19500066",
- "systemMacAddress": "fc:bd:67:3d:13:c5",
- "hwMacAddress": "fc:bd:67:3d:13:c5",
- "configMacAddress": "00:00:00:00:00:00",
- "version": "4.31.1F-34361447.fraserrel (engineering build)",
- "architecture": "x86_64",
- "internalVersion": "4.31.1F-34361447.fraserrel",
- "internalBuildId": "4940d112-a2fc-4970-8b5a-a16cd03fd08c",
- "imageFormatVersion": "3.0",
- "imageOptimization": "Default",
- "bootupTimestamp": 1700729434.5892005,
- "uptime": 20666.78,
- "memTotal": 8099732,
- "memFree": 4989568,
- "isIntlVersion": False,
- }
+ "return_value": [
+ {
+ "mfgName": "Arista",
+ "modelName": "DCS-7280CR3-32P4-F",
+ "hardwareRevision": "11.00",
+ "serialNumber": "JPE19500066",
+ "systemMacAddress": "fc:bd:67:3d:13:c5",
+ "hwMacAddress": "fc:bd:67:3d:13:c5",
+ "configMacAddress": "00:00:00:00:00:00",
+ "version": "4.31.1F-34361447.fraserrel (engineering build)",
+ "architecture": "x86_64",
+ "internalVersion": "4.31.1F-34361447.fraserrel",
+ "internalBuildId": "4940d112-a2fc-4970-8b5a-a16cd03fd08c",
+ "imageFormatVersion": "3.0",
+ "imageOptimization": "Default",
+ "bootupTimestamp": 1700729434.5892005,
+ "uptime": 20666.78,
+ "memTotal": 8099732,
+ "memFree": 4989568,
+ "isIntlVersion": False,
+ }
+ ],
},
),
"expected": {"is_online": True, "established": True, "hw_model": "DCS-7280CR3-32P4-F"},
@@ -466,7 +472,7 @@ REFRESH_DATA: list[dict[str, Any]] = [
"memTotal": 8099732,
"memFree": 4989568,
"isIntlVersion": False,
- }
+ },
},
),
"expected": {"is_online": False, "established": False, "hw_model": None},
@@ -477,25 +483,27 @@ REFRESH_DATA: list[dict[str, Any]] = [
"patch_kwargs": (
{"return_value": True},
{
- "return_value": {
- "mfgName": "Arista",
- "hardwareRevision": "11.00",
- "serialNumber": "JPE19500066",
- "systemMacAddress": "fc:bd:67:3d:13:c5",
- "hwMacAddress": "fc:bd:67:3d:13:c5",
- "configMacAddress": "00:00:00:00:00:00",
- "version": "4.31.1F-34361447.fraserrel (engineering build)",
- "architecture": "x86_64",
- "internalVersion": "4.31.1F-34361447.fraserrel",
- "internalBuildId": "4940d112-a2fc-4970-8b5a-a16cd03fd08c",
- "imageFormatVersion": "3.0",
- "imageOptimization": "Default",
- "bootupTimestamp": 1700729434.5892005,
- "uptime": 20666.78,
- "memTotal": 8099732,
- "memFree": 4989568,
- "isIntlVersion": False,
- }
+ "return_value": [
+ {
+ "mfgName": "Arista",
+ "hardwareRevision": "11.00",
+ "serialNumber": "JPE19500066",
+ "systemMacAddress": "fc:bd:67:3d:13:c5",
+ "hwMacAddress": "fc:bd:67:3d:13:c5",
+ "configMacAddress": "00:00:00:00:00:00",
+ "version": "4.31.1F-34361447.fraserrel (engineering build)",
+ "architecture": "x86_64",
+ "internalVersion": "4.31.1F-34361447.fraserrel",
+ "internalBuildId": "4940d112-a2fc-4970-8b5a-a16cd03fd08c",
+ "imageFormatVersion": "3.0",
+ "imageOptimization": "Default",
+ "bootupTimestamp": 1700729434.5892005,
+ "uptime": 20666.78,
+ "memTotal": 8099732,
+ "memFree": 4989568,
+ "isIntlVersion": False,
+ }
+ ],
},
),
"expected": {"is_online": True, "established": False, "hw_model": None},
@@ -507,8 +515,12 @@ REFRESH_DATA: list[dict[str, Any]] = [
{"return_value": True},
{
"side_effect": 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=[],
+ ),
},
),
"expected": {"is_online": True, "established": False, "hw_model": None},
@@ -599,21 +611,17 @@ CACHE_STATS_DATA: list[ParameterSet] = [
class TestAntaDevice:
- """
- Test for anta.device.AntaDevice Abstract class
- """
+ """Test for anta.device.AntaDevice Abstract class."""
- @pytest.mark.asyncio
+ @pytest.mark.asyncio()
@pytest.mark.parametrize(
- "device, command_data, expected_data",
- map(lambda d: (d["device"], d["command"], d["expected"]), COLLECT_DATA),
+ ("device", "command_data", "expected_data"),
+ ((d["device"], d["command"], d["expected"]) for d in COLLECT_DATA),
indirect=["device"],
ids=generate_test_ids_list(COLLECT_DATA),
)
async def test_collect(self, device: AntaDevice, command_data: dict[str, Any], expected_data: dict[str, Any]) -> None:
- """
- Test AntaDevice.collect behavior
- """
+ """Test AntaDevice.collect behavior."""
command = AntaCommand(command=command_data["command"], use_cache=command_data["use_cache"])
# Dummy output for cache hit
@@ -646,32 +654,21 @@ class TestAntaDevice:
assert device.cache is None
device._collect.assert_called_once_with(command=command) # type: ignore[attr-defined] # pylint: disable=protected-access
- @pytest.mark.parametrize("device, expected", CACHE_STATS_DATA, indirect=["device"])
+ @pytest.mark.parametrize(("device", "expected"), CACHE_STATS_DATA, indirect=["device"])
def test_cache_statistics(self, device: AntaDevice, expected: dict[str, Any] | None) -> None:
- """
- Verify that when cache statistics attribute does not exist
- TODO add a test where cache has some value
- """
- assert device.cache_statistics == expected
+ """Verify that when cache statistics attribute does not exist.
- def test_supports(self, device: AntaDevice) -> None:
+ TODO add a test where cache has some value.
"""
- Test if the supports() method
- """
- command = AntaCommand(command="show hardware counter drop", errors=["Unavailable command (not supported on this hardware platform) (at token 2: 'counter')"])
- assert device.supports(command) is False
- command = AntaCommand(command="show hardware counter drop")
- assert device.supports(command) is True
+ assert device.cache_statistics == expected
class TestAsyncEOSDevice:
- """
- Test for anta.device.AsyncEOSDevice
- """
+ """Test for anta.device.AsyncEOSDevice."""
@pytest.mark.parametrize("data", INIT_DATA, ids=generate_test_ids_list(INIT_DATA))
def test__init__(self, data: dict[str, Any]) -> None:
- """Test the AsyncEOSDevice constructor"""
+ """Test the AsyncEOSDevice constructor."""
device = AsyncEOSDevice(**data["device"])
assert device.name == data["expected"]["name"]
@@ -683,12 +680,12 @@ class TestAsyncEOSDevice:
assert device.cache_locks is not None
hash(device)
- with patch("anta.device.__DEBUG__", True):
+ with patch("anta.device.__DEBUG__", new=True):
rprint(device)
@pytest.mark.parametrize("data", EQUALITY_DATA, ids=generate_test_ids_list(EQUALITY_DATA))
def test__eq(self, data: dict[str, Any]) -> None:
- """Test the AsyncEOSDevice equality"""
+ """Test the AsyncEOSDevice equality."""
device1 = AsyncEOSDevice(**data["device1"])
device2 = AsyncEOSDevice(**data["device2"])
if data["expected"]:
@@ -696,49 +693,45 @@ class TestAsyncEOSDevice:
else:
assert device1 != device2
- @pytest.mark.asyncio
+ @pytest.mark.asyncio()
@pytest.mark.parametrize(
- "async_device, patch_kwargs, expected",
- map(lambda d: (d["device"], d["patch_kwargs"], d["expected"]), REFRESH_DATA),
+ ("async_device", "patch_kwargs", "expected"),
+ ((d["device"], d["patch_kwargs"], d["expected"]) for d in REFRESH_DATA),
ids=generate_test_ids_list(REFRESH_DATA),
indirect=["async_device"],
)
async def test_refresh(self, async_device: AsyncEOSDevice, patch_kwargs: list[dict[str, Any]], expected: dict[str, Any]) -> None:
# pylint: disable=protected-access
- """Test AsyncEOSDevice.refresh()"""
- with patch.object(async_device._session, "check_connection", **patch_kwargs[0]):
- with patch.object(async_device._session, "cli", **patch_kwargs[1]):
- await async_device.refresh()
- async_device._session.check_connection.assert_called_once()
- if expected["is_online"]:
- async_device._session.cli.assert_called_once()
- assert async_device.is_online == expected["is_online"]
- assert async_device.established == expected["established"]
- assert async_device.hw_model == expected["hw_model"]
+ """Test AsyncEOSDevice.refresh()."""
+ with patch.object(async_device._session, "check_connection", **patch_kwargs[0]), patch.object(async_device._session, "cli", **patch_kwargs[1]):
+ await async_device.refresh()
+ async_device._session.check_connection.assert_called_once()
+ if expected["is_online"]:
+ async_device._session.cli.assert_called_once()
+ assert async_device.is_online == expected["is_online"]
+ assert async_device.established == expected["established"]
+ assert async_device.hw_model == expected["hw_model"]
- @pytest.mark.asyncio
+ @pytest.mark.asyncio()
@pytest.mark.parametrize(
- "async_device, command, expected",
- map(lambda d: (d["device"], d["command"], d["expected"]), AIOEAPI_COLLECT_DATA),
+ ("async_device", "command", "expected"),
+ ((d["device"], d["command"], d["expected"]) for d in AIOEAPI_COLLECT_DATA),
ids=generate_test_ids_list(AIOEAPI_COLLECT_DATA),
indirect=["async_device"],
)
async def test__collect(self, async_device: AsyncEOSDevice, command: dict[str, Any], expected: dict[str, Any]) -> None:
# pylint: disable=protected-access
- """Test AsyncEOSDevice._collect()"""
- if "revision" in command:
- cmd = AntaCommand(command=command["command"], revision=command["revision"])
- else:
- cmd = AntaCommand(command=command["command"])
+ """Test AsyncEOSDevice._collect()."""
+ cmd = AntaCommand(command=command["command"], revision=command["revision"]) if "revision" in command else AntaCommand(command=command["command"])
with patch.object(async_device._session, "cli", **command["patch_kwargs"]):
await async_device.collect(cmd)
- commands = []
+ commands: list[dict[str, Any]] = []
if async_device.enable and async_device._enable_password is not None:
commands.append(
{
"cmd": "enable",
"input": str(async_device._enable_password),
- }
+ },
)
elif async_device.enable:
# No password
@@ -751,15 +744,15 @@ class TestAsyncEOSDevice:
assert cmd.output == expected["output"]
assert cmd.errors == expected["errors"]
- @pytest.mark.asyncio
+ @pytest.mark.asyncio()
@pytest.mark.parametrize(
- "async_device, copy",
- map(lambda d: (d["device"], d["copy"]), AIOEAPI_COPY_DATA),
+ ("async_device", "copy"),
+ ((d["device"], d["copy"]) for d in AIOEAPI_COPY_DATA),
ids=generate_test_ids_list(AIOEAPI_COPY_DATA),
indirect=["async_device"],
)
async def test_copy(self, async_device: AsyncEOSDevice, copy: dict[str, Any]) -> None:
- """Test AsyncEOSDevice.copy()"""
+ """Test AsyncEOSDevice.copy()."""
conn = SSHClientConnection(asyncio.get_event_loop(), SSHClientConnectionOptions())
with patch("asyncssh.connect") as connect_mock:
connect_mock.return_value.__aenter__.return_value = conn