From 1fd6a618b60d7168fd8f37585d5d39d22d775afd Mon Sep 17 00:00:00 2001
From: Daniel Baumann <daniel.baumann@progress-linux.org>
Date: Thu, 28 Mar 2024 07:11:39 +0100
Subject: Adding upstream version 0.13.0.

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>
---
 tests/units/cli/exec/test_utils.py | 134 +++++++++++++++++++++++++++++++++++++
 1 file changed, 134 insertions(+)
 create mode 100644 tests/units/cli/exec/test_utils.py

(limited to 'tests/units/cli/exec/test_utils.py')

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()
-- 
cgit v1.2.3