feat(anta): Add support of CSV file export

anta/cli/nrfu/__init__.py

@@ -141,6 +141,7 @@ def nrfu(
 
 
 nrfu.add_command(commands.table)
+nrfu.add_command(commands.csv)
 nrfu.add_command(commands.json)
 nrfu.add_command(commands.text)
 nrfu.add_command(commands.tpl_report)

anta/cli/nrfu/commands.py

@@ -13,7 +13,7 @@
 
 from anta.cli.utils import exit_with_code
 
-from .utils import print_jinja, print_json, print_table, print_text, run_tests
+from .utils import print_jinja, print_json, print_table, print_text, run_tests, save_to_csv
 
 logger = logging.getLogger(__name__)
 
@@ -27,10 +27,7 @@
     help="Group result by test or device.",
     required=False,
 )
-def table(
-    ctx: click.Context,
-    group_by: Literal["device", "test"] | None,
-) -> None:
+def table(ctx: click.Context, group_by: Literal["device", "test"] | None) -> None:
     """ANTA command to check network states with table result."""
     run_tests(ctx)
     print_table(ctx, group_by=group_by)
@@ -63,6 +60,28 @@ def text(ctx: click.Context) -> None:
     exit_with_code(ctx)
 
 
+@click.command()
+@click.pass_context
+@click.option(
+    "--csv-output",
+    type=click.Path(
+        file_okay=True,
+        dir_okay=False,
+        exists=False,
+        writable=True,
+        path_type=pathlib.Path,
+    ),
+    show_envvar=True,
+    required=False,
+    help="Path to save report as a CSV file. It only saves test results and not the output from --group-by option",
+)
+def csv(ctx: click.Context, csv_output: pathlib.Path) -> None:
+    """ANTA command to check network states with CSV result."""
+    run_tests(ctx)
+    save_to_csv(ctx, csv_file=csv_output)
+    exit_with_code(ctx)
+
+
 @click.command()
 @click.pass_context
 @click.option(

anta/cli/nrfu/utils.py

@@ -11,12 +11,14 @@
 from typing import TYPE_CHECKING, Literal
 
 import rich
+from rich.emoji import Emoji
 from rich.panel import Panel
 from rich.progress import BarColumn, MofNCompleteColumn, Progress, SpinnerColumn, TextColumn, TimeElapsedColumn, TimeRemainingColumn
 
 from anta.cli.console import console
 from anta.models import AntaTest
 from anta.reporter import ReportJinja, ReportTable
+from anta.reporter.csv_reporter import ReportCsv
 from anta.runner import main
 
 if TYPE_CHECKING:
@@ -122,6 +124,14 @@ def print_jinja(results: ResultManager, template: pathlib.Path, output: pathlib.
             file.write(report)
 
 
+def save_to_csv(ctx: click.Context, csv_file: pathlib.Path | None = None) -> None:
+    """Save results to a CSV file."""
+    if csv_file is not None:
+        ReportCsv().csv_report(results=_get_result_manager(ctx), csv_filename=csv_file)
+        checkmark = Emoji("white_check_mark")
+        console.print(f"CSV report saved to {csv_file} {checkmark}", style="cyan")
+
+
 # Adding our own ANTA spinner - overriding rich SPINNERS for our own
 # so ignore warning for redefinition
 rich.spinner.SPINNERS = {  # type: ignore[attr-defined]

anta/reporter/__init__.py

@@ -6,17 +6,18 @@
 # pylint: disable = too-few-public-methods
 from __future__ import annotations
 
+import csv
 import logging
+import pathlib
 from typing import TYPE_CHECKING, Any
 
 from jinja2 import Template
+from pydantic import BaseModel
 from rich.table import Table
 
 from anta import RICH_COLOR_PALETTE, RICH_COLOR_THEME
 
 if TYPE_CHECKING:
-    import pathlib
-
     from anta.custom_types import TestStatus
     from anta.result_manager import ResultManager
     from anta.result_manager.models import TestResult
@@ -27,6 +28,18 @@
 class ReportTable:
     """TableReport Generate a Table based on TestResult."""
 
+    class Headers(BaseModel):
+        """Headers for the table report."""
+
+        device: str = "Device"
+        test_case: str = "Test Name"
+        number_of_success: str = "# of success"
+        number_of_failure: str = "# of failure"
+        number_of_skipped: str = "# of skipped"
+        number_of_errors: str = "# of errors"
+        list_of_error_nodes: str = "List of failed or error nodes"
+        list_of_error_tests: str = "List of failed or error test cases"
+
     def _split_list_to_txt_list(self, usr_list: list[str], delimiter: str | None = None) -> str:
         """Split list to multi-lines string.
 
@@ -62,9 +75,6 @@ def _build_headers(self, headers: list[str], table: Table) -> Table:
         for idx, header in enumerate(headers):
             if idx == 0:
                 table.add_column(header, justify="left", style=RICH_COLOR_PALETTE.HEADER, no_wrap=True)
-            elif header == "Test Name":
-                # We always want the full test name
-                table.add_column(header, justify="left", no_wrap=True)
             else:
                 table.add_column(header, justify="left")
         return table
@@ -135,12 +145,12 @@ def report_summary_tests(
         """
         table = Table(title=title, show_lines=True)
         headers = [
-            "Test Case",
-            "# of success",
-            "# of skipped",
-            "# of failure",
-            "# of errors",
-            "List of failed or error nodes",
+            self.Headers().test_case,
+            self.Headers().number_of_success,
+            self.Headers().number_of_skipped,
+            self.Headers().number_of_failure,
+            self.Headers().number_of_errors,
+            self.Headers().list_of_error_nodes,
         ]
         table = self._build_headers(headers=headers, table=table)
         for test in manager.get_tests():
@@ -183,12 +193,12 @@ def report_summary_devices(
         """
         table = Table(title=title, show_lines=True)
         headers = [
-            "Device",
-            "# of success",
-            "# of skipped",
-            "# of failure",
-            "# of errors",
-            "List of failed or error test cases",
+            self.Headers().device,
+            self.Headers().number_of_success,
+            self.Headers().number_of_skipped,
+            self.Headers().number_of_failure,
+            self.Headers().number_of_errors,
+            self.Headers().list_of_error_tests,
         ]
         table = self._build_headers(headers=headers, table=table)
         for device in manager.get_devices():
@@ -254,3 +264,73 @@ def render(self, data: list[dict[str, Any]], *, trim_blocks: bool = True, lstrip
             template = Template(file_.read(), trim_blocks=trim_blocks, lstrip_blocks=lstrip_blocks)
 
         return template.render({"data": data})
+
+
+class ReportCsv:
+    """Build a CSV report."""
+
+    class Headers(BaseModel):
+        """Headers for the CSV report."""
+
+        device: str = "Device"
+        test_name: str = "Test Name"
+        test_status: str = "Test Status"
+        messages: str = "Message(s)"
+        description: str = "Test description"
+        categories: str = "Test category"
+
+    def _split_list_to_txt_list(self, usr_list: list[str], delimiter: str | None = None) -> str:
+        """Split list to multi-lines string.
+
+        Args:
+        ----
+            usr_list (list[str]): List of string to concatenate
+            delimiter (str, optional): A delimiter to use to start string. Defaults to None.
+
+        Returns
+        -------
+            str: Multi-lines string
+
+        """
+        if delimiter is not None:
+            return "\n".join(f"{delimiter} {line}" for line in usr_list)
+        return "\n".join(f"{line}" for line in usr_list)
+
+    def csv_report(self, results: ResultManager, csv_filename: pathlib.Path) -> None:
+        """Build CSV flle with tests results.
+
+        Args:
+        ----
+            results: A ResultManager instance.
+            csv_filename: File path where to save CSV data.
+        """
+
+        def add_line(result: TestResult) -> list[str]:
+            message = self._split_list_to_txt_list(result.messages) if len(result.messages) > 0 else ""
+            categories = ", ".join(result.categories)
+            return [
+                str(result.name),
+                result.test,
+                result.result,
+                message.replace("\n", "\r\n"),
+                result.description,
+                categories,
+            ]
+
+        headers = [
+            self.Headers().device,
+            self.Headers().test_name,
+            self.Headers().test_status,
+            self.Headers().messages,
+            self.Headers().description,
+            self.Headers().categories,
+        ]
+
+        with pathlib.Path.open(csv_filename, "w", encoding="utf-8") as csvfile:
+            spamwriter = csv.writer(
+                csvfile,
+                delimiter=",",
+            )
+            spamwriter.writerow(headers)
+            for entry in results.results:
+                spamwriter.writerow(add_line(entry))

anta/reporter/csv_reporter.py

@@ -0,0 +1,90 @@
+# Copyright (c) 2023-2024 Arista Networks, Inc.
+# Use of this source code is governed by the Apache License 2.0
+# that can be found in the LICENSE file.
+"""CSV Report management for ANTA."""
+
+# pylint: disable = too-few-public-methods
+from __future__ import annotations
+
+import csv
+import logging
+import pathlib
+from typing import TYPE_CHECKING
+
+from pydantic import BaseModel
+
+if TYPE_CHECKING:
+    from anta.result_manager import ResultManager
+    from anta.result_manager.models import TestResult
+
+logger = logging.getLogger(__name__)
+
+
+class ReportCsv:
+    """Build a CSV report."""
+
+    class Headers(BaseModel):
+        """Headers for the CSV report."""
+
+        device: str = "Device"
+        test_name: str = "Test Name"
+        test_status: str = "Test Status"
+        messages: str = "Message(s)"
+        description: str = "Test description"
+        categories: str = "Test category"
+
+    def _split_list_to_txt_list(self, usr_list: list[str], delimiter: str | None = None) -> str:
+        """Split list to multi-lines string.
+
+        Args:
+        ----
+            usr_list (list[str]): List of string to concatenate
+            delimiter (str, optional): A delimiter to use to start string. Defaults to None.
+
+        Returns
+        -------
+            str: Multi-lines string
+
+        """
+        if delimiter is not None:
+            return "\n".join(f"{delimiter} {line}" for line in usr_list)
+        return "\n".join(f"{line}" for line in usr_list)
+
+    def csv_report(self, results: ResultManager, csv_filename: pathlib.Path) -> None:
+        """Build CSV flle with tests results.
+
+        Args:
+        ----
+            results: A ResultManager instance.
+            csv_filename: File path where to save CSV data.
+        """
+
+        def add_line(result: TestResult) -> list[str]:
+            message = self._split_list_to_txt_list(result.messages) if len(result.messages) > 0 else ""
+            categories = ", ".join(result.categories)
+            return [
+                str(result.name),
+                result.test,
+                result.result,
+                message.replace("\n", "\r\n"),
+                result.description,
+                categories,
+            ]
+
+        headers = [
+            self.Headers().device,
+            self.Headers().test_name,
+            self.Headers().test_status,
+            self.Headers().messages,
+            self.Headers().description,
+            self.Headers().categories,
+        ]
+
+        with pathlib.Path.open(csv_filename, "w", encoding="utf-8") as csvfile:
+            spamwriter = csv.writer(
+                csvfile,
+                delimiter=",",
+            )
+            spamwriter.writerow(headers)
+            for entry in results.results:
+                spamwriter.writerow(add_line(entry))