summaryrefslogtreecommitdiffstats
path: root/tests/units/cli
diff options
context:
space:
mode:
Diffstat (limited to 'tests/units/cli')
-rw-r--r--tests/units/cli/debug/test_commands.py1
-rw-r--r--tests/units/cli/exec/test_utils.py15
-rw-r--r--tests/units/cli/get/test__init__.py2
-rw-r--r--tests/units/cli/get/test_commands.py2
-rw-r--r--tests/units/cli/get/test_utils.py30
-rw-r--r--tests/units/cli/nrfu/test__init__.py17
-rw-r--r--tests/units/cli/nrfu/test_commands.py16
-rw-r--r--tests/units/cli/test__init__.py77
-rw-r--r--tests/units/cli/test_main.py64
9 files changed, 168 insertions, 56 deletions
diff --git a/tests/units/cli/debug/test_commands.py b/tests/units/cli/debug/test_commands.py
index 76c3648..039e09e 100644
--- a/tests/units/cli/debug/test_commands.py
+++ b/tests/units/cli/debug/test_commands.py
@@ -25,6 +25,7 @@ if TYPE_CHECKING:
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"),
+ pytest.param("undefined", None, None, None, "doesnotexist", True, id="Device does not exist"),
],
)
def test_run_cmd(
diff --git a/tests/units/cli/exec/test_utils.py b/tests/units/cli/exec/test_utils.py
index 455568b..ad1a78a 100644
--- a/tests/units/cli/exec/test_utils.py
+++ b/tests/units/cli/exec/test_utils.py
@@ -11,7 +11,7 @@ from unittest.mock import call, patch
import pytest
from anta.cli.exec.utils import (
- clear_counters_utils,
+ clear_counters,
)
from anta.models import AntaCommand
@@ -69,14 +69,14 @@ if TYPE_CHECKING:
),
],
)
-async def test_clear_counters_utils(
+async def test_clear_counters(
caplog: pytest.LogCaptureFixture,
test_inventory: AntaInventory,
inventory_state: dict[str, Any],
per_device_command_output: dict[str, Any],
tags: set[str] | None,
) -> None:
- """Test anta.cli.exec.utils.clear_counters_utils."""
+ """Test anta.cli.exec.utils.clear_counters."""
async def mock_connect_inventory() -> None:
"""Mock connect_inventory coroutine."""
@@ -85,20 +85,19 @@ async def test_clear_counters_utils(
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:
+ async def collect(self: AntaDevice, command: AntaCommand, *args: Any, **kwargs: Any) -> None: # noqa: ARG001, ANN401 #pylint: disable=unused-argument
"""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.device.AsyncEOSDevice.collect", side_effect=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)
+ await clear_counters(test_inventory, tags=tags)
mocked_connect_inventory.assert_awaited_once()
devices_established = test_inventory.get_inventory(established_only=True, tags=tags).devices
@@ -117,6 +116,7 @@ async def test_clear_counters_utils(
output=per_device_command_output.get(device.name, ""),
errors=[],
),
+ collection_id=None,
),
)
if device.hw_model not in ["cEOSLab", "vEOS-lab"]:
@@ -130,6 +130,7 @@ async def test_clear_counters_utils(
ofmt="json",
output=per_device_command_output.get(device.name, ""),
),
+ collection_id=None,
),
)
mocked_collect.assert_has_awaits(calls)
diff --git a/tests/units/cli/get/test__init__.py b/tests/units/cli/get/test__init__.py
index a6a0c3c..1ef65c2 100644
--- a/tests/units/cli/get/test__init__.py
+++ b/tests/units/cli/get/test__init__.py
@@ -7,7 +7,7 @@ from __future__ import annotations
from typing import TYPE_CHECKING
-from anta.cli import anta
+from anta.cli._main import anta
from anta.cli.utils import ExitCode
if TYPE_CHECKING:
diff --git a/tests/units/cli/get/test_commands.py b/tests/units/cli/get/test_commands.py
index 9edc7c3..e0b17a0 100644
--- a/tests/units/cli/get/test_commands.py
+++ b/tests/units/cli/get/test_commands.py
@@ -13,7 +13,7 @@ from unittest.mock import ANY, patch
import pytest
from cvprac.cvp_client_errors import CvpApiError
-from anta.cli import anta
+from anta.cli._main import anta
from anta.cli.utils import ExitCode
if TYPE_CHECKING:
diff --git a/tests/units/cli/get/test_utils.py b/tests/units/cli/get/test_utils.py
index 0dce335..7ce85dc 100644
--- a/tests/units/cli/get/test_utils.py
+++ b/tests/units/cli/get/test_utils.py
@@ -81,14 +81,15 @@ def test_create_inventory_from_cvp(tmp_path: Path, inventory: list[dict[str, Any
@pytest.mark.parametrize(
- ("inventory_filename", "ansible_group", "expected_raise", "expected_inv_length"),
+ ("inventory_filename", "ansible_group", "expected_raise", "expected_log", "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", None, nullcontext(), None, 7, id="no group"),
+ pytest.param("ansible_inventory.yml", "ATD_LEAFS", nullcontext(), None, 4, id="group found"),
pytest.param(
"ansible_inventory.yml",
"DUMMY",
pytest.raises(ValueError, match="Group DUMMY not found in Ansible inventory"),
+ None,
0,
id="group not found",
),
@@ -96,6 +97,7 @@ def test_create_inventory_from_cvp(tmp_path: Path, inventory: list[dict[str, Any
"empty_ansible_inventory.yml",
None,
pytest.raises(ValueError, match="Ansible inventory .* is empty"),
+ None,
0,
id="empty inventory",
),
@@ -103,19 +105,39 @@ def test_create_inventory_from_cvp(tmp_path: Path, inventory: list[dict[str, Any
"wrong_ansible_inventory.yml",
None,
pytest.raises(ValueError, match="Could not parse"),
+ None,
0,
id="os error inventory",
),
+ pytest.param(
+ "ansible_inventory_with_vault.yml",
+ None,
+ pytest.raises(ValueError, match="Could not parse"),
+ "`anta get from-ansible` does not support inline vaulted variables",
+ 0,
+ id="Vault variable in inventory",
+ ),
+ pytest.param(
+ "ansible_inventory_unknown_yaml_tag.yml",
+ None,
+ pytest.raises(ValueError, match="Could not parse"),
+ None,
+ 0,
+ id="Unknown YAML tag in inventory",
+ ),
],
)
def test_create_inventory_from_ansible(
+ caplog: pytest.LogCaptureFixture,
tmp_path: Path,
inventory_filename: Path,
ansible_group: str | None,
expected_raise: AbstractContextManager[Exception],
+ expected_log: str | None,
expected_inv_length: int,
) -> None:
"""Test anta.get.utils.create_inventory_from_ansible."""
+ # pylint: disable=R0913
target_file = tmp_path / "inventory.yml"
inventory_file_path = DATA_DIR / inventory_filename
@@ -130,3 +152,5 @@ def test_create_inventory_from_ansible(
assert len(inv) == expected_inv_length
if not isinstance(expected_raise, nullcontext):
assert not target_file.exists()
+ if expected_log:
+ assert expected_log in caplog.text
diff --git a/tests/units/cli/nrfu/test__init__.py b/tests/units/cli/nrfu/test__init__.py
index 052c7c3..a9dcd9c 100644
--- a/tests/units/cli/nrfu/test__init__.py
+++ b/tests/units/cli/nrfu/test__init__.py
@@ -24,6 +24,14 @@ def test_anta_nrfu_help(click_runner: CliRunner) -> None:
assert "Usage: anta nrfu" in result.output
+def test_anta_nrfu_wrong_subcommand(click_runner: CliRunner) -> None:
+ """Test anta nrfu toast."""
+ result = click_runner.invoke(anta, ["nrfu", "oook"])
+ assert result.exit_code == ExitCode.USAGE_ERROR
+ assert "Usage: anta nrfu" in result.output
+ assert "No such command 'oook'." 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"])
@@ -32,6 +40,15 @@ def test_anta_nrfu(click_runner: CliRunner) -> None:
assert "Tests catalog contains 1 tests" in result.output
+def test_anta_nrfu_dry_run(click_runner: CliRunner) -> None:
+ """Test anta nrfu --dry-run, catalog is given via env."""
+ result = click_runner.invoke(anta, ["nrfu", "--dry-run"])
+ assert result.exit_code == ExitCode.OK
+ assert "ANTA Inventory contains 3 devices" in result.output
+ assert "Tests catalog contains 1 tests" in result.output
+ assert "Dry-run" in result.output
+
+
def test_anta_password_required(click_runner: CliRunner) -> None:
"""Test that password is provided."""
env = default_anta_env()
diff --git a/tests/units/cli/nrfu/test_commands.py b/tests/units/cli/nrfu/test_commands.py
index 4ea40b7..e2b5031 100644
--- a/tests/units/cli/nrfu/test_commands.py
+++ b/tests/units/cli/nrfu/test_commands.py
@@ -54,6 +54,20 @@ def test_anta_nrfu_table(click_runner: CliRunner) -> None:
assert "dummy │ VerifyEOSVersion │ success" in result.output
+def test_anta_nrfu_table_group_by_device(click_runner: CliRunner) -> None:
+ """Test anta nrfu, catalog is given via env."""
+ result = click_runner.invoke(anta, ["nrfu", "table", "--group-by", "device"])
+ assert result.exit_code == ExitCode.OK
+ assert "Summary per device" in result.output
+
+
+def test_anta_nrfu_table_group_by_test(click_runner: CliRunner) -> None:
+ """Test anta nrfu, catalog is given via env."""
+ result = click_runner.invoke(anta, ["nrfu", "table", "--group-by", "test"])
+ assert result.exit_code == ExitCode.OK
+ assert "Summary per test" 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"])
@@ -66,7 +80,7 @@ def test_anta_nrfu_json(click_runner: CliRunner) -> None:
result = click_runner.invoke(anta, ["nrfu", "json"])
assert result.exit_code == ExitCode.OK
assert "JSON results" in result.output
- match = re.search(r"\[\n {[\s\S]+ }\n\]", result.output)
+ match = re.search(r"\[\n {2}{[\s\S]+ {2}}\n\]", result.output)
assert match is not None
result_list = json.loads(match.group())
for res in result_list:
diff --git a/tests/units/cli/test__init__.py b/tests/units/cli/test__init__.py
index 0701083..6e32664 100644
--- a/tests/units/cli/test__init__.py
+++ b/tests/units/cli/test__init__.py
@@ -1,64 +1,55 @@
# 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__."""
+"""Tests for anta.cli._main."""
from __future__ import annotations
-from typing import TYPE_CHECKING
+import sys
+from importlib import reload
+from typing import TYPE_CHECKING, Any
from unittest.mock import patch
import pytest
-from anta.cli import anta, cli
-from anta.cli.utils import ExitCode
+import anta.cli
if TYPE_CHECKING:
- from click.testing import CliRunner
+ from types import ModuleType
+builtins_import = __import__
-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
+# Tried to achieve this with mock
+# http://materials-scientist.com/blog/2021/02/11/mocking-failing-module-import-python/
+def import_mock(name: str, *args: Any) -> ModuleType: # noqa: ANN401
+ """Mock."""
+ if name == "click":
+ msg = "No module named 'click'"
+ raise ModuleNotFoundError(msg)
+ return builtins_import(name, *args)
-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_cli_error_missing(capsys: pytest.CaptureFixture[Any]) -> None:
+ """Test ANTA errors out when anta[cli] was not installed."""
+ with patch.dict(sys.modules) as sys_modules, patch("builtins.__import__", import_mock):
+ del sys_modules["anta.cli._main"]
+ reload(anta.cli)
-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
+ with pytest.raises(SystemExit) as e_info:
+ anta.cli.cli()
+ captured = capsys.readouterr()
+ assert "The ANTA command line client could not run because the required dependencies were not installed." in captured.out
+ assert "Make sure you've installed everything with: pip install 'anta[cli]'" in captured.out
+ assert e_info.value.code == 1
-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
+ # setting ANTA_DEBUG
+ with pytest.raises(SystemExit) as e_info, patch("anta.cli.__DEBUG__", new=True):
+ anta.cli.cli()
-
-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
-
-
-def test_uncaught_failure_anta(caplog: pytest.LogCaptureFixture) -> None:
- """Test uncaught failure when running ANTA cli."""
- with (
- pytest.raises(SystemExit) as e_info,
- patch("anta.cli.anta", side_effect=ZeroDivisionError()),
- ):
- cli()
- assert "CRITICAL" in caplog.text
- assert "Uncaught Exception when running ANTA CLI" in caplog.text
- assert e_info.value.code == 1
+ captured = capsys.readouterr()
+ assert "The ANTA command line client could not run because the required dependencies were not installed." in captured.out
+ assert "Make sure you've installed everything with: pip install 'anta[cli]'" in captured.out
+ assert "The caught exception was:" in captured.out
+ assert e_info.value.code == 1
diff --git a/tests/units/cli/test_main.py b/tests/units/cli/test_main.py
new file mode 100644
index 0000000..31a5e78
--- /dev/null
+++ b/tests/units/cli/test_main.py
@@ -0,0 +1,64 @@
+# 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._main."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+from unittest.mock import patch
+
+import pytest
+
+from anta.cli._main import anta, cli
+from anta.cli.utils import ExitCode
+
+if TYPE_CHECKING:
+ from click.testing import CliRunner
+
+
+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
+
+
+def test_uncaught_failure_anta(caplog: pytest.LogCaptureFixture) -> None:
+ """Test uncaught failure when running ANTA cli."""
+ with (
+ pytest.raises(SystemExit) as e_info,
+ patch("anta.cli._main.anta", side_effect=ZeroDivisionError()),
+ ):
+ cli()
+ assert "CRITICAL" in caplog.text
+ assert "Uncaught Exception when running ANTA CLI" in caplog.text
+ assert e_info.value.code == 1