diff options
author | Daniel Baumann <daniel.baumann@progress-linux.org> | 2024-03-28 06:11:39 +0000 |
---|---|---|
committer | Daniel Baumann <daniel.baumann@progress-linux.org> | 2024-03-28 06:11:39 +0000 |
commit | 1fd6a618b60d7168fd8f37585d5d39d22d775afd (patch) | |
tree | fbc6d0c213b8acdd0a31deafe5c5fc0d05a3a312 /tests/units/cli | |
parent | Initial commit. (diff) | |
download | anta-1fd6a618b60d7168fd8f37585d5d39d22d775afd.tar.xz anta-1fd6a618b60d7168fd8f37585d5d39d22d775afd.zip |
Adding upstream version 0.13.0.upstream/0.13.0
Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>
Diffstat (limited to 'tests/units/cli')
-rw-r--r-- | tests/units/cli/__init__.py | 3 | ||||
-rw-r--r-- | tests/units/cli/check/__init__.py | 3 | ||||
-rw-r--r-- | tests/units/cli/check/test__init__.py | 30 | ||||
-rw-r--r-- | tests/units/cli/check/test_commands.py | 37 | ||||
-rw-r--r-- | tests/units/cli/debug/__init__.py | 3 | ||||
-rw-r--r-- | tests/units/cli/debug/test__init__.py | 30 | ||||
-rw-r--r-- | tests/units/cli/debug/test_commands.py | 60 | ||||
-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 | ||||
-rw-r--r-- | tests/units/cli/get/__init__.py | 3 | ||||
-rw-r--r-- | tests/units/cli/get/test__init__.py | 30 | ||||
-rw-r--r-- | tests/units/cli/get/test_commands.py | 204 | ||||
-rw-r--r-- | tests/units/cli/get/test_utils.py | 115 | ||||
-rw-r--r-- | tests/units/cli/nrfu/__init__.py | 3 | ||||
-rw-r--r-- | tests/units/cli/nrfu/test__init__.py | 111 | ||||
-rw-r--r-- | tests/units/cli/nrfu/test_commands.py | 97 | ||||
-rw-r--r-- | tests/units/cli/test__init__.py | 58 |
19 files changed, 1079 insertions, 0 deletions
diff --git a/tests/units/cli/__init__.py b/tests/units/cli/__init__.py new file mode 100644 index 0000000..e772bee --- /dev/null +++ b/tests/units/cli/__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/check/__init__.py b/tests/units/cli/check/__init__.py new file mode 100644 index 0000000..e772bee --- /dev/null +++ b/tests/units/cli/check/__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/check/test__init__.py b/tests/units/cli/check/test__init__.py new file mode 100644 index 0000000..a3a770b --- /dev/null +++ b/tests/units/cli/check/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.check +""" +from __future__ import annotations + +from click.testing import CliRunner + +from anta.cli import anta +from anta.cli.utils import ExitCode + + +def test_anta_check(click_runner: CliRunner) -> None: + """ + Test anta check + """ + result = click_runner.invoke(anta, ["check"]) + assert result.exit_code == ExitCode.OK + assert "Usage: anta check" in result.output + + +def test_anta_check_help(click_runner: CliRunner) -> None: + """ + Test anta check --help + """ + result = click_runner.invoke(anta, ["check", "--help"]) + assert result.exit_code == ExitCode.OK + assert "Usage: anta check" in result.output diff --git a/tests/units/cli/check/test_commands.py b/tests/units/cli/check/test_commands.py new file mode 100644 index 0000000..746b315 --- /dev/null +++ b/tests/units/cli/check/test_commands.py @@ -0,0 +1,37 @@ +# 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.check.commands +""" +from __future__ import annotations + +from pathlib import Path +from typing import TYPE_CHECKING + +import pytest + +from anta.cli import anta +from anta.cli.utils import ExitCode + +if TYPE_CHECKING: + from click.testing import CliRunner + +DATA_DIR: Path = Path(__file__).parents[3].resolve() / "data" + + +@pytest.mark.parametrize( + "catalog_path, expected_exit, expected_output", + [ + pytest.param("ghost_catalog.yml", ExitCode.USAGE_ERROR, "Error: Invalid value for '--catalog'", id="catalog does not exist"), + pytest.param("test_catalog_with_undefined_module.yml", ExitCode.USAGE_ERROR, "Test catalog is invalid!", id="catalog is not valid"), + pytest.param("test_catalog.yml", ExitCode.OK, "Catalog is valid", id="catalog valid"), + ], +) +def test_catalog(click_runner: CliRunner, catalog_path: Path, expected_exit: int, expected_output: str) -> None: + """ + Test `anta check catalog -c catalog + """ + result = click_runner.invoke(anta, ["check", "catalog", "-c", str(DATA_DIR / catalog_path)]) + assert result.exit_code == expected_exit + assert expected_output in result.output diff --git a/tests/units/cli/debug/__init__.py b/tests/units/cli/debug/__init__.py new file mode 100644 index 0000000..e772bee --- /dev/null +++ b/tests/units/cli/debug/__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/debug/test__init__.py b/tests/units/cli/debug/test__init__.py new file mode 100644 index 0000000..062182d --- /dev/null +++ b/tests/units/cli/debug/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.debug +""" +from __future__ import annotations + +from click.testing import CliRunner + +from anta.cli import anta +from anta.cli.utils import ExitCode + + +def test_anta_debug(click_runner: CliRunner) -> None: + """ + Test anta debug + """ + result = click_runner.invoke(anta, ["debug"]) + assert result.exit_code == ExitCode.OK + assert "Usage: anta debug" in result.output + + +def test_anta_debug_help(click_runner: CliRunner) -> None: + """ + Test anta debug --help + """ + result = click_runner.invoke(anta, ["debug", "--help"]) + assert result.exit_code == ExitCode.OK + assert "Usage: anta debug" in result.output diff --git a/tests/units/cli/debug/test_commands.py b/tests/units/cli/debug/test_commands.py new file mode 100644 index 0000000..6d9ac29 --- /dev/null +++ b/tests/units/cli/debug/test_commands.py @@ -0,0 +1,60 @@ +# 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.debug.commands +""" +from __future__ import annotations + +from typing import TYPE_CHECKING, Literal + +import pytest + +from anta.cli import anta +from anta.cli.utils import ExitCode + +if TYPE_CHECKING: + from click.testing import CliRunner + + +@pytest.mark.parametrize( + "command, ofmt, version, revision, device, failed", + [ + pytest.param("show version", "json", None, None, "dummy", False, id="json command"), + pytest.param("show version", "text", None, None, "dummy", False, id="text command"), + pytest.param("show version", None, "latest", None, "dummy", False, id="version-latest"), + pytest.param("show version", None, "1", None, "dummy", False, id="version"), + pytest.param("show version", None, None, 3, "dummy", False, id="revision"), + pytest.param("undefined", None, None, None, "dummy", True, id="command fails"), + ], +) +def test_run_cmd( + click_runner: CliRunner, command: str, ofmt: Literal["json", "text"], version: Literal["1", "latest"] | None, revision: int | None, device: str, failed: bool +) -> None: + """ + Test `anta debug run-cmd` + """ + # pylint: disable=too-many-arguments + cli_args = ["-l", "debug", "debug", "run-cmd", "--command", command, "--device", device] + + # ofmt + if ofmt is not None: + cli_args.extend(["--ofmt", ofmt]) + + # version + if version is not None: + cli_args.extend(["--version", version]) + + # revision + if revision is not None: + cli_args.extend(["--revision", str(revision)]) + + result = click_runner.invoke(anta, cli_args) + if failed: + assert result.exit_code == ExitCode.USAGE_ERROR + else: + assert result.exit_code == ExitCode.OK + if revision is not None: + assert f"revision={revision}" in result.output + if version is not None: + assert (f"version='{version}'" if version == "latest" else f"version={version}") in result.output 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() diff --git a/tests/units/cli/get/__init__.py b/tests/units/cli/get/__init__.py new file mode 100644 index 0000000..e772bee --- /dev/null +++ b/tests/units/cli/get/__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/get/test__init__.py b/tests/units/cli/get/test__init__.py new file mode 100644 index 0000000..b18ef88 --- /dev/null +++ b/tests/units/cli/get/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.get +""" +from __future__ import annotations + +from click.testing import CliRunner + +from anta.cli import anta +from anta.cli.utils import ExitCode + + +def test_anta_get(click_runner: CliRunner) -> None: + """ + Test anta get + """ + result = click_runner.invoke(anta, ["get"]) + assert result.exit_code == ExitCode.OK + assert "Usage: anta get" in result.output + + +def test_anta_get_help(click_runner: CliRunner) -> None: + """ + Test anta get --help + """ + result = click_runner.invoke(anta, ["get", "--help"]) + assert result.exit_code == ExitCode.OK + assert "Usage: anta get" in result.output diff --git a/tests/units/cli/get/test_commands.py b/tests/units/cli/get/test_commands.py new file mode 100644 index 0000000..aa6dc4f --- /dev/null +++ b/tests/units/cli/get/test_commands.py @@ -0,0 +1,204 @@ +# 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.get.commands +""" +from __future__ import annotations + +import filecmp +from pathlib import Path +from typing import TYPE_CHECKING +from unittest.mock import ANY, patch + +import pytest +from cvprac.cvp_client import CvpClient +from cvprac.cvp_client_errors import CvpApiError + +from anta.cli import anta +from anta.cli.utils import ExitCode + +if TYPE_CHECKING: + from click.testing import CliRunner + +DATA_DIR: Path = Path(__file__).parents[3].resolve() / "data" + + +@pytest.mark.parametrize( + "cvp_container, cvp_connect_failure", + [ + pytest.param(None, False, id="all devices"), + pytest.param("custom_container", False, id="custom container"), + pytest.param(None, True, id="cvp connect failure"), + ], +) +def test_from_cvp( + tmp_path: Path, + click_runner: CliRunner, + cvp_container: str | None, + cvp_connect_failure: bool, +) -> None: + """ + Test `anta get from-cvp` + + This test verifies that username and password are NOT mandatory to run this command + """ + output: Path = tmp_path / "output.yml" + cli_args = ["get", "from-cvp", "--output", str(output), "--host", "42.42.42.42", "--username", "anta", "--password", "anta"] + + if cvp_container is not None: + cli_args.extend(["--container", cvp_container]) + + def mock_cvp_connect(self: CvpClient, *args: str, **kwargs: str) -> None: + # pylint: disable=unused-argument + if cvp_connect_failure: + raise CvpApiError(msg="mocked CvpApiError") + + # always get a token + with patch("anta.cli.get.commands.get_cv_token", return_value="dummy_token"), patch( + "cvprac.cvp_client.CvpClient.connect", autospec=True, side_effect=mock_cvp_connect + ) as mocked_cvp_connect, patch("cvprac.cvp_client.CvpApi.get_inventory", autospec=True, return_value=[]) as mocked_get_inventory, patch( + "cvprac.cvp_client.CvpApi.get_devices_in_container", autospec=True, return_value=[] + ) as mocked_get_devices_in_container: + result = click_runner.invoke(anta, cli_args) + + if not cvp_connect_failure: + assert output.exists() + + mocked_cvp_connect.assert_called_once() + if not cvp_connect_failure: + assert "Connected to CloudVision" in result.output + if cvp_container is not None: + mocked_get_devices_in_container.assert_called_once_with(ANY, cvp_container) + else: + mocked_get_inventory.assert_called_once() + assert result.exit_code == ExitCode.OK + else: + assert "Error connecting to CloudVision" in result.output + assert result.exit_code == ExitCode.USAGE_ERROR + + +@pytest.mark.parametrize( + "ansible_inventory, ansible_group, expected_exit, expected_log", + [ + pytest.param("ansible_inventory.yml", None, ExitCode.OK, None, id="no group"), + pytest.param("ansible_inventory.yml", "ATD_LEAFS", ExitCode.OK, None, id="group found"), + pytest.param("ansible_inventory.yml", "DUMMY", ExitCode.USAGE_ERROR, "Group DUMMY not found in Ansible inventory", id="group not found"), + pytest.param("empty_ansible_inventory.yml", None, ExitCode.USAGE_ERROR, "is empty", id="empty inventory"), + ], +) +def test_from_ansible( + tmp_path: Path, + click_runner: CliRunner, + ansible_inventory: Path, + ansible_group: str | None, + expected_exit: int, + expected_log: str | None, +) -> None: + """ + Test `anta get from-ansible` + + This test verifies: + * the parsing of an ansible-inventory + * the ansible_group functionaliy + + The output path is ALWAYS set to a non existing file. + """ + output: Path = tmp_path / "output.yml" + ansible_inventory_path = DATA_DIR / ansible_inventory + # Init cli_args + cli_args = ["get", "from-ansible", "--output", str(output), "--ansible-inventory", str(ansible_inventory_path)] + + # Set --ansible-group + if ansible_group is not None: + cli_args.extend(["--ansible-group", ansible_group]) + + result = click_runner.invoke(anta, cli_args) + + assert result.exit_code == expected_exit + + if expected_exit != ExitCode.OK: + assert expected_log + assert expected_log in result.output + else: + assert output.exists() + # TODO check size of generated inventory to validate the group functionality! + + +@pytest.mark.parametrize( + "env_set, overwrite, is_tty, prompt, expected_exit, expected_log", + [ + pytest.param(True, False, True, "y", ExitCode.OK, "", id="no-overwrite-tty-init-prompt-yes"), + pytest.param(True, False, True, "N", ExitCode.INTERNAL_ERROR, "Aborted", id="no-overwrite-tty-init-prompt-no"), + pytest.param( + True, + False, + False, + None, + ExitCode.USAGE_ERROR, + "Conversion aborted since destination file is not empty (not running in interactive TTY)", + id="no-overwrite-no-tty-init", + ), + pytest.param(False, False, True, None, ExitCode.OK, "", id="no-overwrite-tty-no-init"), + pytest.param(False, False, False, None, ExitCode.OK, "", id="no-overwrite-no-tty-no-init"), + pytest.param(True, True, True, None, ExitCode.OK, "", id="overwrite-tty-init"), + pytest.param(True, True, False, None, ExitCode.OK, "", id="overwrite-no-tty-init"), + pytest.param(False, True, True, None, ExitCode.OK, "", id="overwrite-tty-no-init"), + pytest.param(False, True, False, None, ExitCode.OK, "", id="overwrite-no-tty-no-init"), + ], +) +def test_from_ansible_overwrite( + tmp_path: Path, + click_runner: CliRunner, + temp_env: dict[str, str | None], + env_set: bool, + overwrite: bool, + is_tty: bool, + prompt: str | None, + expected_exit: int, + expected_log: str | None, +) -> None: + # pylint: disable=too-many-arguments + """ + Test `anta get from-ansible` overwrite mechanism + + The test uses a static ansible-inventory and output as these are tested in other functions + + This test verifies: + * that overwrite is working as expected with or without init data in the target file + * that when the target file is not empty and a tty is present, the user is prompt with confirmation + * Check the behavior when the prompt is filled + + The initial content of the ANTA inventory is set using init_anta_inventory, if it is None, no inventory is set. + + * With overwrite True, the expectation is that the from-ansible command succeeds + * With no init (init_anta_inventory == None), the expectation is also that command succeeds + """ + ansible_inventory_path = DATA_DIR / "ansible_inventory.yml" + expected_anta_inventory_path = DATA_DIR / "expected_anta_inventory.yml" + tmp_output = tmp_path / "output.yml" + cli_args = ["get", "from-ansible", "--ansible-inventory", str(ansible_inventory_path)] + + if env_set: + tmp_inv = Path(str(temp_env["ANTA_INVENTORY"])) + else: + temp_env["ANTA_INVENTORY"] = None + tmp_inv = tmp_output + cli_args.extend(["--output", str(tmp_output)]) + + if overwrite: + cli_args.append("--overwrite") + + # Verify initial content is different + if tmp_inv.exists(): + assert not filecmp.cmp(tmp_inv, expected_anta_inventory_path) + + with patch("sys.stdin.isatty", return_value=is_tty): + result = click_runner.invoke(anta, cli_args, env=temp_env, input=prompt) + + assert result.exit_code == expected_exit + if expected_exit == ExitCode.OK: + assert filecmp.cmp(tmp_inv, expected_anta_inventory_path) + elif expected_exit == ExitCode.INTERNAL_ERROR: + assert expected_log + assert expected_log in result.output diff --git a/tests/units/cli/get/test_utils.py b/tests/units/cli/get/test_utils.py new file mode 100644 index 0000000..b335880 --- /dev/null +++ b/tests/units/cli/get/test_utils.py @@ -0,0 +1,115 @@ +# 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.get.utils +""" +from __future__ import annotations + +from contextlib import nullcontext +from pathlib import Path +from typing import Any +from unittest.mock import MagicMock, patch + +import pytest +import requests + +from anta.cli.get.utils import create_inventory_from_ansible, create_inventory_from_cvp, get_cv_token +from anta.inventory import AntaInventory + +DATA_DIR: Path = Path(__file__).parents[3].resolve() / "data" + + +def test_get_cv_token() -> None: + """ + Test anta.get.utils.get_cv_token + """ + ip = "42.42.42.42" + username = "ant" + password = "formica" + + with patch("anta.cli.get.utils.requests.request") as patched_request: + mocked_ret = MagicMock(autospec=requests.Response) + mocked_ret.json.return_value = {"sessionId": "simple"} + patched_request.return_value = mocked_ret + res = get_cv_token(ip, username, password) + patched_request.assert_called_once_with( + "POST", + "https://42.42.42.42/cvpservice/login/authenticate.do", + headers={"Content-Type": "application/json", "Accept": "application/json"}, + data='{"userId": "ant", "password": "formica"}', + verify=False, + timeout=10, + ) + assert res == "simple" + + +# truncated inventories +CVP_INVENTORY = [ + { + "hostname": "device1", + "containerName": "DC1", + "ipAddress": "10.20.20.97", + }, + { + "hostname": "device2", + "containerName": "DC2", + "ipAddress": "10.20.20.98", + }, + { + "hostname": "device3", + "containerName": "", + "ipAddress": "10.20.20.99", + }, +] + + +@pytest.mark.parametrize( + "inventory", + [ + pytest.param(CVP_INVENTORY, id="some container"), + pytest.param([], id="empty_inventory"), + ], +) +def test_create_inventory_from_cvp(tmp_path: Path, inventory: list[dict[str, Any]]) -> None: + """ + Test anta.get.utils.create_inventory_from_cvp + """ + output = tmp_path / "output.yml" + + create_inventory_from_cvp(inventory, output) + + assert output.exists() + # This validate the file structure ;) + inv = AntaInventory.parse(str(output), "user", "pass") + assert len(inv) == len(inventory) + + +@pytest.mark.parametrize( + "inventory_filename, ansible_group, expected_raise, expected_inv_length", + [ + pytest.param("ansible_inventory.yml", None, nullcontext(), 7, id="no group"), + pytest.param("ansible_inventory.yml", "ATD_LEAFS", nullcontext(), 4, id="group found"), + pytest.param("ansible_inventory.yml", "DUMMY", pytest.raises(ValueError, match="Group DUMMY not found in Ansible inventory"), 0, id="group not found"), + pytest.param("empty_ansible_inventory.yml", None, pytest.raises(ValueError, match="Ansible inventory .* is empty"), 0, id="empty inventory"), + pytest.param("wrong_ansible_inventory.yml", None, pytest.raises(ValueError, match="Could not parse"), 0, id="os error inventory"), + ], +) +def test_create_inventory_from_ansible(tmp_path: Path, inventory_filename: Path, ansible_group: str | None, expected_raise: Any, expected_inv_length: int) -> None: + """ + Test anta.get.utils.create_inventory_from_ansible + """ + target_file = tmp_path / "inventory.yml" + inventory_file_path = DATA_DIR / inventory_filename + + with expected_raise: + if ansible_group: + create_inventory_from_ansible(inventory_file_path, target_file, ansible_group) + else: + create_inventory_from_ansible(inventory_file_path, target_file) + + assert target_file.exists() + inv = AntaInventory().parse(str(target_file), "user", "pass") + assert len(inv) == expected_inv_length + if not isinstance(expected_raise, nullcontext): + assert not target_file.exists() diff --git a/tests/units/cli/nrfu/__init__.py b/tests/units/cli/nrfu/__init__.py new file mode 100644 index 0000000..e772bee --- /dev/null +++ b/tests/units/cli/nrfu/__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/nrfu/test__init__.py b/tests/units/cli/nrfu/test__init__.py new file mode 100644 index 0000000..fea641c --- /dev/null +++ b/tests/units/cli/nrfu/test__init__.py @@ -0,0 +1,111 @@ +# 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.nrfu +""" +from __future__ import annotations + +from click.testing import CliRunner + +from anta.cli import anta +from anta.cli.utils import ExitCode +from tests.lib.utils import default_anta_env + +# TODO: write unit tests for ignore-status and ignore-error + + +def test_anta_nrfu_help(click_runner: CliRunner) -> None: + """ + Test anta nrfu --help + """ + result = click_runner.invoke(anta, ["nrfu", "--help"]) + assert result.exit_code == ExitCode.OK + assert "Usage: anta nrfu" in result.output + + +def test_anta_nrfu(click_runner: CliRunner) -> None: + """ + Test anta nrfu, catalog is given via env + """ + result = click_runner.invoke(anta, ["nrfu"]) + assert result.exit_code == ExitCode.OK + assert "ANTA Inventory contains 3 devices" in result.output + assert "Tests catalog contains 1 tests" in result.output + + +def test_anta_password_required(click_runner: CliRunner) -> None: + """ + Test that password is provided + """ + env = default_anta_env() + env["ANTA_PASSWORD"] = None + result = click_runner.invoke(anta, ["nrfu"], env=env) + + assert result.exit_code == ExitCode.USAGE_ERROR + assert "EOS password needs to be provided by using either the '--password' option or the '--prompt' option." in result.output + + +def test_anta_password(click_runner: CliRunner) -> None: + """ + Test that password can be provided either via --password or --prompt + """ + env = default_anta_env() + env["ANTA_PASSWORD"] = None + result = click_runner.invoke(anta, ["nrfu", "--password", "secret"], env=env) + assert result.exit_code == ExitCode.OK + result = click_runner.invoke(anta, ["nrfu", "--prompt"], input="password\npassword\n", env=env) + assert result.exit_code == ExitCode.OK + + +def test_anta_enable_password(click_runner: CliRunner) -> None: + """ + Test that enable password can be provided either via --enable-password or --prompt + """ + # Both enable and enable-password + result = click_runner.invoke(anta, ["nrfu", "--enable", "--enable-password", "secret"]) + assert result.exit_code == ExitCode.OK + + # enable and prompt y + result = click_runner.invoke(anta, ["nrfu", "--enable", "--prompt"], input="y\npassword\npassword\n") + assert "Is a password required to enter EOS privileged EXEC mode? [y/N]:" in result.output + assert "Please enter a password to enter EOS privileged EXEC mode" in result.output + assert result.exit_code == ExitCode.OK + + # enable and prompt N + result = click_runner.invoke(anta, ["nrfu", "--enable", "--prompt"], input="N\n") + assert "Is a password required to enter EOS privileged EXEC mode? [y/N]:" in result.output + assert "Please enter a password to enter EOS privileged EXEC mode" not in result.output + assert result.exit_code == ExitCode.OK + + # enable and enable-password and prompt (redundant) + result = click_runner.invoke(anta, ["nrfu", "--enable", "--enable-password", "blah", "--prompt"], input="y\npassword\npassword\n") + assert "Is a password required to enter EOS privileged EXEC mode? [y/N]:" not in result.output + assert "Please enter a password to enter EOS privileged EXEC mode" not in result.output + assert result.exit_code == ExitCode.OK + + # enabled-password without enable + result = click_runner.invoke(anta, ["nrfu", "--enable-password", "blah"]) + assert result.exit_code == ExitCode.USAGE_ERROR + assert "Providing a password to access EOS Privileged EXEC mode requires '--enable' option." in result.output + + +def test_anta_enable_alone(click_runner: CliRunner) -> None: + """ + Test that enable can be provided either without enable-password + """ + result = click_runner.invoke(anta, ["nrfu", "--enable"]) + assert result.exit_code == ExitCode.OK + + +def test_disable_cache(click_runner: CliRunner) -> None: + """ + Test that disable_cache is working on inventory + """ + result = click_runner.invoke(anta, ["nrfu", "--disable-cache"]) + stdout_lines = result.stdout.split("\n") + # All caches should be disabled from the inventory + for line in stdout_lines: + if "disable_cache" in line: + assert "True" in line + assert result.exit_code == ExitCode.OK diff --git a/tests/units/cli/nrfu/test_commands.py b/tests/units/cli/nrfu/test_commands.py new file mode 100644 index 0000000..4639671 --- /dev/null +++ b/tests/units/cli/nrfu/test_commands.py @@ -0,0 +1,97 @@ +# 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.nrfu.commands +""" +from __future__ import annotations + +import json +import re +from pathlib import Path + +from click.testing import CliRunner + +from anta.cli import anta +from anta.cli.utils import ExitCode + +DATA_DIR: Path = Path(__file__).parent.parent.parent.parent.resolve() / "data" + + +def test_anta_nrfu_table_help(click_runner: CliRunner) -> None: + """ + Test anta nrfu table --help + """ + result = click_runner.invoke(anta, ["nrfu", "table", "--help"]) + assert result.exit_code == ExitCode.OK + assert "Usage: anta nrfu table" in result.output + + +def test_anta_nrfu_text_help(click_runner: CliRunner) -> None: + """ + Test anta nrfu text --help + """ + result = click_runner.invoke(anta, ["nrfu", "text", "--help"]) + assert result.exit_code == ExitCode.OK + assert "Usage: anta nrfu text" in result.output + + +def test_anta_nrfu_json_help(click_runner: CliRunner) -> None: + """ + Test anta nrfu json --help + """ + result = click_runner.invoke(anta, ["nrfu", "json", "--help"]) + assert result.exit_code == ExitCode.OK + assert "Usage: anta nrfu json" in result.output + + +def test_anta_nrfu_template_help(click_runner: CliRunner) -> None: + """ + Test anta nrfu tpl-report --help + """ + result = click_runner.invoke(anta, ["nrfu", "tpl-report", "--help"]) + assert result.exit_code == ExitCode.OK + assert "Usage: anta nrfu tpl-report" in result.output + + +def test_anta_nrfu_table(click_runner: CliRunner) -> None: + """ + Test anta nrfu, catalog is given via env + """ + result = click_runner.invoke(anta, ["nrfu", "table"]) + assert result.exit_code == ExitCode.OK + assert "dummy │ VerifyEOSVersion │ success" in result.output + + +def test_anta_nrfu_text(click_runner: CliRunner) -> None: + """ + Test anta nrfu, catalog is given via env + """ + result = click_runner.invoke(anta, ["nrfu", "text"]) + assert result.exit_code == ExitCode.OK + assert "dummy :: VerifyEOSVersion :: SUCCESS" in result.output + + +def test_anta_nrfu_json(click_runner: CliRunner) -> None: + """ + Test anta nrfu, catalog is given via env + """ + result = click_runner.invoke(anta, ["nrfu", "json"]) + assert result.exit_code == ExitCode.OK + assert "JSON results of all tests" in result.output + m = re.search(r"\[\n {[\s\S]+ }\n\]", result.output) + assert m is not None + result_list = json.loads(m.group()) + for r in result_list: + if r["name"] == "dummy": + assert r["test"] == "VerifyEOSVersion" + assert r["result"] == "success" + + +def test_anta_nrfu_template(click_runner: CliRunner) -> None: + """ + Test anta nrfu, catalog is given via env + """ + result = click_runner.invoke(anta, ["nrfu", "tpl-report", "--template", str(DATA_DIR / "template.j2")]) + assert result.exit_code == ExitCode.OK + assert "* VerifyEOSVersion is SUCCESS for dummy" in result.output diff --git a/tests/units/cli/test__init__.py b/tests/units/cli/test__init__.py new file mode 100644 index 0000000..0e84e14 --- /dev/null +++ b/tests/units/cli/test__init__.py @@ -0,0 +1,58 @@ +# 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.__init__ +""" + +from __future__ import annotations + +from click.testing import CliRunner + +from anta.cli import anta +from anta.cli.utils import ExitCode + + +def test_anta(click_runner: CliRunner) -> None: + """ + Test anta main entrypoint + """ + result = click_runner.invoke(anta) + assert result.exit_code == ExitCode.OK + assert "Usage" in result.output + + +def test_anta_help(click_runner: CliRunner) -> None: + """ + Test anta --help + """ + result = click_runner.invoke(anta, ["--help"]) + assert result.exit_code == ExitCode.OK + assert "Usage" 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 + + +def test_anta_debug_help(click_runner: CliRunner) -> None: + """ + Test anta debug --help + """ + result = click_runner.invoke(anta, ["debug", "--help"]) + assert result.exit_code == ExitCode.OK + assert "Usage: anta debug" in result.output + + +def test_anta_get_help(click_runner: CliRunner) -> None: + """ + Test anta get --help + """ + result = click_runner.invoke(anta, ["get", "--help"]) + assert result.exit_code == ExitCode.OK + assert "Usage: anta get" in result.output |