feat(anta): add test atomic results

.github/workflows/release.yml

@@ -30,7 +30,6 @@ jobs:
   release-doc:
     name: "Publish documentation for release ${{github.ref_name}}"
     runs-on: ubuntu-latest
-    needs: [release-coverage]
     steps:
       - uses: actions/checkout@v4
         with:

anta/cli/debug/utils.py

@@ -42,13 +42,11 @@ def debug_options(f: Callable[..., Any]) -> Callable[..., Any]:
     @functools.wraps(f)
     def wrapper(
         ctx: click.Context,
-        *args: tuple[Any],
+        *args: Any,  # noqa: ANN401
         inventory: AntaInventory,
         device: str,
-        **kwargs: Any,
-    ) -> Any:
-        # TODO: @gmuloc - tags come from context https://github.com/aristanetworks/anta/issues/584
-        # ruff: noqa: ARG001
+        **kwargs: Any,  # noqa: ANN401
+    ) -> Callable[..., Any]:
         if (d := inventory.get(device)) is None:
             logger.error("Device '%s' does not exist in Inventory", device)
             ctx.exit(ExitCode.USAGE_ERROR)

anta/cli/get/commands.py

@@ -7,10 +7,9 @@
 from __future__ import annotations
 
 import asyncio
-import json
 import logging
 from pathlib import Path
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING
 
 import click
 import requests
@@ -115,8 +114,6 @@ def from_ansible(ctx: click.Context, output: Path, ansible_group: str, ansible_i
 @click.option("--connected/--not-connected", help="Display inventory after connection has been created", default=False, required=False)
 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
     logger.debug("Requesting devices for tags: %s", tags)
     console.print("Current inventory content is:", style="white on blue")
 
@@ -129,13 +126,13 @@ def inventory(inventory: AntaInventory, tags: set[str] | None, *, connected: boo
 
 @click.command
 @inventory_options
-def tags(inventory: AntaInventory, **kwargs: Any) -> None:
+def tags(inventory: AntaInventory, tags: set[str] | None) -> None:  # noqa: ARG001
     """Get list of configured tags in user inventory."""
-    tags: set[str] = set()
+    t: set[str] = set()
     for device in inventory.values():
-        tags.update(device.tags)
-    console.print("Tags found:")
-    console.print_json(json.dumps(sorted(tags), indent=2))
+        t.update(device.tags)
+    console.print("Tags defined in inventory:")
+    console.print_json(data=sorted(t), indent=2)
 
 
 @click.command

anta/cli/get/utils.py

@@ -59,11 +59,11 @@ def inventory_output_options(f: Callable[..., Any]) -> Callable[..., Any]:
     @functools.wraps(f)
     def wrapper(
         ctx: click.Context,
-        *args: tuple[Any],
+        *args: Any,  # noqa: ANN401
         output: Path,
         overwrite: bool,
-        **kwargs: dict[str, Any],
-    ) -> Any:
+        **kwargs: Any,  # noqa: ANN401
+    ) -> Callable[..., Any]:
         # Boolean to check if the file is empty
         output_is_not_empty = output.exists() and output.stat().st_size != 0
         # Check overwrite when file is not empty

anta/cli/nrfu/commands.py

@@ -27,10 +27,19 @@
     help="Group result by test or device.",
     required=False,
 )
-def table(ctx: click.Context, group_by: Literal["device", "test"] | None) -> None:
+@click.option(
+    "--expand-atomic",
+    "-x",
+    default=False,
+    show_envvar=True,
+    is_flag=True,
+    show_default=True,
+    help="Flag to indicate if atomic results should be rendered",
+)
+def table(ctx: click.Context, group_by: Literal["device", "test"] | None, expand_atomic: bool) -> None:
     """ANTA command to check network state with table results."""
     run_tests(ctx)
-    print_table(ctx, group_by=group_by)
+    print_table(ctx, expand_atomic, group_by)
     exit_with_code(ctx)
 
 
