feat: add '--log-file' option to send logs to a file and introduced '--prompt' CLI option to prompt for passwords

anta/__init__.py

@@ -15,7 +15,7 @@
 ]
 __copyright__ = "Copyright 2022, Arista EMEA AS"
 
-# Global ANTA debug environment variable. Can be set using the cli 'anta --debug'.
+# Global ANTA debug mode environment variable
 __DEBUG__ = bool(os.environ.get("ANTA_DEBUG", "").lower() == "true")
 
 

anta/cli/__init__.py

@@ -6,7 +6,7 @@
 
 import logging
 import pathlib
-from typing import Any, Callable, Dict, List, Tuple
+from typing import Any, Callable, Dict, List, Literal, Tuple
 
 import click
 
@@ -15,7 +15,9 @@
 from anta.cli.exec import commands as exec_commands
 from anta.cli.get import commands as get_commands
 from anta.cli.nrfu import commands as check_commands
-from anta.cli.utils import IgnoreRequiredWithHelp, parse_catalog, parse_inventory, prompt_enable_password, prompt_password, setup_logging
+from anta.cli.utils import IgnoreRequiredWithHelp, parse_catalog, parse_inventory
+from anta.loader import setup_logging
+from anta.result_manager import ResultManager
 from anta.result_manager.models import TestResult
 
 
@@ -25,54 +27,67 @@
 @click.version_option(__version__)
 @click.option(
     "--username",
-    show_envvar=True,
     help="Username to connect to EOS",
+    show_envvar=True,
     required=True,
 )
-@click.option("--password", show_envvar=True, help="Password to connect to EOS", callback=prompt_password)
+@click.option("--password", help="Password to connect to EOS that must be provided. It can be prompted using '--prompt' option.", show_envvar=True)
+@click.option(
+    "--enable-password",
+    help="Password to access EOS Privileged EXEC mode. It can be prompted using '--prompt' option. Requires '--enable' option.",
+    show_envvar=True,
+)
 @click.option(
     "--enable",
+    help="Some commands may require EOS Privileged EXEC mode. This option tries to access this mode before sending a command to the device.",
+    default=False,
     show_envvar=True,
     is_flag=True,
-    default=False,
-    help="Some commands may require EOS Privileged EXEC mode. This option tries to access this mode before sending a command to the device.",
     show_default=True,
 )
 @click.option(
-    "--enable-password",
-    show_envvar=True,
-    help="If a password is required to access EOS Privileged EXEC mode, it must be provided. --enable must be set.",
-    callback=prompt_enable_password,
+    "--prompt",
+    "-P",
+    help="Prompt for passwords if they are not provided.",
+    default=False,
+    is_flag=True,
+    show_default=True,
 )
 @click.option(
     "--timeout",
-    show_envvar=True,
-    default=5,
     help="Global connection timeout",
+    default=30,
+    show_envvar=True,
     show_default=True,
 )
 @click.option(
     "--insecure",
+    help="Disable SSH Host Key validation",
+    default=False,
     show_envvar=True,
     is_flag=True,
-    default=False,
-    help="Disable SSH Host Key validation",
     show_default=True,
 )
 @click.option(
     "--inventory",
     "-i",
+    help="Path to the inventory YAML file",
     show_envvar=True,
     required=True,
-    help="Path to the inventory YAML file",
     type=click.Path(file_okay=True, dir_okay=False, exists=True, readable=True, path_type=pathlib.Path),
 )
+@click.option(
+    "--log-file",
+    help="Send the logs to a file. If logging level is DEBUG, only INFO or higher will be sent to stdout.",
+    show_envvar=True,
+    type=click.Path(file_okay=True, dir_okay=False, writable=True, path_type=pathlib.Path),
+)
 @click.option(
     "--log-level",
     "--log",
-    show_envvar=True,
     help="ANTA logging level",
     default=logging.getLevelName(logging.INFO),
+    show_envvar=True,
     show_default=True,
     type=click.Choice(
         [
@@ -84,17 +99,36 @@
         ],
         case_sensitive=False,
     ),
-    callback=setup_logging,
 )
-@click.option("--ignore-status", show_envvar=True, is_flag=True, default=False, help="Always exit with success")
-@click.option("--ignore-error", show_envvar=True, is_flag=True, default=False, help="Only report failures and not errors")
-def anta(ctx: click.Context, inventory: pathlib.Path, ignore_status: bool, ignore_error: bool, **kwargs: Any) -> None:
+@click.option("--ignore-status", help="Always exit with success", show_envvar=True, is_flag=True, default=False)
+@click.option("--ignore-error", help="Only report failures and not errors", show_envvar=True, is_flag=True, default=False)
+def anta(
+    ctx: click.Context, inventory: pathlib.Path, log_level: Literal["CRITICAL", "ERROR", "WARNING", "INFO", "DEBUG"], log_file: pathlib.Path, **kwargs: Any
+) -> None:
     # pylint: disable=unused-argument
     """Arista Network Test Automation (ANTA) CLI"""
