feat(anta.cli): anta nrfu enchancements: add --device and --test to filter tests and --hide (error|failure|success|skipped) to hide results.

anta/catalog.py

@@ -336,11 +336,20 @@ def from_list(data: ListAntaTestTuples) -> AntaCatalog:
             raise
         return AntaCatalog(tests)
 
-    def get_tests_by_tags(self, tags: list[str], *, strict: bool = False) -> list[AntaTestDefinition]:
+    def get_tests_by_tags(self, tags: set[str], *, strict: bool = False) -> list[AntaTestDefinition]:
         """Return all the tests that have matching tags in their input filters.
 
-        If strict=True, returns only tests that match all the tags provided as input.
+        If strict=True, return only tests that match all the tags provided as input.
         If strict=False, return all the tests that match at least one tag provided as input.
+
+        Args:
+        ----
+            tags: Tags of the tests to get.
+            strict: Specify if the returned tests must match all the tags provided.
+
+        Returns
+        -------
+            List of AntaTestDefinition that match the tags
         """
         result: list[AntaTestDefinition] = []
         for test in self.tests:
@@ -351,3 +360,16 @@ def get_tests_by_tags(self, tags: list[str], *, strict: bool = False) -> list[An
                 elif any(t in tags for t in f):
                     result.append(test)
         return result
+
+    def get_tests_by_names(self, names: set[str]) -> list[AntaTestDefinition]:
+        """Return all the tests that have matching a list of tests names.
+
+        Args:
+        ----
+            names: Names of the tests to get.
+
+        Returns
+        -------
+            List of AntaTestDefinition that match the names
+        """
+        return [test for test in self.tests if test.test.name in names]

anta/cli/debug/commands.py

@@ -35,6 +35,7 @@ def run_cmd(
     version: Literal["1", "latest"],
     revision: int,
 ) -> None:
+    # pylint: disable=too-many-arguments
     """Run arbitrary command to an ANTA device."""
     console.print(f"Run command [green]{command}[/green] on [red]{device.name}[/red]")
     # I do not assume the following line, but click make me do it

anta/cli/debug/utils.py

@@ -44,7 +44,7 @@ def wrapper(
         ctx: click.Context,
         *args: tuple[Any],
         inventory: AntaInventory,
-        tags: list[str] | None,
+        tags: set[str] | None,
         device: str,
         **kwargs: Any,
     ) -> Any:

anta/cli/exec/commands.py

@@ -27,7 +27,7 @@
 
 @click.command
 @inventory_options
-def clear_counters(inventory: AntaInventory, tags: list[str] | None) -> None:
+def clear_counters(inventory: AntaInventory, tags: set[str] | None) -> None:
     """Clear counter statistics on EOS devices."""
     asyncio.run(clear_counters_utils(inventory, tags=tags))
 
@@ -51,7 +51,7 @@ def clear_counters(inventory: AntaInventory, tags: list[str] | None) -> None:
     default=f"anta_snapshot_{datetime.now(tz=timezone.utc).astimezone().strftime('%Y-%m-%d_%H_%M_%S')}",
     show_default=True,
 )
-def snapshot(inventory: AntaInventory, tags: list[str] | None, commands_list: Path, output: Path) -> None:
+def snapshot(inventory: AntaInventory, tags: set[str] | None, commands_list: Path, output: Path) -> None:
     """Collect commands output from devices in inventory."""
     console.print(f"Collecting data for {commands_list}")
     console.print(f"Output directory is {output}")
