diff options
Diffstat (limited to 'tests/units/test_device.py')
| -rw-r--r-- | tests/units/test_device.py | 221 |
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 |