diff options
author | Daniel Baumann <daniel.baumann@progress-linux.org> | 2024-03-28 06:11:39 +0000 |
---|---|---|
committer | Daniel Baumann <daniel.baumann@progress-linux.org> | 2024-03-28 06:11:39 +0000 |
commit | 1fd6a618b60d7168fd8f37585d5d39d22d775afd (patch) | |
tree | fbc6d0c213b8acdd0a31deafe5c5fc0d05a3a312 /tests/units/cli/exec/test_utils.py | |
parent | Initial commit. (diff) | |
download | anta-1fd6a618b60d7168fd8f37585d5d39d22d775afd.tar.xz anta-1fd6a618b60d7168fd8f37585d5d39d22d775afd.zip |
Adding upstream version 0.13.0.upstream/0.13.0
Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>
Diffstat (limited to 'tests/units/cli/exec/test_utils.py')
-rw-r--r-- | tests/units/cli/exec/test_utils.py | 134 |
1 files changed, 134 insertions, 0 deletions
diff --git a/tests/units/cli/exec/test_utils.py b/tests/units/cli/exec/test_utils.py new file mode 100644 index 0000000..6df1c86 --- /dev/null +++ b/tests/units/cli/exec/test_utils.py @@ -0,0 +1,134 @@ +# 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 for anta.cli.exec.utils +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any +from unittest.mock import call, patch + +import pytest + +from anta.cli.exec.utils import clear_counters_utils # , collect_commands, collect_scheduled_show_tech +from anta.device import AntaDevice +from anta.inventory import AntaInventory +from anta.models import AntaCommand + +if TYPE_CHECKING: + from pytest import LogCaptureFixture + + +# TODO complete test cases +@pytest.mark.asyncio +@pytest.mark.parametrize( + "inventory_state, per_device_command_output, tags", + [ + pytest.param( + {"dummy": {"is_online": False}, "dummy2": {"is_online": False}, "dummy3": {"is_online": False}}, + {}, + None, + id="no_connected_device", + ), + pytest.param( + {"dummy": {"is_online": True, "hw_model": "cEOSLab"}, "dummy2": {"is_online": True, "hw_model": "vEOS-lab"}, "dummy3": {"is_online": False}}, + {}, + None, + id="cEOSLab and vEOS-lab devices", + ), + pytest.param( + {"dummy": {"is_online": True}, "dummy2": {"is_online": True}, "dummy3": {"is_online": False}}, + {"dummy": None}, # None means the command failed to collect + None, + id="device with error", + ), + pytest.param( + {"dummy": {"is_online": True}, "dummy2": {"is_online": True}, "dummy3": {"is_online": True}}, + {}, + ["spine"], + id="tags", + ), + ], +) +async def test_clear_counters_utils( + caplog: LogCaptureFixture, + test_inventory: AntaInventory, + inventory_state: dict[str, Any], + per_device_command_output: dict[str, Any], + tags: list[str] | None, +) -> None: + """ + Test anta.cli.exec.utils.clear_counters_utils + """ + + async def mock_connect_inventory() -> None: + """ + mocking connect_inventory coroutine + """ + for name, device in test_inventory.items(): + device.is_online = inventory_state[name].get("is_online", True) + device.established = inventory_state[name].get("established", device.is_online) + device.hw_model = inventory_state[name].get("hw_model", "dummy") + + async def dummy_collect(self: AntaDevice, command: AntaCommand) -> None: + """ + mocking collect coroutine + """ + command.output = per_device_command_output.get(self.name, "") + + # Need to patch the child device class + with patch("anta.device.AsyncEOSDevice.collect", side_effect=dummy_collect, autospec=True) as mocked_collect, patch( + "anta.inventory.AntaInventory.connect_inventory", + side_effect=mock_connect_inventory, + ) as mocked_connect_inventory: + print(mocked_collect) + mocked_collect.side_effect = dummy_collect + await clear_counters_utils(test_inventory, tags=tags) + + mocked_connect_inventory.assert_awaited_once() + devices_established = list(test_inventory.get_inventory(established_only=True, tags=tags).values()) + if devices_established: + # Building the list of calls + calls = [] + for device in devices_established: + calls.append( + call( + device, + **{ + "command": AntaCommand( + command="clear counters", + version="latest", + revision=None, + ofmt="json", + output=per_device_command_output.get(device.name, ""), + errors=[], + ) + }, + ) + ) + if device.hw_model not in ["cEOSLab", "vEOS-lab"]: + calls.append( + call( + device, + **{ + "command": AntaCommand( + command="clear hardware counter drop", + version="latest", + revision=None, + ofmt="json", + output=per_device_command_output.get(device.name, ""), + ) + }, + ) + ) + mocked_collect.assert_has_awaits(calls) + # Check error + for key, value in per_device_command_output.items(): + if value is None: + # means some command failed to collect + assert "ERROR" in caplog.text + assert f"Could not clear counters on device {key}: []" in caplog.text + else: + mocked_collect.assert_not_awaited() |