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
|
# 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()
|
