feat(anta): Add dry-run mode for NRFU

anta/cli/nrfu/__init__.py

@@ -99,6 +99,14 @@ def parse_args(self, ctx: click.Context, args: list[str]) -> list[str]:
     help="Group result by test or device.",
     required=False,
 )
+@click.option(
+    "--dry-run",
+    help="Run anta nrfu command but stop before starting to execute the tests. Considers all devices as connected.",
+    type=str,
+    show_envvar=True,
+    is_flag=True,
+    default=False,
+)
 # pylint: disable=too-many-arguments
 def nrfu(
     ctx: click.Context,
@@ -111,6 +119,7 @@ def nrfu(
     *,
     ignore_status: bool,
     ignore_error: bool,
+    dry_run: bool,
 ) -> None:
     """Run ANTA tests on selected inventory devices."""
     # If help is invoke somewhere, skip the command
@@ -124,7 +133,19 @@ def nrfu(
     ctx.obj["hide"] = set(hide) if hide else None
     print_settings(inventory, catalog)
     with anta_progress_bar() as AntaTest.progress:
-        asyncio.run(main(ctx.obj["result_manager"], inventory, catalog, tags=tags, devices=set(device) if device else None, tests=set(test) if test else None))
+        asyncio.run(
+            main(
+                ctx.obj["result_manager"],
+                inventory,
+                catalog,
+                tags=tags,
+                devices=set(device) if device else None,
+                tests=set(test) if test else None,
+                dry_run=dry_run,
+            )
+        )
+    if dry_run:
+        return
     # Invoke `anta nrfu table` if no command is passed
     if ctx.invoked_subcommand is None:
         ctx.invoke(commands.table)

anta/cli/nrfu/utils.py

@@ -38,7 +38,7 @@ def print_settings(
     catalog: AntaCatalog,
 ) -> None:
     """Print ANTA settings before running tests."""
-    message = f"Running ANTA tests:\n- {inventory}\n- Tests catalog contains {len(catalog.tests)} tests"
+    message = f"- {inventory}\n- Tests catalog contains {len(catalog.tests)} tests"
     console.print(Panel.fit(message, style="cyan", title="[green]Settings"))
     console.print()
 

anta/runner.py

@@ -92,12 +92,9 @@ async def setup_inventory(inventory: AntaInventory, tags: set[str] | None, devic
     # Filter the inventory based on the CLI provided tags and devices if any
     selected_inventory = inventory.get_inventory(tags=tags, devices=devices) if tags or devices else inventory
 
-    with Catchtime() as t:
-        logger.info("Connecting to devices...")
+    with Catchtime(logger=logger, message="Connecting to devices"):
         # Connect to the devices
         await selected_inventory.connect_inventory()
-    msg = f"Connecting to devices completed in {t.time}"
-    logger.info(msg)
 
     # Remove devices that are unreachable
     selected_inventory = selected_inventory.get_inventory(established_only=established_only)
@@ -157,7 +154,7 @@ async def prepare_tests(
     return device_to_tests
 
 
-async def main(  # noqa: PLR0913
+async def main(  # noqa: PLR0913,C901
     manager: ResultManager,
     inventory: AntaInventory,
     catalog: AntaCatalog,
@@ -166,6 +163,7 @@ async def main(  # noqa: PLR0913
     tags: set[str] | None = None,
     *,
     established_only: bool = True,
+    dry_run: bool = False,
 ) -> None:
     # pylint: disable=too-many-arguments
     """Run ANTA.
@@ -182,6 +180,7 @@ async def main(  # noqa: PLR0913
         tests: Tests to run against devices. None means all tests. These may come from the `--test / -t` CLI option in NRFU.
         tags: Tags to filter devices from the inventory. These may come from the `--tags` CLI option in NRFU.
         established_only: Include only established device(s).
+        dry_run: Build the list of coroutine to run and stop before test execution.
     """
     # Adjust the maximum number of open file descriptors for the ANTA process
     limits = adjust_rlimit_nofile()
@@ -190,19 +189,16 @@ async def main(  # noqa: PLR0913
         logger.info("The list of tests is empty, exiting")
         return
 
-    logger.info("Preparing ANTA NRFU Run...")
-    with Catchtime() as prepare_t:
+    with Catchtime(logger=logger, message="Preparing ANTA NRFU Run"):
         # Setup the inventory
-        selected_inventory = await setup_inventory(inventory, tags, devices, established_only=established_only)
+        selected_inventory = inventory if dry_run else await setup_inventory(inventory, tags, devices, established_only=established_only)
         if selected_inventory is None:
             return
 
-        with Catchtime() as t:
-            logger.info("Preparing the tests...")
+        with Catchtime(logger=logger, message="Preparing the tests"):
             selected_tests = await prepare_tests(selected_inventory, catalog, tests, tags)
             if selected_tests is None:
                 return
-        logger.info("Preparing the tests completed in %s.", t.time)
 
         run_info = (
             "--- ANTA NRFU Run Information ---\n"
@@ -238,17 +234,18 @@ async def main(  # noqa: PLR0913
                     )
                     anta_log_exception(e, message, logger)
 
-    logger.info("Preparing ANTA NRFU Run completed in %s", prepare_t.time)
+    if dry_run:
+        logger.info("Dry-run mode, exiting before running the tests.")
+        for coro in coros:
+            coro.close()
+        return
 
-    with Catchtime() as run_t:
-        if AntaTest.progress is not None:
-            AntaTest.nrfu_task = AntaTest.progress.add_task("Running NRFU Tests...", total=len(coros))
+    if AntaTest.progress is not None:
+        AntaTest.nrfu_task = AntaTest.progress.add_task("Running NRFU Tests...", total=len(coros))
 
-        logger.info("Running ANTA tests...")
+    with Catchtime(logger=logger, message="Running ANTA tests"):
         test_results = await asyncio.gather(*coros)
         for r in test_results:
             manager.add(r)
 
-    logger.info("Running ANTA tests completed in %s", run_t.time)
-
     log_cache_statistics(selected_inventory.devices)

anta/tools.py

@@ -12,6 +12,7 @@
 
 if TYPE_CHECKING:
     import sys
+    from logging import Logger
     from types import TracebackType
 
     if sys.version_info >= (3, 11):
@@ -270,12 +271,20 @@ class Catchtime:
     raw_time: float
     time: str
 
+    def __init__(self, logger: Logger | None = None, message: str | None = None) -> None:
+        self.logger = logger
+        self.message = message
+
     def __enter__(self) -> Self:
         """__enter__ method."""
         self.start = perf_counter()
+        if self.logger and self.message:
+            self.logger.info("%s ...", self.message)
         return self
 
     def __exit__(self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None) -> None:
         """__exit__ method."""
         self.raw_time = perf_counter() - self.start
         self.time = format_td(self.raw_time, 3)
+        if self.logger and self.message:
+            self.logger.info("%s completed in: %s.", self.message, self.time)

docs/cli/nrfu.md

@@ -173,7 +173,7 @@ The `--output` option allows you to choose the path where the final report will
 ```bash
 anta nrfu --tags LEAF tpl-report --template ./custom_template.j2
 ```
-[![anta nrfu json results](../imgs/anta-nrfu-tpl-report-output.png){ loading=lazy width="1600" }](../imgs/anta-nrfu-tpl-report-output.png)
+[![anta nrfu tpl_resultss](../imgs/anta-nrfu-tpl-report-output.png){ loading=lazy width="1600" }](../imgs/anta-nrfu-tpl-report-output.png)
 
 The template `./custom_template.j2` is a simple Jinja2 template:
 
@@ -200,3 +200,9 @@ cat nrfu-tpl-report.txt
 * VerifyMlagConfigSanity is [green]SUCCESS[/green] for DC1-LEAF1A
 * VerifyMlagReloadDelay is [green]SUCCESS[/green] for DC1-LEAF1A
 ```
+
+## Dry-run mode
+
+It is possible to run `anta nrfu --dry-run` to execute ANTA up to the point where it should communicate with the network to execute the tests. When using `--dry-run`, all inventory devices are assumed to be online. This can be useful to check how many tests would be run using the catalog and inventory.
+
+[![anta nrfu dry_run](../imgs/anta_nrfu___dry_run.svg){ loading=lazy width="1600" }](../imgs/anta_nrfu___dry_run.svg)