diff options
Diffstat (limited to 'tests/units/cli/exec')
| -rw-r--r-- | tests/units/cli/exec/__init__.py | 3 | ||||
| -rw-r--r-- | tests/units/cli/exec/test__init__.py | 30 | ||||
| -rw-r--r-- | tests/units/cli/exec/test_commands.py | 125 | ||||
| -rw-r--r-- | tests/units/cli/exec/test_utils.py | 134 |
4 files changed, 292 insertions, 0 deletions
diff --git a/tests/units/cli/exec/__init__.py b/tests/units/cli/exec/__init__.py new file mode 100644 index 0000000..e772bee --- /dev/null +++ b/tests/units/cli/exec/__init__.py @@ -0,0 +1,3 @@ +# 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. diff --git a/tests/units/cli/exec/test__init__.py b/tests/units/cli/exec/test__init__.py new file mode 100644 index 0000000..f8ad365 --- /dev/null +++ b/tests/units/cli/exec/test__init__.py @@ -0,0 +1,30 @@ +# 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 +""" +from __future__ import annotations + +from click.testing import CliRunner + +from anta.cli import anta +from anta.cli.utils import ExitCode + + +def test_anta_exec(click_runner: CliRunner) -> None: + """ + Test anta exec + """ + result = click_runner.invoke(anta, ["exec"]) + assert result.exit_code == ExitCode.OK + assert "Usage: anta exec" in result.output + + +def test_anta_exec_help(click_runner: CliRunner) -> None: + """ + Test anta exec --help + """ + result = click_runner.invoke(anta, ["exec", "--help"]) + assert result.exit_code == ExitCode.OK + assert "Usage: anta exec" in result.output diff --git a/tests/units/cli/exec/test_commands.py b/tests/units/cli/exec/test_commands.py new file mode 100644 index 0000000..f96d7f6 --- /dev/null +++ b/tests/units/cli/exec/test_commands.py @@ -0,0 +1,125 @@ +# 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.commands +""" + +from __future__ import annotations + +from pathlib import Path +from typing import TYPE_CHECKING + +import pytest + +from anta.cli import anta +from anta.cli.exec.commands import clear_counters, collect_tech_support, snapshot +from anta.cli.utils import ExitCode + +if TYPE_CHECKING: + from click.testing import CliRunner + + +def test_clear_counters_help(click_runner: CliRunner) -> None: + """ + Test `anta exec clear-counters --help` + """ + result = click_runner.invoke(clear_counters, ["--help"]) + assert result.exit_code == 0 + assert "Usage" in result.output + + +def test_snapshot_help(click_runner: CliRunner) -> None: + """ + Test `anta exec snapshot --help` + """ + result = click_runner.invoke(snapshot, ["--help"]) + assert result.exit_code == 0 + assert "Usage" in result.output + + +def test_collect_tech_support_help(click_runner: CliRunner) -> None: + """ + Test `anta exec collect-tech-support --help` + """ + result = click_runner.invoke(collect_tech_support, ["--help"]) + assert result.exit_code == 0 + assert "Usage" in result.output + + +@pytest.mark.parametrize( + "tags", + [ + pytest.param(None, id="no tags"), + pytest.param("leaf,spine", id="with tags"), + ], +) +def test_clear_counters(click_runner: CliRunner, tags: str | None) -> None: + """ + Test `anta exec clear-counters` + """ + cli_args = ["exec", "clear-counters"] + if tags is not None: + cli_args.extend(["--tags", tags]) + result = click_runner.invoke(anta, cli_args) + assert result.exit_code == ExitCode.OK + + +COMMAND_LIST_PATH_FILE = Path(__file__).parent.parent.parent.parent / "data" / "test_snapshot_commands.yml" + + +@pytest.mark.parametrize( + "commands_path, tags", + [ + pytest.param(None, None, id="missing command list"), + pytest.param(Path("/I/do/not/exist"), None, id="wrong path for command_list"), + pytest.param(COMMAND_LIST_PATH_FILE, None, id="command-list only"), + pytest.param(COMMAND_LIST_PATH_FILE, "leaf,spine", id="with tags"), + ], +) +def test_snapshot(tmp_path: Path, click_runner: CliRunner, commands_path: Path | None, tags: str | None) -> None: + """ + Test `anta exec snapshot` + """ + cli_args = ["exec", "snapshot", "--output", str(tmp_path)] + # Need to mock datetetime + if commands_path is not None: + cli_args.extend(["--commands-list", str(commands_path)]) + if tags is not None: + cli_args.extend(["--tags", tags]) + result = click_runner.invoke(anta, cli_args) + # Failure scenarios + if commands_path is None: + assert result.exit_code == ExitCode.USAGE_ERROR + return + if not Path.exists(Path(commands_path)): + assert result.exit_code == ExitCode.USAGE_ERROR + return + assert result.exit_code == ExitCode.OK + + +@pytest.mark.parametrize( + "output, latest, configure, tags", + [ + pytest.param(None, None, False, None, id="no params"), + pytest.param("/tmp/dummy", None, False, None, id="with output"), + pytest.param(None, 1, False, None, id="only last show tech"), + pytest.param(None, None, True, None, id="configure"), + pytest.param(None, None, False, "leaf,spine", id="with tags"), + ], +) +def test_collect_tech_support(click_runner: CliRunner, output: str | None, latest: str | None, configure: bool | None, tags: str | None) -> None: + """ + Test `anta exec collect-tech-support` + """ + cli_args = ["exec", "collect-tech-support"] + if output is not None: + cli_args.extend(["--output", output]) + if latest is not None: + cli_args.extend(["--latest", latest]) + if configure is True: + cli_args.extend(["--configure"]) + if tags is not None: + cli_args.extend(["--tags", tags]) + result = click_runner.invoke(anta, cli_args) + assert result.exit_code == ExitCode.OK 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() |