+    setup_logging(log_level, log_file)
+
+    if not ctx.obj.get("_anta_help"):
+        if ctx.params.get("prompt"):
+            # User asked for a password prompt
+            if ctx.params.get("password") is None:
+                ctx.params["password"] = click.prompt("Please enter a password to connect to EOS", type=str, hide_input=True, confirmation_prompt=True)
+            if ctx.params.get("enable"):
+                if ctx.params.get("enable_password") is None:
+                    if click.confirm("Is a password required to enter EOS privileged EXEC mode?"):
+                        ctx.params["enable_password"] = click.prompt(
+                            "Please enter a password to enter EOS privileged EXEC mode", type=str, hide_input=True, confirmation_prompt=True
+                        )
+        if ctx.params.get("password") is None:
+            raise click.BadParameter(
+                f"EOS password needs to be provided by using either the '{anta.params[2].opts[0]}' option or the '{anta.params[5].opts[0]}' option."
+            )
+        if not ctx.params.get("enable") and ctx.params.get("enable_password"):
+            raise click.BadParameter(f"Providing a password to access EOS Privileged EXEC mode requires '{anta.params[4].opts[0]}' option.")
+
     ctx.ensure_object(dict)
     ctx.obj["inventory"] = parse_inventory(ctx, inventory)
-    ctx.obj["ignore_status"] = ignore_status
-    ctx.obj["ignore_error"] = ignore_error
 
 
 @anta.group(cls=IgnoreRequiredWithHelp)
