diff --git a/anta/cli/nrfu/__init__.py b/anta/cli/nrfu/__init__.py index 596654686..a85277102 100644 --- a/anta/cli/nrfu/__init__.py +++ b/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) diff --git a/anta/cli/nrfu/commands.py b/anta/cli/nrfu/commands.py index 7581116c6..cd750cb85 100644 --- a/anta/cli/nrfu/commands.py +++ b/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( diff --git a/anta/cli/nrfu/utils.py b/anta/cli/nrfu/utils.py index d4cd1317d..284c9b709 100644 --- a/anta/cli/nrfu/utils.py +++ b/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] diff --git a/anta/reporter/__init__.py b/anta/reporter/__init__.py index 685608dc2..7c911f243 100644 --- a/anta/reporter/__init__.py +++ b/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(): diff --git a/anta/reporter/csv_reporter.py b/anta/reporter/csv_reporter.py new file mode 100644 index 000000000..221cbec81 --- /dev/null +++ b/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 diff --git a/docs/cli/nrfu.md b/docs/cli/nrfu.md index 90b4a4045..afed25949 100644 --- a/docs/cli/nrfu.md +++ b/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. diff --git a/docs/imgs/anta_nrfu_csv.png b/docs/imgs/anta_nrfu_csv.png new file mode 100644 index 000000000..5e4129aad Binary files /dev/null and b/docs/imgs/anta_nrfu_csv.png differ diff --git a/docs/snippets/anta_nrfu_help.txt b/docs/snippets/anta_nrfu_help.txt index 48bb15b9b..365da0474 100644 --- a/docs/snippets/anta_nrfu_help.txt +++ b/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. diff --git a/tests/units/cli/nrfu/test_commands.py b/tests/units/cli/nrfu/test_commands.py index e2b5031eb..8ad7745f4 100644 --- a/tests/units/cli/nrfu/test_commands.py +++ b/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() diff --git a/tests/units/reporter/test_csv.py b/tests/units/reporter/test_csv.py new file mode 100644 index 000000000..e0a9d4e1f --- /dev/null +++ b/tests/units/reporter/test_csv.py @@ -0,0 +1,93 @@ +# 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. +"""Test anta.report.csv_reporter.py.""" + +# pylint: disable=too-few-public-methods + +import csv +import pathlib +from typing import Any, Callable + +import pytest + +from anta.reporter.csv_reporter import ReportCsv +from anta.result_manager import ResultManager + + +class TestReportCsv: + """Tester for ReportCsv class.""" + + def compare_csv_and_result(self, rows: list[Any], index: int, result_manager: ResultManager) -> None: + """Compare CSV and TestResult.""" + assert rows[index + 1][0] == result_manager.results[index].name + assert rows[index + 1][1] == result_manager.results[index].test + assert rows[index + 1][2] == result_manager.results[index].result + assert rows[index + 1][3] == ReportCsv().split_list_to_txt_list(result_manager.results[index].messages) + assert rows[index + 1][4] == result_manager.results[index].description + assert rows[index + 1][5] == ReportCsv().split_list_to_txt_list(result_manager.results[index].categories) + + def test_report_csv_generate( + self, + result_manager_factory: Callable[[int], ResultManager], + tmp_path: pathlib.Path, + ) -> None: + """Test CSV reporter.""" + max_test_entries = 10 + + # Create a temporary CSV file path + csv_filename = tmp_path / "test.csv" + + # Create a ResultManager instance with dummy test results + result_manager = result_manager_factory(max_test_entries) + # Test usecase with list of messages + result_manager.results[0].messages = ["Message 1", "Message 2"] + # Test usecase with list of categories + result_manager.results[1].messages = ["Cat 1", "Cat 2"] + + # Generate the CSV report + ReportCsv.generate(result_manager, csv_filename) + + # Read the generated CSV file + with pathlib.Path.open(csv_filename, encoding="utf-8") as csvfile: + reader = csv.reader(csvfile, delimiter=",") + rows = list(reader) + + # Assert the headers + assert rows[0] == [ + ReportCsv.Headers.device, + ReportCsv.Headers.test_name, + ReportCsv.Headers.test_status, + ReportCsv.Headers.messages, + ReportCsv.Headers.description, + ReportCsv.Headers.categories, + ] + + # Assert the test result rows + for index in [0, max_test_entries - 1]: + self.compare_csv_and_result(rows, index, result_manager) + + # Assert number of lines: Number of TestResults + CSV Headers + assert len(rows) == len(result_manager.results) + 1 + + def test_report_csv_generate_os_error( + self, + result_manager_factory: Callable[[int], ResultManager], + tmp_path: pathlib.Path, + caplog: pytest.LogCaptureFixture, + ) -> None: + """Test CSV reporter OSError.""" + # Create a ResultManager instance with dummy test results + max_test_entries = 10 + result_manager = result_manager_factory(max_test_entries) + + # Create a temporary CSV file path and make tmp_path read_only + tmp_path.chmod(0o400) + csv_filename = tmp_path / "read_only.csv" + + with pytest.raises(OSError, match="Permission denied"): + # Generate the CSV report + ReportCsv.generate(result_manager, csv_filename) + + assert len(caplog.record_tuples) == 1 + assert "OSError caught while writing the CSV file" in caplog.text