fix(anta.cli): Use configured anta inventory as output by default in …

…get from-ansible (#469)

* fix(anta.cli): Use configured anta inventory as output by default in get from-ansible

* ci: update unit test for get from-ansible

* feat: Ask prompt if file is not empty

* fix: Ask prompt if file is not empty

* ci: Add unit tests

* doc: Update doc accordingly

* ci: Update unit tests

* ci: Update unit tests stage 2

* fix: Update CLI logic as per review

* doc: Update CLI logic as per review

anta/cli/get/commands.py

@@ -9,6 +9,7 @@
 from __future__ import annotations
 
 import asyncio
+import io
 import json
 import logging
 import os
@@ -20,6 +21,7 @@
 from cvprac.cvp_client import CvpClient
 from cvprac.cvp_client_errors import CvpApiError
 from rich.pretty import pretty_repr
+from rich.prompt import Confirm
 
 from anta.cli.console import console
 from anta.cli.utils import ExitCode, parse_tags
@@ -86,16 +88,61 @@ def from_cvp(inventory_directory: str, cvp_ip: str, cvp_username: str, cvp_passw
 @click.option(
     "--output",
     "-o",
-    default="inventory-ansible.yml",
-    help="Path to save inventory file",
+    required=False,
+    help="Path to save inventory file. If not configured, use anta inventory file",
     type=click.Path(file_okay=True, dir_okay=False, exists=False, writable=True, path_type=Path),
 )
-def from_ansible(ctx: click.Context, output: Path, ansible_inventory: Path, ansible_group: str) -> None:
+@click.option(
+    "--overwrite",
+    help="Confirm script can overwrite existing inventory file",
+    default=False,
+    is_flag=True,
+    show_default=True,
+    required=False,
+    show_envvar=True,
+)
+def from_ansible(ctx: click.Context, output: Path, ansible_inventory: Path, ansible_group: str, overwrite: bool) -> None:
+    # pylint: disable=too-many-arguments
     """Build ANTA inventory from an ansible inventory YAML file"""
     logger.info(f"Building inventory from ansible file {ansible_inventory}")
 
+    try:
+        is_tty = os.isatty(sys.stdout.fileno())
+    except io.UnsupportedOperation:
+        is_tty = False
+
     # Create output directory
+    if output is None:
+        if ctx.obj.get("inventory_path") is not None:
+            output = ctx.obj.get("inventory_path")
+        else:
+            logger.error("Inventory output is not set. You should use either anta --inventory or anta get from-ansible --output")
+            sys.exit(ExitCode.USAGE_ERROR)
+
+    logger.debug(f"output: {output}\noverwrite: {overwrite}\nis tty: {is_tty}")
+
+    # Count number of lines in a file
+    anta_inventory_number_lines = 0
+    if output.exists():
+        with open(output, "r", encoding="utf-8") as f:
+            anta_inventory_number_lines = sum(1 for _ in f)
+
+    # File has content and it is not interactive TTY nor overwrite set to True --> execution stop
+    if anta_inventory_number_lines > 0 and not is_tty and not overwrite:
+        logger.critical("conversion aborted since destination file is not empty (not running in interactive TTY)")
+        sys.exit(ExitCode.USAGE_ERROR)
+
+    # File has content and it is in an interactive TTY --> Prompt user
+    if anta_inventory_number_lines > 0 and is_tty and not overwrite:
+        confirm_overwrite = Confirm.ask(f"Your destination file ({output}) is not empty, continue?")
+        try:
+            assert confirm_overwrite is True
+        except AssertionError:
+            logger.critical("conversion aborted by user because destination file is not empty")
+            sys.exit(ExitCode.USAGE_ERROR)
+
     output.parent.mkdir(parents=True, exist_ok=True)
+    logger.info(f"output anta inventory is: {output}")
     try:
         create_inventory_from_ansible(
             inventory=ansible_inventory,
@@ -105,6 +152,7 @@ def from_ansible(ctx: click.Context, output: Path, ansible_inventory: Path, ansi
     except ValueError as e:
         logger.error(str(e))
         ctx.exit(ExitCode.USAGE_ERROR)
+    ctx.exit(ExitCode.OK)
 
 
 @click.command()

docs/cli/inv-from-ansible.md

@@ -11,18 +11,20 @@ In large setups, it might be beneficial to construct your inventory based on you
 ### Command overview
 
 ```bash
-anta get from-ansible --help
+$ anta get from-ansible --help
 Usage: anta get from-ansible [OPTIONS]
 
   Build ANTA inventory from an ansible inventory YAML file
 
 Options:
-  -g, --ansible-group TEXT        Ansible group to filter
-  -i, --ansible-inventory FILENAME
-                                  Path to your ansible inventory file to read
-  -o, --output FILENAME           Path to save inventory file
-  -d, --inventory-directory PATH  Directory to save inventory file
-  --help                          Show this message and exit.
+  -g, --ansible-group TEXT      Ansible group to filter
+  -i, --ansible-inventory FILE  Path to your ansible inventory file to read
+  -o, --output FILE             Path to save inventory file. If not
+                                configured, use anta inventory file
+  --overwrite                   Confirm script can overwrite existing
+                                inventory file  [env var:
+                                ANTA_GET_FROM_ANSIBLE_OVERWRITE]
+  --help                        Show this message and exit.
 ```
 
 The output is an inventory where the name of the container is added as a tag for each host:
@@ -41,6 +43,11 @@ anta_inventory:
 !!! warning
     The current implementation only considers devices directly attached to a specific Ansible group and does not support inheritence when using the `--ansible-group` option.
 
+By default, if user does not provide `--output` file, anta will save output to configured anta inventory (`anta --inventory`). If the output file has content, anta will ask user to overwrite when running in interactive console. This mechanism can be controlled by triggers in case of CI usage: `--overwrite` to force anta to overwrite file. If not set, anta will exit
+
+
+### Command output
+
 `host` value is coming from the `ansible_host` key in your inventory while `name` is the name you defined for your host. Below is an ansible inventory example used to generate previous inventory:
 
 ```yaml

tests/units/cli/get/test_commands.py

@@ -7,6 +7,7 @@
 from __future__ import annotations
 
 import os
+import shutil
 from pathlib import Path
 from typing import TYPE_CHECKING, cast
 from unittest.mock import ANY, patch
@@ -16,7 +17,7 @@
 from cvprac.cvp_client_errors import CvpApiError
 
 from anta.cli import anta
-from anta.cli.get.commands import from_ansible, from_cvp
+from anta.cli.get.commands import from_cvp
 from tests.lib.utils import default_anta_env
 
 if TYPE_CHECKING:
@@ -131,9 +132,11 @@ def test_from_ansible(
         out_dir = Path() / output
     else:
         # Get inventory-directory default
-        default_dir: Path = cast(Path, from_ansible.params[2].default)
+        default_dir: Path = cast(Path, f"{tmp_path}/output.yml")
         out_dir = Path() / default_dir
 
+    cli_args.extend(["--output", str(out_dir)])
+
     if ansible_inventory is not None:
         ansible_inventory_path = DATA_DIR / ansible_inventory
         cli_args.extend(["--ansible-inventory", str(ansible_inventory_path)])
@@ -145,11 +148,78 @@ def test_from_ansible(
         print(cli_args)
         result = click_runner.invoke(anta, cli_args, env=env, auto_envvar_prefix="ANTA")
 
-        print(result)
+    print(f"Runner args: {cli_args}")
+    print(f"Runner result is: {result}")
+    print(caplog.records)
 
     assert result.exit_code == expected_exit
-    print(caplog.records)
     if expected_exit != 0:
-        assert len(caplog.records) == 2
+        assert len(caplog.records) in {2, 3}
     else:
         assert out_dir.exists()
+
+
+@pytest.mark.parametrize(
+    "ansible_inventory, ansible_group, output_option, expected_exit",
+    [
+        pytest.param("ansible_inventory.yml", None, None, 4, id="no group-no-overwrite"),
+        pytest.param("ansible_inventory.yml", None, "--overwrite", 0, id="no group-overwrite"),
+        pytest.param("ansible_inventory.yml", "ATD_LEAFS", "--overwrite", 0, id="group found"),
+        pytest.param("ansible_inventory.yml", "DUMMY", "--overwrite", 4, id="group not found"),
+        pytest.param("empty_ansible_inventory.yml", None, None, 4, id="empty inventory"),
+    ],
+)
+# pylint: disable-next=too-many-arguments
+def test_from_ansible_default_inventory(
+    tmp_path: Path,
+    caplog: LogCaptureFixture,
+    capsys: CaptureFixture[str],
+    click_runner: CliRunner,
+    ansible_inventory: Path,
+    ansible_group: str | None,
+    output_option: str | None,
+    expected_exit: int,
+) -> None:
+    """
+    Test `anta get from-ansible`
+    """
+
+    def custom_anta_env(tmp_path: Path) -> dict[str, str]:
+        """
+        Return a default_anta_environement which can be passed to a cliRunner.invoke method
+        """
+        return {
+            "ANTA_USERNAME": "anta",
+            "ANTA_PASSWORD": "formica",
+            "ANTA_INVENTORY": str(tmp_path / "test_inventory02.yml"),
+            "ANTA_CATALOG": str(Path(__file__).parent.parent / "data" / "test_catalog.yml"),
+        }
+
+    env = custom_anta_env(tmp_path)
+    shutil.copyfile(str(Path(__file__).parent.parent.parent.parent / "data" / "test_inventory.yml"), env["ANTA_INVENTORY"])
+
+    cli_args = ["get", "from-ansible"]
+
+    os.chdir(tmp_path)
+    if output_option is not None:
+        cli_args.extend([str(output_option)])
+
+    if ansible_inventory is not None:
+        ansible_inventory_path = DATA_DIR / ansible_inventory
+        cli_args.extend(["--ansible-inventory", str(ansible_inventory_path)])
+
+    if ansible_group is not None:
+        cli_args.extend(["--ansible-group", ansible_group])
+
+    with capsys.disabled():
+        print(cli_args)
+        result = click_runner.invoke(anta, cli_args, env=env, auto_envvar_prefix="ANTA")
+
+    print(f"Runner args: {cli_args}")
+    print(f"Runner result is: {result}")
+
+    assert result.exit_code == expected_exit
+    print(caplog.records)
+    if expected_exit != 0:
+        assert len(caplog.records) in {1, 2, 3}
+    # Path(env["ANTA_INVENTORY"]).unlink(missing_ok=True)