@@ -111,6 +145,7 @@ def anta(ctx: click.Context, inventory: pathlib.Path, ignore_status: bool, ignor
 def nrfu(ctx: click.Context, catalog: List[Tuple[Callable[..., TestResult], Dict[Any, Any]]]) -> None:
     """Run NRFU against inventory devices"""
     ctx.obj["catalog"] = catalog
+    ctx.obj["result_manager"] = ResultManager()
 
 
 @anta.group("exec")

anta/cli/nrfu/commands.py

@@ -7,14 +7,12 @@
 import asyncio
 import logging
 import pathlib
-import sys
 from typing import List, Optional
 
 import click
 
-from anta.cli.utils import parse_tags, return_code
+from anta.cli.utils import exit_with_code, parse_tags
 from anta.models import AntaTest
-from anta.result_manager import ResultManager
 from anta.runner import main
 
 from .utils import anta_progress_bar, print_jinja, print_json, print_settings, print_table, print_text
@@ -33,16 +31,10 @@
 def table(ctx: click.Context, tags: Optional[List[str]], device: Optional[str], test: Optional[str], group_by: str) -> None:
     """ANTA command to check network states with table result"""
     print_settings(ctx)
-    results = ResultManager()
     with anta_progress_bar() as AntaTest.progress:
-        asyncio.run(main(results, ctx.obj["inventory"], ctx.obj["catalog"], tags=tags))
-
-    print_table(results=results, device=device, group_by=group_by, test=test)
-
-    # TODO make a util method to avoid repeating the same three line
-    ignore_status = ctx.obj["ignore_status"]
-    ignore_error = ctx.obj["ignore_error"]
-    sys.exit(return_code(results, ignore_error, ignore_status))
+        asyncio.run(main(ctx.obj["result_manager"], ctx.obj["inventory"], ctx.obj["catalog"], tags=tags))
+    print_table(results=ctx.obj["result_manager"], device=device, group_by=group_by, test=test)
+    exit_with_code(ctx)
 
 
 @click.command()
@@ -59,14 +51,10 @@ def table(ctx: click.Context, tags: Optional[List[str]], device: Optional[str],
 def json(ctx: click.Context, tags: Optional[List[str]], output: Optional[pathlib.Path]) -> None:
     """ANTA command to check network state with JSON result"""
     print_settings(ctx)
-    results = ResultManager()
     with anta_progress_bar() as AntaTest.progress:
-        asyncio.run(main(results, ctx.obj["inventory"], ctx.obj["catalog"], tags=tags))
-    print_json(results=results, output=output)
-
-    ignore_status = ctx.obj["ignore_status"]
-    ignore_error = ctx.obj["ignore_error"]
-    sys.exit(return_code(results, ignore_error, ignore_status))
+        asyncio.run(main(ctx.obj["result_manager"], ctx.obj["inventory"], ctx.obj["catalog"], tags=tags))
+    print_json(results=ctx.obj["result_manager"], output=output)
+    exit_with_code(ctx)
 
 
 @click.command()
@@ -77,14 +65,10 @@ def json(ctx: click.Context, tags: Optional[List[str]], output: Optional[pathlib
 def text(ctx: click.Context, tags: Optional[List[str]], search: Optional[str], skip_error: bool) -> None:
     """ANTA command to check network states with text result"""
     print_settings(ctx)
-    results = ResultManager()
     with anta_progress_bar() as AntaTest.progress:
-        asyncio.run(main(results, ctx.obj["inventory"], ctx.obj["catalog"], tags=tags))
-    print_text(results=results, search=search, skip_error=skip_error)
-
-    ignore_status = ctx.obj["ignore_status"]
-    ignore_error = ctx.obj["ignore_error"]
-    sys.exit(return_code(results, ignore_error, ignore_status))
+        asyncio.run(main(ctx.obj["result_manager"], ctx.obj["inventory"], ctx.obj["catalog"], tags=tags))
+    print_text(results=ctx.obj["result_manager"], search=search, skip_error=skip_error)
+    exit_with_code(ctx)
 
 
 @click.command()
@@ -109,11 +93,7 @@ def text(ctx: click.Context, tags: Optional[List[str]], search: Optional[str], s
 def tpl_report(ctx: click.Context, tags: Optional[List[str]], template: pathlib.Path, output: Optional[pathlib.Path]) -> None:
     """ANTA command to check network state with templated report"""
     print_settings(ctx, template, output)
-    results = ResultManager()
     with anta_progress_bar() as AntaTest.progress:
-        asyncio.run(main(results, ctx.obj["inventory"], ctx.obj["catalog"], tags=tags))
-    print_jinja(results=results, template=template, output=output)
-
-    ignore_status = ctx.obj["ignore_status"]
-    ignore_error = ctx.obj["ignore_error"]
-    sys.exit(return_code(results, ignore_error, ignore_status))
+        asyncio.run(main(ctx.obj["result_manager"], ctx.obj["inventory"], ctx.obj["catalog"], tags=tags))
+    print_jinja(results=ctx.obj["result_manager"], template=template, output=output)
+    exit_with_code(ctx)

anta/cli/utils.py

@@ -23,8 +23,6 @@
 if TYPE_CHECKING:
     from click import Option
 
-    from anta.result_manager import ResultManager
-
 
 class ExitCode(enum.IntEnum):
     """
@@ -79,35 +77,6 @@ def parse_tags(ctx: click.Context, param: Option, value: str) -> Optional[List[s
     return None
 
 
-def prompt_password(ctx: click.Context, param: Option, value: Optional[str]) -> Optional[str]:
-    # pylint: disable=unused-argument
-    """
-    Click option callback to ensure that enable is True when the option is set
-    """
-    if ctx.obj.get("_anta_help"):
-        # Currently looking for help for a subcommand so no
-        # need to prompt the password
-        return None
-    if value is None:
-        return click.prompt("Please enter a password to connect to EOS", type=str, hide_input=True, confirmation_prompt=True)
-    return value
-
-
-def prompt_enable_password(ctx: click.Context, param: Option, value: Optional[str]) -> Optional[str]:
-    """
-    Click option callback to ensure that enable is True when the option is set
-    """
-    if ctx.obj.get("_anta_help"):
-        # Currently looking for help for a subcommand so no
-        # need to prompt the password
-        return None
-    if value is not None and ctx.params.get("enable") is not True:
-        raise click.BadParameter(f"'{param.opts[0]}' requires '--enable'")
-    if value is None and ctx.params.get("enable") is True:
-        return click.prompt("Please enter a password to enter EOS privileged EXEC mode", type=str, hide_input=True, confirmation_prompt=True)
-    return value
-
-
 def parse_catalog(ctx: click.Context, param: Option, value: str) -> List[Tuple[Callable[..., TestResult], Dict[Any, Any]]]:
     # pylint: disable=unused-argument
     """
@@ -130,47 +99,34 @@ def parse_catalog(ctx: click.Context, param: Option, value: str) -> List[Tuple[C
     return anta.loader.parse_catalog(data)
 
 
-def setup_logging(ctx: click.Context, param: Option, value: str) -> str:
-    # pylint: disable=unused-argument
+def exit_with_code(ctx: click.Context) -> None:
     """
-    Click option callback to set ANTA logging level
-    """
-    try:
-        anta.loader.setup_logging(value)
-    except Exception as e:  # pylint: disable=broad-exception-caught
-        message = f"Unable to set ANTA logging level '{value}'"
-        anta_log_exception(e, message, logger)
-        ctx.fail(message)
-
-    return value
+    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.
+    If flag `ignore_status` is set, the exit code will always be 0.
+    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`
 
-
-def return_code(result_manager: ResultManager, ignore_error: bool, ignore_status: bool) -> int:
-    """
     Args:
-        result_manager (ResultManager)
-        ignore_error (bool): Ignore error status
-        ignore_status (bool): Ignore status completely and always return 0
-
-    Returns:
-        exit_code (int):
-          * 0 if ignore_status is True or status is in ["unset", "skipped", "success"]
-          * 1 if status is "failure"
-          * 2 if status is "error"
+        ctx: Click Context
     """
 
-    if ignore_status:
-        return 0
+    if ctx.params.get("ignore_status"):
+        ctx.exit(0)
 
     # If ignore_error is True then status can never be "error"
-    status = result_manager.get_status(ignore_error=ignore_error)
+    status = ctx.obj["result_manager"].get_status(ignore_error=bool(ctx.params.get("ignore_error")))
 
     if status in {"unset", "skipped", "success"}:
-        return ExitCode.OK
+        ctx.exit(ExitCode.OK)
     if status == "failure":
-        return ExitCode.TESTS_FAILED
+        ctx.exit(ExitCode.TESTS_FAILED)
     if status == "error":
-        return ExitCode.TESTS_ERROR
+        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.")