diff options
Diffstat (limited to 'tests/units/cli/exec')
| -rw-r--r-- | tests/units/cli/exec/__init__.py | 1 | ||||
| -rw-r--r-- | tests/units/cli/exec/test__init__.py | 18 | ||||
| -rw-r--r-- | tests/units/cli/exec/test_commands.py | 32 | ||||
| -rw-r--r-- | tests/units/cli/exec/test_utils.py | 113 |
4 files changed, 79 insertions, 85 deletions
diff --git a/tests/units/cli/exec/__init__.py b/tests/units/cli/exec/__init__.py index e772bee..4ed48bc 100644 --- a/tests/units/cli/exec/__init__.py +++ b/tests/units/cli/exec/__init__.py @@ -1,3 +1,4 @@ # 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. +"""Test anta.cli.exec submodule.""" diff --git a/tests/units/cli/exec/test__init__.py b/tests/units/cli/exec/test__init__.py index f8ad365..124d4af 100644 --- a/tests/units/cli/exec/test__init__.py +++ b/tests/units/cli/exec/test__init__.py @@ -1,30 +1,28 @@ # 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 -""" +"""Tests for anta.cli.exec.""" + from __future__ import annotations -from click.testing import CliRunner +from typing import TYPE_CHECKING from anta.cli import anta from anta.cli.utils import ExitCode +if TYPE_CHECKING: + from click.testing import CliRunner + def test_anta_exec(click_runner: CliRunner) -> None: - """ - Test anta exec - """ + """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 - """ + """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 index f96d7f6..4a72d63 100644 --- a/tests/units/cli/exec/test_commands.py +++ b/tests/units/cli/exec/test_commands.py @@ -1,9 +1,7 @@ # 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 -""" +"""Tests for anta.cli.exec.commands.""" from __future__ import annotations @@ -21,27 +19,21 @@ if TYPE_CHECKING: def test_clear_counters_help(click_runner: CliRunner) -> None: - """ - Test `anta exec clear-counters --help` - """ + """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` - """ + """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` - """ + """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 @@ -55,9 +47,7 @@ def test_collect_tech_support_help(click_runner: CliRunner) -> None: ], ) def test_clear_counters(click_runner: CliRunner, tags: str | None) -> None: - """ - Test `anta exec clear-counters` - """ + """Test `anta exec clear-counters`.""" cli_args = ["exec", "clear-counters"] if tags is not None: cli_args.extend(["--tags", tags]) @@ -69,7 +59,7 @@ COMMAND_LIST_PATH_FILE = Path(__file__).parent.parent.parent.parent / "data" / " @pytest.mark.parametrize( - "commands_path, tags", + ("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"), @@ -78,9 +68,7 @@ COMMAND_LIST_PATH_FILE = Path(__file__).parent.parent.parent.parent / "data" / " ], ) def test_snapshot(tmp_path: Path, click_runner: CliRunner, commands_path: Path | None, tags: str | None) -> None: - """ - Test `anta exec snapshot` - """ + """Test `anta exec snapshot`.""" cli_args = ["exec", "snapshot", "--output", str(tmp_path)] # Need to mock datetetime if commands_path is not None: @@ -99,7 +87,7 @@ def test_snapshot(tmp_path: Path, click_runner: CliRunner, commands_path: Path | @pytest.mark.parametrize( - "output, latest, configure, tags", + ("output", "latest", "configure", "tags"), [ pytest.param(None, None, False, None, id="no params"), pytest.param("/tmp/dummy", None, False, None, id="with output"), @@ -109,9 +97,7 @@ def test_snapshot(tmp_path: Path, click_runner: CliRunner, commands_path: Path | ], ) 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` - """ + """Test `anta exec collect-tech-support`.""" cli_args = ["exec", "collect-tech-support"] if output is not None: cli_args.extend(["--output", output]) diff --git a/tests/units/cli/exec/test_utils.py b/tests/units/cli/exec/test_utils.py index 6df1c86..455568b 100644 --- a/tests/units/cli/exec/test_utils.py +++ b/tests/units/cli/exec/test_utils.py @@ -1,9 +1,7 @@ # 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 -""" +"""Tests for anta.cli.exec.utils.""" from __future__ import annotations @@ -12,40 +10,59 @@ 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.cli.exec.utils import ( + clear_counters_utils, +) from anta.models import AntaCommand +# , collect_commands, collect_scheduled_show_tech + if TYPE_CHECKING: - from pytest import LogCaptureFixture + from anta.device import AntaDevice + from anta.inventory import AntaInventory -# TODO complete test cases -@pytest.mark.asyncio +# TODO: complete test cases +@pytest.mark.asyncio() @pytest.mark.parametrize( - "inventory_state, per_device_command_output, tags", + ("inventory_state", "per_device_command_output", "tags"), [ pytest.param( - {"dummy": {"is_online": False}, "dummy2": {"is_online": False}, "dummy3": {"is_online": False}}, + { + "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}}, + { + "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": {"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}}, + { + "dummy": {"is_online": True}, + "dummy2": {"is_online": True}, + "dummy3": {"is_online": True}, + }, {}, ["spine"], id="tags", @@ -53,42 +70,38 @@ if TYPE_CHECKING: ], ) async def test_clear_counters_utils( - caplog: LogCaptureFixture, + caplog: pytest.LogCaptureFixture, test_inventory: AntaInventory, inventory_state: dict[str, Any], per_device_command_output: dict[str, Any], - tags: list[str] | None, + tags: set[str] | None, ) -> None: - """ - Test anta.cli.exec.utils.clear_counters_utils - """ + """Test anta.cli.exec.utils.clear_counters_utils.""" async def mock_connect_inventory() -> None: - """ - mocking connect_inventory coroutine - """ + """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: - """ - mocking collect coroutine - """ + """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: - print(mocked_collect) + 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 = list(test_inventory.get_inventory(established_only=True, tags=tags).values()) + devices_established = test_inventory.get_inventory(established_only=True, tags=tags).devices if devices_established: # Building the list of calls calls = [] @@ -96,32 +109,28 @@ async def test_clear_counters_utils( calls.append( call( device, - **{ - "command": AntaCommand( - command="clear counters", - version="latest", - revision=None, - ofmt="json", - output=per_device_command_output.get(device.name, ""), - errors=[], - ) - }, - ) + 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, ""), - ) - }, - ) + 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 |