@@ -91,7 +91,7 @@ def snapshot(inventory: AntaInventory, tags: list[str] | None, commands_list: Pa
 )
 def collect_tech_support(
     inventory: AntaInventory,
-    tags: list[str] | None,
+    tags: set[str] | None,
     output: Path,
     latest: int | None,
     *,

anta/cli/exec/utils.py

@@ -29,7 +29,7 @@
 logger = logging.getLogger(__name__)
 
 
-async def clear_counters_utils(anta_inventory: AntaInventory, tags: list[str] | None = None) -> None:
+async def clear_counters_utils(anta_inventory: AntaInventory, tags: set[str] | None = None) -> None:
     """Clear counters."""
 
     async def clear(dev: AntaDevice) -> None:
@@ -44,7 +44,7 @@ async def clear(dev: AntaDevice) -> None:
 
     logger.info("Connecting to devices...")
     await anta_inventory.connect_inventory()
-    devices = anta_inventory.get_inventory(established_only=True, tags=tags).values()
+    devices = anta_inventory.get_inventory(established_only=True, tags=tags).devices
     logger.info("Clearing counters on remote devices...")
     await asyncio.gather(*(clear(device) for device in devices))
 
@@ -53,7 +53,7 @@ async def collect_commands(
     inv: AntaInventory,
     commands: dict[str, str],
     root_dir: Path,
-    tags: list[str] | None = None,
+    tags: set[str] | None = None,
 ) -> None:
     """Collect EOS commands."""
 
@@ -78,7 +78,7 @@ async def collect(dev: AntaDevice, command: str, outformat: Literal["json", "tex
 
     logger.info("Connecting to devices...")
     await inv.connect_inventory()
-    devices = inv.get_inventory(established_only=True, tags=tags).values()
+    devices = inv.get_inventory(established_only=True, tags=tags).devices
     logger.info("Collecting commands from remote devices")
     coros = []
     if "json_format" in commands:
@@ -91,7 +91,7 @@ async def collect(dev: AntaDevice, command: str, outformat: Literal["json", "tex
             logger.error("Error when collecting commands: %s", str(r))
 
 
-async def collect_scheduled_show_tech(inv: AntaInventory, root_dir: Path, *, configure: bool, tags: list[str] | None = None, latest: int | None = None) -> None:
+async def collect_scheduled_show_tech(inv: AntaInventory, root_dir: Path, *, configure: bool, tags: set[str] | None = None, latest: int | None = None) -> None:
     """Collect scheduled show-tech on devices."""
 
     async def collect(device: AntaDevice) -> None:
@@ -154,5 +154,5 @@ async def collect(device: AntaDevice) -> None:
 
     logger.info("Connecting to devices...")
     await inv.connect_inventory()
-    devices = inv.get_inventory(established_only=True, tags=tags).values()
+    devices = inv.get_inventory(established_only=True, tags=tags).devices
     await asyncio.gather(*(collect(device) for device in devices))

anta/cli/get/commands.py

@@ -10,7 +10,7 @@
 import json
 import logging
 from pathlib import Path
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
 
 import click
 from cvprac.cvp_client import CvpClient
@@ -37,6 +37,7 @@
 @click.option("--password", "-p", help="CloudVision password", type=str, required=True)
 @click.option("--container", "-c", help="CloudVision container where devices are configured", type=str)
 def from_cvp(ctx: click.Context, output: Path, host: str, username: str, password: str, container: str | None) -> None:
+    # pylint: disable=too-many-arguments
     """Build ANTA inventory from Cloudvision.
 
     TODO - handle get_inventory and get_devices_in_container failure
@@ -91,7 +92,7 @@ def from_ansible(ctx: click.Context, output: Path, ansible_group: str, ansible_i
 @click.command
 @inventory_options
 @click.option("--connected/--not-connected", help="Display inventory after connection has been created", default=False, required=False)
-def inventory(inventory: AntaInventory, tags: list[str] | None, *, connected: bool) -> None:
+def inventory(inventory: AntaInventory, tags: set[str] | None, *, connected: bool) -> None:
     """Show inventory loaded in ANTA."""
     # TODO: @gmuloc - tags come from context - we cannot have everything..
     # ruff: noqa: ARG001
@@ -107,11 +108,11 @@ def inventory(inventory: AntaInventory, tags: list[str] | None, *, connected: bo
 
 @click.command
 @inventory_options
-def tags(inventory: AntaInventory, tags: list[str] | None) -> None:  # pylint: disable=unused-argument
+def tags(inventory: AntaInventory, **kwargs: Any) -> None:
+    # pylint: disable=unused-argument
     """Get list of configured tags in user inventory."""
-    tags_found = []
+    tags: set[str] = set()
     for device in inventory.values():
-        tags_found += device.tags
-    tags_found = sorted(set(tags_found))
+        tags.update(device.tags)
     console.print("Tags found:")
-    console.print_json(json.dumps(tags_found, indent=2))
+    console.print_json(json.dumps(sorted(tags), indent=2))

anta/cli/get/utils.py

@@ -108,7 +108,7 @@ def create_inventory_from_cvp(inv: list[dict[str, Any]], output: Path) -> None:
             AntaInventoryHost(
                 name=dev["hostname"],
                 host=dev["ipAddress"],
-                tags=[dev["containerName"].lower()],
+                tags={dev["containerName"].lower()},
             )
         )
     write_inventory_to_file(hosts, output)

anta/cli/nrfu/__init__.py

@@ -6,12 +6,13 @@
 from __future__ import annotations
 
 import asyncio
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, get_args
 
 import click
 
 from anta.cli.nrfu import commands
 from anta.cli.utils import AliasedGroup, catalog_options, inventory_options
+from anta.custom_types import TestStatus
 from anta.models import AntaTest
 from anta.result_manager import ResultManager
 from anta.runner import main
@@ -52,14 +53,65 @@ def parse_args(self, ctx: click.Context, args: list[str]) -> list[str]:
             return super().parse_args(ctx, args)
 
 
+HIDE_STATUS: list[str] = list(get_args(TestStatus))
+HIDE_STATUS.remove("unset")
+
+
 @click.group(invoke_without_command=True, cls=IgnoreRequiredWithHelp)
 @click.pass_context
 @inventory_options
 @catalog_options
-@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 nrfu(ctx: click.Context, inventory: AntaInventory, tags: list[str] | None, catalog: AntaCatalog, *, ignore_status: bool, ignore_error: bool) -> None:
-    """Run ANTA tests on devices."""
+@click.option(
+    "--device",
+    "-d",
+    help="Run tests on a specific device. Can be provided multiple times.",
+    type=str,
+    multiple=True,
+    required=False,
+)
+@click.option(
+    "--test",
+    help="Run a specific test. Can be provided multiple times.",
+    type=str,
+    multiple=True,
+    required=False,
+)
+@click.option(
+    "--ignore-status",
+    help="Exit code will always be 0.",
+    show_envvar=True,
+    is_flag=True,
+    default=False,
+)
+@click.option(
+    "--ignore-error",
+    help="Exit code will be 0 if all tests succeeded or 1 if any test failed.",
+    show_envvar=True,
+    is_flag=True,
+    default=False,
+)
+@click.option(
+    "--hide",
+    default=None,
+    type=click.Choice(HIDE_STATUS, case_sensitive=False),
+    multiple=True,
+    help="Group result by test or device.",
+    required=False,
+)
+# pylint: disable=too-many-arguments
+def nrfu(
+    ctx: click.Context,
+    inventory: AntaInventory,
+    tags: set[str] | None,
+    catalog: AntaCatalog,
+    device: tuple[str],
+    test: tuple[str],
+    hide: tuple[str],
+    *,
+    ignore_status: bool,
+    ignore_error: bool,
+) -> None:
+    """Run ANTA tests on selected inventory devices."""
     # If help is invoke somewhere, skip the command
     if ctx.obj.get("_anta_help"):
         return
@@ -68,9 +120,10 @@ def nrfu(ctx: click.Context, inventory: AntaInventory, tags: list[str] | None, c
     ctx.obj["result_manager"] = ResultManager()
     ctx.obj["ignore_status"] = ignore_status
     ctx.obj["ignore_error"] = ignore_error
+    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))
+        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))
     # Invoke `anta nrfu table` if no command is passed
     if ctx.invoked_subcommand is None:
         ctx.invoke(commands.table)

anta/cli/nrfu/commands.py

@@ -7,6 +7,7 @@
 
 import logging
 import pathlib
+from typing import Literal
 
 import click
 
@@ -19,18 +20,19 @@
 
 @click.command()
 @click.pass_context
-@click.option("--device", "-d", help="Show a summary for this device", type=str, required=False)
-@click.option("--test", "-t", help="Show a summary for this test", type=str, required=False)
 @click.option(
     "--group-by",
     default=None,
     type=click.Choice(["device", "test"], case_sensitive=False),
-    help="Group result by test or host. default none",
+    help="Group result by test or device.",
     required=False,
 )
-def table(ctx: click.Context, device: str | None, test: str | None, group_by: str) -> None:
+def table(
+    ctx: click.Context,
+    group_by: Literal["device", "test"] | None,
+) -> None:
     """ANTA command to check network states with table result."""
-    print_table(results=ctx.obj["result_manager"], device=device, group_by=group_by, test=test)
+    print_table(ctx, group_by=group_by)
     exit_with_code(ctx)
 
 
@@ -46,17 +48,15 @@ def table(ctx: click.Context, device: str | None, test: str | None, group_by: st
 )
 def json(ctx: click.Context, output: pathlib.Path | None) -> None:
     """ANTA command to check network state with JSON result."""
-    print_json(results=ctx.obj["result_manager"], output=output)
+    print_json(ctx, output=output)
     exit_with_code(ctx)
 
 
 @click.command()
 @click.pass_context
-@click.option("--search", "-s", help="Regular expression to search in both name and test", type=str, required=False)
-@click.option("--skip-error", help="Hide tests in errors due to connectivity issue", default=False, is_flag=True, show_default=True, required=False)
-def text(ctx: click.Context, search: str | None, *, skip_error: bool) -> None:
+def text(ctx: click.Context) -> None:
     """ANTA command to check network states with text result."""
-    print_text(results=ctx.obj["result_manager"], search=search, skip_error=skip_error)
+    print_text(ctx)
     exit_with_code(ctx)