1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
|
# 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,
)
from anta.models import AntaCommand
# , collect_commands, collect_scheduled_show_tech
if TYPE_CHECKING:
from anta.device import AntaDevice
from anta.inventory import AntaInventory
# 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: pytest.LogCaptureFixture,
test_inventory: AntaInventory,
inventory_state: dict[str, Any],
per_device_command_output: dict[str, Any],
tags: set[str] | None,
) -> None:
"""Test anta.cli.exec.utils.clear_counters_utils."""
async def mock_connect_inventory() -> None:
"""Mock 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:
"""Mock 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,
):
mocked_collect.side_effect = dummy_collect
await clear_counters_utils(test_inventory, tags=tags)
mocked_connect_inventory.assert_awaited_once()
devices_established = test_inventory.get_inventory(established_only=True, tags=tags).devices
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()
|
