Adding upstream version 0.13.0.

Signed-off-by: Daniel Baumann <daniel@debian.org>

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.

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

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

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