# 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.""" from __future__ import annotations import asyncio from contextlib import AbstractContextManager from contextlib import nullcontext as does_not_raise from pathlib import Path from typing import TYPE_CHECKING, Any from unittest.mock import patch import pytest from asyncssh import SSHClientConnection, SSHClientConnectionOptions from httpx import ConnectError, HTTPError, TimeoutException from rich import print as rprint from anta.device import AntaDevice, AsyncEOSDevice from anta.models import AntaCommand from asynceapi import EapiCommandError from tests.units.conftest import COMMAND_OUTPUT if TYPE_CHECKING: from _pytest.mark.structures import ParameterSet INIT_PARAMS: list[ParameterSet] = [ pytest.param({"host": "42.42.42.42", "username": "anta", "password": "anta"}, {"name": "42.42.42.42"}, does_not_raise(), id="no name, no port"), pytest.param({"host": "42.42.42.42", "username": "anta", "password": "anta", "port": 666}, {"name": "42.42.42.42:666"}, does_not_raise(), id="no name, port"), pytest.param( {"host": "42.42.42.42", "username": "anta", "password": "anta", "name": "test.anta.ninja", "disable_cache": True}, {"name": "test.anta.ninja"}, does_not_raise(), id="name", ), pytest.param( {"host": "42.42.42.42", "username": "anta", "password": "anta", "name": "test.anta.ninja", "insecure": True}, {"name": "test.anta.ninja"}, does_not_raise(), id="insecure", ), pytest.param( {"host": None, "username": "anta", "password": "anta", "name": "test.anta.ninja"}, None, pytest.raises(ValueError, match="'host' is required to create an AsyncEOSDevice"), id="host is None", ), pytest.param( {"host": "42.42.42.42", "username": None, "password": "anta", "name": "test.anta.ninja"}, None, pytest.raises(ValueError, match="'username' is required to instantiate device 'test.anta.ninja'"), id="username is None", ), pytest.param( {"host": "42.42.42.42", "username": "anta", "password": None, "name": "test.anta.ninja"}, None, pytest.raises(ValueError, match="'password' is required to instantiate device 'test.anta.ninja'"), id="password is None", ), ] EQUALITY_PARAMS: list[ParameterSet] = [ pytest.param({"host": "42.42.42.42", "username": "anta", "password": "anta"}, {"host": "42.42.42.42", "username": "anta", "password": "blah"}, True, id="equal"), pytest.param( {"host": "42.42.42.42", "username": "anta", "password": "anta", "name": "device1"}, {"host": "42.42.42.42", "username": "plop", "password": "anta", "name": "device2"}, True, id="equals-name", ), pytest.param( {"host": "42.42.42.42", "username": "anta", "password": "anta"}, {"host": "42.42.42.42", "username": "anta", "password": "anta", "port": 666}, False, id="not-equal-port", ), pytest.param( {"host": "42.42.42.41", "username": "anta", "password": "anta"}, {"host": "42.42.42.42", "username": "anta", "password": "anta"}, False, id="not-equal-host", ), ] ASYNCEAPI_COLLECT_PARAMS: list[ParameterSet] = [ pytest.param( {}, { "command": "show version", "patch_kwargs": { "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, } ] }, }, { "output": { "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, }, "errors": [], }, id="command", ), pytest.param( {"enable": True}, { "command": "show version", "patch_kwargs": { "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, }, ] }, }, { "output": { "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, }, "errors": [], }, id="enable", ), pytest.param( {"enable": True, "enable_password": "anta"}, { "command": "show version", "patch_kwargs": { "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, }, ] }, }, { "output": { "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, }, "errors": [], }, id="enable password", ), pytest.param( {}, { "command": "show version", "revision": 3, "patch_kwargs": { "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, }, ] }, }, { "output": { "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, }, "errors": [], }, id="revision", ), pytest.param( {}, { "command": "show version", "patch_kwargs": { "side_effect": EapiCommandError( passed=[], failed="show version", errors=["Authorization denied for command 'show version'"], errmsg="Invalid command", not_exec=[], ) }, }, {"output": None, "errors": ["Authorization denied for command 'show version'"]}, id="asynceapi.EapiCommandError - Authorization denied", ), pytest.param( {}, { "command": "show version", "patch_kwargs": { "side_effect": EapiCommandError( passed=[], failed="show version", errors=["not supported on this hardware platform"], errmsg="Invalid command", not_exec=[], ) }, }, {"output": None, "errors": ["not supported on this hardware platform"]}, id="asynceapi.EapiCommandError - not supported", ), pytest.param( {}, { "command": "show version", "patch_kwargs": { "side_effect": EapiCommandError( passed=[], failed="show version", errors=["BGP inactive"], errmsg="Invalid command", not_exec=[], ) }, }, {"output": None, "errors": ["BGP inactive"]}, id="asynceapi.EapiCommandError - known EOS error", ), pytest.param( {}, { "command": "show version", "patch_kwargs": { "side_effect": EapiCommandError( passed=[], failed="show version", errors=["Invalid input (privileged mode required)"], errmsg="Invalid command", not_exec=[], ) }, }, {"output": None, "errors": ["Invalid input (privileged mode required)"]}, id="asynceapi.EapiCommandError - requires privileges", ), pytest.param( {}, {"command": "show version", "patch_kwargs": {"side_effect": HTTPError("404")}}, {"output": None, "errors": ["HTTPError: 404"]}, id="httpx.HTTPError", ), pytest.param( {}, {"command": "show version", "patch_kwargs": {"side_effect": ConnectError("Cannot open port")}}, {"output": None, "errors": ["ConnectError: Cannot open port"]}, id="httpx.ConnectError", ), pytest.param( {}, {"command": "show version", "patch_kwargs": {"side_effect": TimeoutException("Test")}}, {"output": None, "errors": ["TimeoutException: Test"]}, id="httpx.TimeoutException", ), ] ASYNCEAPI_COPY_PARAMS: list[ParameterSet] = [ pytest.param({}, {"sources": [Path("/mnt/flash"), Path("/var/log/agents")], "destination": Path(), "direction": "from"}, id="from"), pytest.param({}, {"sources": [Path("/mnt/flash"), Path("/var/log/agents")], "destination": Path(), "direction": "to"}, id="to"), pytest.param({}, {"sources": [Path("/mnt/flash"), Path("/var/log/agents")], "destination": Path(), "direction": "wrong"}, id="wrong"), ] REFRESH_PARAMS: list[ParameterSet] = [ pytest.param( {}, ( {"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, } ] }, ), {"is_online": True, "established": True, "hw_model": "DCS-7280CR3-32P4-F"}, id="established", ), pytest.param( {}, ( {"return_value": 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, } }, ), {"is_online": False, "established": False, "hw_model": None}, id="is not online", ), pytest.param( {}, ( {"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, } ] }, ), {"is_online": True, "established": False, "hw_model": None}, id="cannot parse command", ), pytest.param( {}, ( {"return_value": True}, { "side_effect": EapiCommandError( passed=[], failed="show version", errors=["Authorization denied for command 'show version'"], errmsg="Invalid command", not_exec=[], ) }, ), {"is_online": True, "established": False, "hw_model": None}, id="asynceapi.EapiCommandError", ), pytest.param( {}, ({"return_value": True}, {"side_effect": HTTPError("404")}), {"is_online": True, "established": False, "hw_model": None}, id="httpx.HTTPError", ), pytest.param( {}, ({"return_value": True}, {"side_effect": ConnectError("Cannot open port")}), {"is_online": True, "established": False, "hw_model": None}, id="httpx.ConnectError", ), pytest.param( {}, ( {"return_value": True}, { "return_value": [ { "mfgName": "Arista", "modelName": "", } ] }, ), {"is_online": True, "established": False, "hw_model": ""}, id="modelName empty string", ), ] COLLECT_PARAMS: list[ParameterSet] = [ pytest.param( {"disable_cache": False}, {"command": "show version", "use_cache": True}, {"cache_hit": False}, id="device cache enabled, command cache enabled, no cache hit", ), pytest.param( {"disable_cache": False}, {"command": "show version", "use_cache": True}, {"cache_hit": True}, id="device cache enabled, command cache enabled, cache hit", ), pytest.param({"disable_cache": True}, {"command": "show version", "use_cache": True}, {}, id="device cache disabled, command cache enabled"), pytest.param( {"disable_cache": False}, {"command": "show version", "use_cache": False}, {"cache_hit": True}, id="device cache enabled, command cache disabled, cache has command", ), pytest.param( {"disable_cache": False}, {"command": "show version", "use_cache": False}, {"cache_hit": False}, id="device cache enabled, command cache disabled, cache does not have data", ), pytest.param({"disable_cache": True}, {"command": "show version", "use_cache": False}, {}, id="device cache disabled, command cache disabled"), ] CACHE_STATS_PARAMS: list[ParameterSet] = [ pytest.param({"disable_cache": False}, {"total_commands_sent": 0, "cache_hits": 0, "cache_hit_ratio": "0.00%"}, id="with_cache"), pytest.param({"disable_cache": True}, None, id="without_cache"), ] class TestAntaDevice: """Test for anta.device.AntaDevice Abstract class.""" @pytest.mark.parametrize(("device", "command", "expected"), COLLECT_PARAMS, indirect=["device"]) async def test_collect(self, device: AntaDevice, command: dict[str, Any], expected: dict[str, Any]) -> None: """Test AntaDevice.collect behavior.""" cmd = AntaCommand(command=command["command"], use_cache=command["use_cache"]) # Dummy output for cache hit cached_output = "cached_value" if device.cache is not None and expected["cache_hit"] is True: await device.cache.set(cmd.uid, cached_output) await device.collect(cmd) if device.cache is not None: # device_cache is enabled current_cached_data = await device.cache.get(cmd.uid) if cmd.use_cache is True: # command is allowed to use cache if expected["cache_hit"] is True: assert cmd.output == cached_output assert current_cached_data == cached_output assert device.cache.hit_miss_ratio["hits"] == 2 else: assert cmd.output == COMMAND_OUTPUT assert current_cached_data == COMMAND_OUTPUT assert device.cache.hit_miss_ratio["hits"] == 1 else: # command is not allowed to use cache device._collect.assert_called_once_with(command=cmd, collection_id=None) # type: ignore[attr-defined] assert cmd.output == COMMAND_OUTPUT if expected["cache_hit"] is True: assert current_cached_data == cached_output else: assert current_cached_data is None else: # device is disabled assert device.cache is None device._collect.assert_called_once_with(command=cmd, collection_id=None) # type: ignore[attr-defined] @pytest.mark.parametrize(("device", "expected"), CACHE_STATS_PARAMS, 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 class TestAsyncEOSDevice: """Test for anta.device.AsyncEOSDevice.""" @pytest.mark.parametrize(("device", "expected", "expected_raise"), INIT_PARAMS) def test__init__(self, device: dict[str, Any], expected: dict[str, Any] | None, expected_raise: AbstractContextManager[Exception]) -> None: """Test the AsyncEOSDevice constructor.""" with expected_raise: dev = AsyncEOSDevice(**device) assert expected is not None assert dev.name == expected["name"] if device.get("disable_cache") is True: assert dev.cache is None assert dev.cache_locks is None else: # False or None assert dev.cache is not None assert dev.cache_locks is not None hash(dev) with patch("anta.device.__DEBUG__", new=True): rprint(dev) @pytest.mark.parametrize(("device1", "device2", "expected"), EQUALITY_PARAMS) def test__eq(self, device1: dict[str, Any], device2: dict[str, Any], expected: bool) -> None: """Test the AsyncEOSDevice equality.""" dev1 = AsyncEOSDevice(**device1) dev2 = AsyncEOSDevice(**device2) if expected: assert dev1 == dev2 else: assert dev1 != dev2 @pytest.mark.parametrize( ("async_device", "patch_kwargs", "expected"), REFRESH_PARAMS, indirect=["async_device"], ) async def test_refresh(self, async_device: AsyncEOSDevice, patch_kwargs: list[dict[str, Any]], expected: dict[str, Any]) -> None: """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() # type: ignore[attr-defined] # asynceapi.Device.check_connection is patched if expected["is_online"]: async_device._session.cli.assert_called_once() # type: ignore[attr-defined] # asynceapi.Device.cli is patched assert async_device.is_online == expected["is_online"] assert async_device.established == expected["established"] assert async_device.hw_model == expected["hw_model"] @pytest.mark.parametrize( ("async_device", "command", "expected"), ASYNCEAPI_COLLECT_PARAMS, indirect=["async_device"], ) async def test__collect(self, async_device: AsyncEOSDevice, command: dict[str, Any], expected: dict[str, Any]) -> None: """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"]): collection_id = "pytest" await async_device.collect(cmd, collection_id=collection_id) 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 commands.append({"cmd": "enable"}) if cmd.revision: commands.append({"cmd": cmd.command, "revision": cmd.revision}) else: commands.append({"cmd": cmd.command}) async_device._session.cli.assert_called_once_with(commands=commands, ofmt=cmd.ofmt, version=cmd.version, req_id=f"ANTA-{collection_id}-{id(cmd)}") # type: ignore[attr-defined] # asynceapi.Device.cli is patched assert cmd.output == expected["output"] assert cmd.errors == expected["errors"] @pytest.mark.parametrize( ("async_device", "copy"), ASYNCEAPI_COPY_PARAMS, indirect=["async_device"], ) async def test_copy(self, async_device: AsyncEOSDevice, copy: dict[str, Any]) -> None: """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 with patch("asyncssh.scp") as scp_mock: await async_device.copy(copy["sources"], copy["destination"], copy["direction"]) if copy["direction"] == "from": src = [(conn, file) for file in copy["sources"]] dst = copy["destination"] elif copy["direction"] == "to": src = copy["sources"] dst = conn, copy["destination"] else: scp_mock.assert_not_awaited() return scp_mock.assert_awaited_once_with(src, dst)