feat(anta): Add support of CSV file export (#672)

---------
Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
Co-authored-by: Guillaume Mulocher <[email protected]>
Co-authored-by: Carl Baillargeon <[email protected]>

anta/cli/nrfu/__init__.py

@@ -143,6 +143,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",
+)
+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

@@ -15,8 +15,10 @@
 from rich.progress import BarColumn, MofNCompleteColumn, Progress, SpinnerColumn, TextColumn, TimeElapsedColumn, TimeRemainingColumn
 
 from anta.cli.console import console
+from anta.cli.utils import ExitCode
 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,16 @@ 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:
+    """Save results to a CSV file."""
+    try:
+        ReportCsv.generate(results=_get_result_manager(ctx), csv_filename=csv_file)
+        console.print(f"CSV report saved to {csv_file} ✅", style="cyan")
+    except OSError:
+        console.print(f"Failed to save CSV report to {csv_file} ❌", style="cyan")
+        ctx.exit(ExitCode.USAGE_ERROR)
+
+
 # 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

@@ -7,6 +7,7 @@
 from __future__ import annotations
 
 import logging
+from dataclasses import dataclass
 from typing import TYPE_CHECKING, Any
 
 from jinja2 import Template
@@ -27,6 +28,19 @@
 class ReportTable:
     """TableReport Generate a Table based on TestResult."""
 
+    @dataclass()
+    class Headers:  # pylint: disable=too-many-instance-attributes
+        """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 +76,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 +146,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 +194,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():

anta/reporter/csv_reporter.py

@@ -0,0 +1,109 @@
+# 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
+from dataclasses import dataclass
+from typing import TYPE_CHECKING
+
+from anta.logger import anta_log_exception
+
+if TYPE_CHECKING:
+    import pathlib
+
+    from anta.result_manager import ResultManager
+    from anta.result_manager.models import TestResult
+
+logger = logging.getLogger(__name__)
+
+
+class ReportCsv:
+    """Build a CSV report."""
+
+    @dataclass()
+    class Headers:
+        """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"
+
+    @classmethod
+    def split_list_to_txt_list(cls, usr_list: list[str], delimiter: str = " - ") -> str:
+        """Split list to multi-lines string.
+
+        Parameters
+        ----------
+            usr_list: List of string to concatenate
+            delimiter: A delimiter to use to start string. Defaults to None.
+
+        Returns
+        -------
+            str: Multi-lines string
+
+        """
+        return f"{delimiter}".join(f"{line}" for line in usr_list)
+
+    @classmethod
+    def convert_to_list(cls, result: TestResult) -> list[str]:
+        """
+        Convert a TestResult into a list of string for creating file content.
+
+        Args:
+        ----
+            results: A TestResult to convert into list.
+        """
+        message = cls.split_list_to_txt_list(result.messages) if len(result.messages) > 0 else ""
+        categories = cls.split_list_to_txt_list(result.categories) if len(result.categories) > 0 else "None"
+        return [
+            str(result.name),
+            result.test,
+            result.result,
+            message,
+            result.description,
+            categories,
+        ]
+
+    @classmethod
+    def generate(cls, results: ResultManager, csv_filename: pathlib.Path) -> None:
+        """Build CSV flle with tests results.
+
+        Parameter
+        ---------
+            results: A ResultManager instance.
+            csv_filename: File path where to save CSV data.
+
+        Raise
+        -----
+            OSError if any is raised while writing the CSV file.
+        """
+        headers = [
+            cls.Headers.device,
+            cls.Headers.test_name,
+            cls.Headers.test_status,
+            cls.Headers.messages,
+            cls.Headers.description,
+            cls.Headers.categories,
+        ]
+
+        try:
+            with csv_filename.open(mode="w", encoding="utf-8") as csvfile:
+                csvwriter = csv.writer(
+                    csvfile,
+                    delimiter=",",
+                )
+                csvwriter.writerow(headers)
+                for entry in results.results:
+                    csvwriter.writerow(cls.convert_to_list(entry))
+        except OSError as exc:
+            message = f"OSError caught while writing the CSV file '{csv_filename.resolve()}'."
+            anta_log_exception(exc, message, logger)
+            raise

docs/cli/nrfu.md

@@ -145,6 +145,28 @@ anta nrfu --tags LEAF json
 ```
 ![$1anta nrfu json results](../imgs/anta-nrfu-json-output.png){ loading=lazy width="1600" }
 
+## Performing NRFU and saving results in a CSV file.
+
+The `csv` command in NRFU testing is useful for generating a CSV file with all tests result. This file can be easily analyzed and filtered by operator for reporting purposes.
+
+### Command overview
+
+```bash
+anta nrfu csv --help
+Usage: anta nrfu csv [OPTIONS]
+
+  ANTA command to check network states with CSV result.
+
+Options:
+  --csv-output FILE  Path to save report as a CSV file  [env var:
+                     ANTA_NRFU_CSV_CSV_OUTPUT]
+  --help             Show this message and exit.
+```
+
+### Example
+
+![anta nrfu csv results](../imgs/anta_nrfu_csv.png){ loading=lazy width="1600" }
+
 ## Performing NRFU with custom reports
 
 ANTA offers a CLI option for creating custom reports. This leverages the Jinja2 template system, allowing you to tailor reports to your specific needs.

docs/snippets/anta_nrfu_help.txt

@@ -43,7 +43,7 @@ Options:
                                   or 1 if any test failed.  [env var:
                                   ANTA_NRFU_IGNORE_ERROR]
   --hide [success|failure|error|skipped]
-                                  Hide result by type: success / failure /
+                                  Hide results by type: success / failure /
                                   error / skipped'.
   --dry-run                       Run anta nrfu command but stop before
                                   starting to execute the tests. Considers all
@@ -52,6 +52,7 @@ Options:
   --help                          Show this message and exit.
 
 Commands:
+  csv         ANTA command to check network state with CSV report.
   json        ANTA command to check network state with JSON result.
   table       ANTA command to check network states with table result.
   text        ANTA command to check network states with text result.

tests/units/cli/nrfu/test_commands.py

@@ -9,6 +9,7 @@
 import re
 from pathlib import Path
 from typing import TYPE_CHECKING
+from unittest.mock import patch
 
 from anta.cli import anta
 from anta.cli.utils import ExitCode
@@ -94,3 +95,22 @@ def test_anta_nrfu_template(click_runner: CliRunner) -> None:
     result = click_runner.invoke(anta, ["nrfu", "tpl-report", "--template", str(DATA_DIR / "template.j2")])
     assert result.exit_code == ExitCode.OK
     assert "* VerifyEOSVersion is SUCCESS for dummy" in result.output
+
+
+def test_anta_nrfu_csv(click_runner: CliRunner, tmp_path: Path) -> None:
+    """Test anta nrfu csv."""
+    csv_output = tmp_path / "test.csv"
+    result = click_runner.invoke(anta, ["nrfu", "csv", "--csv-output", str(csv_output)])
+    assert result.exit_code == ExitCode.OK
+    assert "CSV report saved to" in result.output
+    assert csv_output.exists()
+
+
+def test_anta_nrfu_csv_failure(click_runner: CliRunner, tmp_path: Path) -> None:
+    """Test anta nrfu csv."""
+    csv_output = tmp_path / "test.csv"
+    with patch("anta.reporter.csv_reporter.ReportCsv.generate", side_effect=OSError()):
+        result = click_runner.invoke(anta, ["nrfu", "csv", "--csv-output", str(csv_output)])
+    assert result.exit_code == ExitCode.USAGE_ERROR
+    assert "Failed to save CSV report to" in result.output
+    assert not csv_output.exists()