summaryrefslogtreecommitdiffstats
path: root/anta/cli/utils.py
diff options
context:
space:
mode:
Diffstat (limited to 'anta/cli/utils.py')
-rw-r--r--anta/cli/utils.py125
1 files changed, 75 insertions, 50 deletions
diff --git a/anta/cli/utils.py b/anta/cli/utils.py
index 97a0862..53a9719 100644
--- a/anta/cli/utils.py
+++ b/anta/cli/utils.py
@@ -1,16 +1,15 @@
# 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.
-"""
-Utils functions to use with anta.cli module.
-"""
+"""Utils functions to use with anta.cli module."""
+
from __future__ import annotations
import enum
import functools
import logging
from pathlib import Path
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, Callable
import click
from pydantic import ValidationError
@@ -18,7 +17,7 @@ from yaml import YAMLError
from anta.catalog import AntaCatalog
from anta.inventory import AntaInventory
-from anta.inventory.exceptions import InventoryIncorrectSchema, InventoryRootKeyError
+from anta.inventory.exceptions import InventoryIncorrectSchemaError, InventoryRootKeyError
if TYPE_CHECKING:
from click import Option
@@ -27,10 +26,7 @@ logger = logging.getLogger(__name__)
class ExitCode(enum.IntEnum):
- """
- Encodes the valid exit codes by anta
- inspired from pytest
- """
+ """Encodes the valid exit codes by anta inspired from pytest."""
# Tests passed.
OK = 0
@@ -44,19 +40,18 @@ class ExitCode(enum.IntEnum):
TESTS_FAILED = 4
-def parse_tags(ctx: click.Context, param: Option, value: str) -> list[str] | None:
+def parse_tags(ctx: click.Context, param: Option, value: str | None) -> set[str] | None:
# pylint: disable=unused-argument
- """
- Click option callback to parse an ANTA inventory tags
- """
+ # ruff: noqa: ARG001
+ """Click option callback to parse an ANTA inventory tags."""
if value is not None:
- return value.split(",") if "," in value else [value]
+ return set(value.split(",")) if "," in value else {value}
return None
def exit_with_code(ctx: click.Context) -> None:
- """
- Exit the Click application with an exit code.
+ """Exit the Click application with an exit code.
+
This function determines the global test status to be either `unset`, `skipped`, `success` or `error`
from the `ResultManger` instance.
If flag `ignore_error` is set, the `error` status will be ignored in all the tests.
@@ -64,10 +59,12 @@ def exit_with_code(ctx: click.Context) -> None:
Exit the application with the following exit code:
* 0 if `ignore_status` is `True` or global test status is `unset`, `skipped` or `success`
* 1 if status is `failure`
- * 2 if status is `error`
+ * 2 if status is `error`.
Args:
+ ----
ctx: Click Context
+
"""
if ctx.obj.get("ignore_status"):
ctx.exit(ExitCode.OK)
@@ -83,18 +80,19 @@ def exit_with_code(ctx: click.Context) -> None:
ctx.exit(ExitCode.TESTS_ERROR)
logger.error("Please gather logs and open an issue on Github.")
- raise ValueError(f"Unknown status returned by the ResultManager: {status}. Please gather logs and open an issue on Github.")
+ msg = f"Unknown status returned by the ResultManager: {status}. Please gather logs and open an issue on Github."
+ raise ValueError(msg)
class AliasedGroup(click.Group):
- """
- Implements a subclass of Group that accepts a prefix for a command.
+ """Implements a subclass of Group that accepts a prefix for a command.
+
If there were a command called push, it would accept pus as an alias (so long as it was unique)
- From Click documentation
+ From Click documentation.
"""
def get_command(self, ctx: click.Context, cmd_name: str) -> Any:
- """Todo: document code"""
+ """Todo: document code."""
rv = click.Group.get_command(self, ctx, cmd_name)
if rv is not None:
return rv
@@ -107,15 +105,16 @@ class AliasedGroup(click.Group):
return None
def resolve_command(self, ctx: click.Context, args: Any) -> Any:
- """Todo: document code"""
+ """Todo: document code."""
# always return the full command name
_, cmd, args = super().resolve_command(ctx, args)
- return cmd.name, cmd, args # type: ignore
+ if not cmd:
+ return None, None, None
+ return cmd.name, cmd, args
-# TODO: check code of click.pass_context that raise mypy errors for types and adapt this decorator
-def inventory_options(f: Any) -> Any:
- """Click common options when requiring an inventory to interact with devices"""
+def inventory_options(f: Callable[..., Any]) -> Callable[..., Any]:
+ """Click common options when requiring an inventory to interact with devices."""
@click.option(
"--username",
@@ -159,26 +158,34 @@ def inventory_options(f: Any) -> Any:
)
@click.option(
"--timeout",
- help="Global connection timeout",
- default=30,
+ help="Global API timeout. This value will be used for all devices.",
+ default=30.0,
show_envvar=True,
envvar="ANTA_TIMEOUT",
show_default=True,
)
@click.option(
"--insecure",
- help="Disable SSH Host Key validation",
+ help="Disable SSH Host Key validation.",
default=False,
show_envvar=True,
envvar="ANTA_INSECURE",
is_flag=True,
show_default=True,
)
- @click.option("--disable-cache", help="Disable cache globally", show_envvar=True, envvar="ANTA_DISABLE_CACHE", show_default=True, is_flag=True, default=False)
+ @click.option(
+ "--disable-cache",
+ help="Disable cache globally.",
+ show_envvar=True,
+ envvar="ANTA_DISABLE_CACHE",
+ show_default=True,
+ is_flag=True,
+ default=False,
+ )
@click.option(
"--inventory",
"-i",
- help="Path to the inventory YAML file",
+ help="Path to the inventory YAML file.",
envvar="ANTA_INVENTORY",
show_envvar=True,
required=True,
@@ -186,8 +193,7 @@ def inventory_options(f: Any) -> Any:
)
@click.option(
"--tags",
- "-t",
- help="List of tags using comma as separator: tag1,tag2,tag3",
+ help="List of tags using comma as separator: tag1,tag2,tag3.",
show_envvar=True,
envvar="ANTA_TAGS",
type=str,
@@ -200,13 +206,13 @@ def inventory_options(f: Any) -> Any:
ctx: click.Context,
*args: tuple[Any],
inventory: Path,
- tags: list[str] | None,
+ tags: set[str] | None,
username: str,
password: str | None,
enable_password: str | None,
enable: bool,
prompt: bool,
- timeout: int,
+ timeout: float,
insecure: bool,
disable_cache: bool,
**kwargs: dict[str, Any],
@@ -218,17 +224,25 @@ def inventory_options(f: Any) -> Any:
if prompt:
# User asked for a password prompt
if password is None:
- password = click.prompt("Please enter a password to connect to EOS", type=str, hide_input=True, confirmation_prompt=True)
- if enable:
- if enable_password is None:
- if click.confirm("Is a password required to enter EOS privileged EXEC mode?"):
- enable_password = click.prompt(
- "Please enter a password to enter EOS privileged EXEC mode", type=str, hide_input=True, confirmation_prompt=True
- )
+ password = click.prompt(
+ "Please enter a password to connect to EOS",
+ type=str,
+ hide_input=True,
+ confirmation_prompt=True,
+ )
+ if enable and enable_password is None and click.confirm("Is a password required to enter EOS privileged EXEC mode?"):
+ enable_password = click.prompt(
+ "Please enter a password to enter EOS privileged EXEC mode",
+ type=str,
+ hide_input=True,
+ confirmation_prompt=True,
+ )
if password is None:
- raise click.BadParameter("EOS password needs to be provided by using either the '--password' option or the '--prompt' option.")
+ msg = "EOS password needs to be provided by using either the '--password' option or the '--prompt' option."
+ raise click.BadParameter(msg)
if not enable and enable_password:
- raise click.BadParameter("Providing a password to access EOS Privileged EXEC mode requires '--enable' option.")
+ msg = "Providing a password to access EOS Privileged EXEC mode requires '--enable' option."
+ raise click.BadParameter(msg)
try:
i = AntaInventory.parse(
filename=inventory,
@@ -240,15 +254,15 @@ def inventory_options(f: Any) -> Any:
insecure=insecure,
disable_cache=disable_cache,
)
- except (ValidationError, TypeError, ValueError, YAMLError, OSError, InventoryIncorrectSchema, InventoryRootKeyError):
+ except (ValidationError, TypeError, ValueError, YAMLError, OSError, InventoryIncorrectSchemaError, InventoryRootKeyError):
ctx.exit(ExitCode.USAGE_ERROR)
return f(*args, inventory=i, tags=tags, **kwargs)
return wrapper
-def catalog_options(f: Any) -> Any:
- """Click common options when requiring a test catalog to execute ANTA tests"""
+def catalog_options(f: Callable[..., Any]) -> Callable[..., Any]:
+ """Click common options when requiring a test catalog to execute ANTA tests."""
@click.option(
"--catalog",
@@ -256,12 +270,23 @@ def catalog_options(f: Any) -> Any:
envvar="ANTA_CATALOG",
show_envvar=True,
help="Path to the test catalog YAML file",
- type=click.Path(file_okay=True, dir_okay=False, exists=True, readable=True, path_type=Path),
+ type=click.Path(
+ file_okay=True,
+ dir_okay=False,
+ exists=True,
+ readable=True,
+ path_type=Path,
+ ),
required=True,
)
@click.pass_context
@functools.wraps(f)
- def wrapper(ctx: click.Context, *args: tuple[Any], catalog: Path, **kwargs: dict[str, Any]) -> Any:
+ def wrapper(
+ ctx: click.Context,
+ *args: tuple[Any],
+ catalog: Path,
+ **kwargs: dict[str, Any],
+ ) -> Any:
# If help is invoke somewhere, do not parse catalog
if ctx.obj.get("_anta_help"):
return f(*args, catalog=None, **kwargs)
generated by cgit v1.2.3 (git 2.39.1) at 2024-06-29 11:24:01 +0000