@@ -53,10 +62,19 @@ def json(ctx: click.Context, output: pathlib.Path | None) -> None:
 
 @click.command()
 @click.pass_context
-def text(ctx: click.Context) -> None:
+@click.option(
+    "--expand-atomic",
+    "-x",
+    default=False,
+    show_envvar=True,
+    is_flag=True,
+    show_default=True,
+    help="Flag to indicate if atomic results should be rendered",
+)
+def text(ctx: click.Context, expand_atomic: bool) -> None:
     """ANTA command to check network state with text results."""
     run_tests(ctx)
-    print_text(ctx)
+    print_text(ctx, expand_atomic)
     exit_with_code(ctx)
 
 

anta/cli/nrfu/utils.py

@@ -80,7 +80,7 @@ def print_settings(
     console.print()
 
 
-def print_table(ctx: click.Context, group_by: Literal["device", "test"] | None = None) -> None:
+def print_table(ctx: click.Context, expand_atomic: bool, group_by: Literal["device", "test"] | None) -> None:
     """Print result in a table."""
     reporter = ReportTable()
     console.print()
@@ -90,8 +90,10 @@ def print_table(ctx: click.Context, group_by: Literal["device", "test"] | None =
         console.print(reporter.report_summary_devices(results))
     elif group_by == "test":
         console.print(reporter.report_summary_tests(results))
+    elif expand_atomic:
+        console.print(reporter.report_expanded(results))
     else:
-        console.print(reporter.report_all(results))
+        console.print(reporter.report(results))
 
 
 def print_json(ctx: click.Context, output: pathlib.Path | None = None) -> None:
@@ -112,16 +114,18 @@ def print_json(ctx: click.Context, output: pathlib.Path | None = None) -> None:
             ctx.exit(ExitCode.USAGE_ERROR)
 
 
-def print_text(ctx: click.Context) -> None:
+def print_text(ctx: click.Context, expand_atomic: bool) -> None:
     """Print results as simple text."""
     console.print()
-    for test in _get_result_manager(ctx).results:
-        if len(test.messages) <= 1:
-            message = test.messages[0] if len(test.messages) == 1 else ""
-            console.print(f"{test.name} :: {test.test} :: [{test.result}]{test.result.upper()}[/{test.result}]({message})", highlight=False)
-        else:  # len(test.messages) > 1
-            console.print(f"{test.name} :: {test.test} :: [{test.result}]{test.result.upper()}[/{test.result}]", highlight=False)
-            console.print("\n".join(f"    {message}" for message in test.messages), highlight=False)
+    for result in _get_result_manager(ctx).results:
+        console.print(f"{result.name} :: {result.test} :: [{result.result}]{result.result.upper()}[/{result.result}]", highlight=False)
+        if result.messages and not expand_atomic:
+            console.print("\n".join(f"    {message}" for message in result.messages), highlight=False)
+        if expand_atomic:
+            for r in result.atomic_results:
+                console.print(f"    {r.description} :: [{r.result}]{r.result.upper()}[/{r.result}]", highlight=False)
+                if r.messages:
+                    console.print("\n".join(f"      {message}" for message in r.messages), highlight=False)
 
 
 def print_jinja(results: ResultManager, template: pathlib.Path, output: pathlib.Path | None = None) -> None:

anta/cli/utils.py

@@ -90,8 +90,8 @@ class AliasedGroup(click.Group):
     From Click documentation.
     """
 
-    def get_command(self, ctx: click.Context, cmd_name: str) -> Any:
-        """Todo: document code."""
+    def get_command(self, ctx: click.Context, cmd_name: str) -> click.Command | None:
+        """Try to find a command name based on a prefix."""
         rv = click.Group.get_command(self, ctx, cmd_name)
         if rv is not None:
             return rv
@@ -103,12 +103,11 @@ def get_command(self, ctx: click.Context, cmd_name: str) -> Any:
         ctx.fail(f"Too many matches: {', '.join(sorted(matches))}")
         return None
 
-    def resolve_command(self, ctx: click.Context, args: Any) -> Any:
-        """Todo: document code."""
-        # always return the full command name
+    def resolve_command(self, ctx: click.Context, args: list[str]) -> tuple[str | None, click.Command | None, list[str]]:
+        """Return the full command name as first tuple element."""
         _, cmd, args = super().resolve_command(ctx, args)
         if not cmd:
-            return None, None, None
+            return None, None, []
         return cmd.name, cmd, args
 
 
@@ -194,7 +193,7 @@ def core_options(f: Callable[..., Any]) -> Callable[..., Any]:
     @functools.wraps(f)
     def wrapper(
         ctx: click.Context,
-        *args: tuple[Any],
+        *args: Any,  # noqa: ANN401
         inventory: Path,
         username: str,
         password: str | None,
@@ -204,8 +203,8 @@ def wrapper(
         timeout: float,
         insecure: bool,
         disable_cache: bool,
-        **kwargs: dict[str, Any],
-    ) -> Any:
+        **kwargs: Any,  # noqa: ANN401
+    ) -> Callable[..., Any]:
         # If help is invoke somewhere, do not parse inventory
         if ctx.obj.get("_anta_help"):
             return f(*args, inventory=None, **kwargs)
@@ -266,10 +265,10 @@ def inventory_options(f: Callable[..., Any]) -> Callable[..., Any]:
     @functools.wraps(f)
     def wrapper(
         ctx: click.Context,
-        *args: tuple[Any],
+        *args: Any,  # noqa: ANN401
         tags: set[str] | None,
-        **kwargs: dict[str, Any],
-    ) -> Any:
+        **kwargs: Any,  # noqa: ANN401
+    ) -> Callable[..., Any]:
         # If help is invoke somewhere, do not parse inventory
         if ctx.obj.get("_anta_help"):
             return f(*args, tags=tags, **kwargs)
@@ -308,11 +307,11 @@ def catalog_options(f: Callable[..., Any]) -> Callable[..., Any]:
     @functools.wraps(f)
     def wrapper(
         ctx: click.Context,
-        *args: tuple[Any],
+        *args: Any,  # noqa: ANN401
         catalog: Path,
         catalog_format: str,
-        **kwargs: dict[str, Any],
-    ) -> Any:
+        **kwargs: Any,  # noqa: ANN401
+    ) -> Callable[..., Any]:
         # If help is invoke somewhere, do not parse catalog
         if ctx.obj.get("_anta_help"):
             return f(*args, catalog=None, **kwargs)

anta/constants.py

@@ -43,5 +43,6 @@
     r".* does not support IP",
     r"IS-IS (.*) is disabled because: .*",
     r"No source interface .*",
+    r"There seem to be no power supplies connected.",
 ]
 """List of known EOS errors that should set a test status to 'failure' with the error message."""

anta/input_models/connectivity.py

@@ -18,6 +18,8 @@ class Host(BaseModel):
     """Model for a remote host to ping."""
 
     model_config = ConfigDict(extra="forbid")
+    description: str | None = None
+    """Description of the remote destination."""
     destination: IPv4Address
     """IPv4 address to ping."""
     source: IPv4Address | Interface
@@ -32,15 +34,8 @@ class Host(BaseModel):
     """Enable do not fragment bit in IP header. Defaults to False."""
 
     def __str__(self) -> str:
-        """Return a human-readable string representation of the Host for reporting.
-
-        Examples
-        --------
-        Host 10.1.1.1 (src: 10.2.2.2, vrf: mgmt, size: 100B, repeat: 2)
-
-        """
-        df_status = ", df-bit: enabled" if self.df_bit else ""
-        return f"Host {self.destination} (src: {self.source}, vrf: {self.vrf}, size: {self.size}B, repeat: {self.repeat}{df_status})"
+        """Return a human-readable string representation of the Host for reporting."""
+        return f"Destination {self.destination}{f' ({self.description})' if self.description is not None else ''} from {self.source} in VRF {self.vrf}"
 
 
 class LLDPNeighbor(BaseModel):

anta/inventory/__init__.py

@@ -60,7 +60,7 @@ def _update_disable_cache(kwargs: dict[str, Any], *, inventory_disable_cache: bo
     def _parse_hosts(
         inventory_input: AntaInventoryInput,
         inventory: AntaInventory,
-        **kwargs: dict[str, Any],
+        **kwargs: Any,  # noqa: ANN401
     ) -> None:
         """Parse the host section of an AntaInventoryInput and add the devices to the inventory.
 
@@ -92,7 +92,7 @@ def _parse_hosts(
     def _parse_networks(
         inventory_input: AntaInventoryInput,
         inventory: AntaInventory,
-        **kwargs: dict[str, Any],
+        **kwargs: Any,  # noqa: ANN401
     ) -> None:
         """Parse the network section of an AntaInventoryInput and add the devices to the inventory.
 
@@ -129,7 +129,7 @@ def _parse_networks(
     def _parse_ranges(
         inventory_input: AntaInventoryInput,
         inventory: AntaInventory,
-        **kwargs: dict[str, Any],
+        **kwargs: Any,  # noqa: ANN401
     ) -> None:
         """Parse the range section of an AntaInventoryInput and add the devices to the inventory.
 

anta/models.py

@@ -19,7 +19,7 @@
 from anta.constants import KNOWN_EOS_ERRORS
 from anta.custom_types import REGEXP_EOS_BLACKLIST_CMDS, Revision
 from anta.logger import anta_log_exception, exc_to_str
-from anta.result_manager.models import AntaTestStatus, TestResult
+from anta.result_manager.models import TestResult
 
 if TYPE_CHECKING:
     from collections.abc import Coroutine
@@ -448,15 +448,16 @@ def __init__(
         self.device: AntaDevice = device
         self.inputs: AntaTest.Input
         self.instance_commands: list[AntaCommand] = []
-        self.result: TestResult = TestResult(
-            name=device.name,
-            test=self.name,
-            categories=self.categories,
-            description=self.description,
-        )
+        self.result: TestResult = TestResult(name=device.name, test=self.name, categories=self.categories, description=self.description)
         self._init_inputs(inputs)
-        if self.result.result == AntaTestStatus.UNSET:
+        if hasattr(self, "inputs"):
             self._init_commands(eos_data)
+            if res_ow := self.inputs.result_overwrite:
+                if res_ow.categories:
+                    self.result.categories = res_ow.categories
+                if res_ow.description:
+                    self.result.description = res_ow.description
+                self.result.custom_field = res_ow.custom_field
 
     def _init_inputs(self, inputs: dict[str, Any] | AntaTest.Input | None) -> None:
         """Instantiate the `inputs` instance attribute with an `AntaTest.Input` instance to validate test inputs using the model.
@@ -477,12 +478,7 @@ def _init_inputs(self, inputs: dict[str, Any] | AntaTest.Input | None) -> None:
             self.logger.error(message)
             self.result.is_error(message=message)
             return
-        if res_ow := self.inputs.result_overwrite:
-            if res_ow.categories:
-                self.result.categories = res_ow.categories
-            if res_ow.description:
-                self.result.description = res_ow.description
-            self.result.custom_field = res_ow.custom_field
+        self.result.inputs = self.inputs
 
     def _init_commands(self, eos_data: list[dict[Any, Any] | str] | None) -> None:
         """Instantiate the `instance_commands` instance attribute from the `commands` class attribute.
@@ -615,7 +611,7 @@ def anta_test(function: F) -> Callable[..., Coroutine[Any, Any, TestResult]]:
         async def wrapper(
             self: AntaTest,
             eos_data: list[dict[Any, Any] | str] | None = None,
-            **kwargs: dict[str, Any],
+            **kwargs: Any,  # noqa: ANN401
         ) -> TestResult:
             """Inner function for the anta_test decorator.