Merge branch 'main' into dependabot/pip/griffe-gte-0.46-and-lt-2.0.0

.pre-commit-config.yaml

@@ -32,7 +32,7 @@ repos:
       - name: Check and insert license on Markdown files
         id: insert-license
         files: .*\.md$
-        # exclude:
+        exclude: ^tests/data/.*\.md$
         args:
           - --license-filepath
           - .github/license-short.txt
@@ -43,7 +43,7 @@ repos:
           - '<!--|  ~|  -->'
 
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: v0.6.1
+    rev: v0.6.4
     hooks:
     - id: ruff
       name: Run Ruff linter
@@ -52,7 +52,7 @@ repos:
       name: Run Ruff formatter
 
   - repo: https://github.com/pycqa/pylint
-    rev: "v3.2.6"
+    rev: "v3.2.7"
     hooks:
       - id: pylint
         name: Check code style with pylint
@@ -80,7 +80,7 @@ repos:
         types: [text]
 
   - repo: https://github.com/pre-commit/mirrors-mypy
-    rev: v1.11.1
+    rev: v1.11.2
     hooks:
       - id: mypy
         name: Check typing with mypy

anta/catalog.py

@@ -10,9 +10,11 @@
 import math
 from collections import defaultdict
 from inspect import isclass
+from itertools import chain
 from json import load as json_load
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, Literal, Optional, Union
+from warnings import warn
 
 from pydantic import BaseModel, ConfigDict, RootModel, ValidationError, ValidationInfo, field_validator, model_serializer, model_validator
 from pydantic.types import ImportString
@@ -386,6 +388,21 @@ def from_list(data: ListAntaTestTuples) -> AntaCatalog:
             raise
         return AntaCatalog(tests)
 
+    @classmethod
+    def merge_catalogs(cls, catalogs: list[AntaCatalog]) -> AntaCatalog:
+        """Merge multiple AntaCatalog instances.
+
+        Parameters
+        ----------
+            catalogs: A list of AntaCatalog instances to merge.
+
+        Returns
+        -------
+            A new AntaCatalog instance containing the tests of all the input catalogs.
+        """
+        combined_tests = list(chain(*(catalog.tests for catalog in catalogs)))
+        return cls(tests=combined_tests)
+
     def merge(self, catalog: AntaCatalog) -> AntaCatalog:
         """Merge two AntaCatalog instances.
 
@@ -397,7 +414,13 @@ def merge(self, catalog: AntaCatalog) -> AntaCatalog:
         -------
             A new AntaCatalog instance containing the tests of the two instances.
         """
-        return AntaCatalog(tests=self.tests + catalog.tests)
+        # TODO: Use a decorator to deprecate this method instead. See https://github.com/aristanetworks/anta/issues/754
+        warn(
+            message="AntaCatalog.merge() is deprecated and will be removed in ANTA v2.0. Use AntaCatalog.merge_catalogs() instead.",
+            category=DeprecationWarning,
+            stacklevel=2,
+        )
+        return self.merge_catalogs([self, catalog])
 
     def dump(self) -> AntaCatalogFile:
         """Return an AntaCatalogFile instance from this AntaCatalog instance.

anta/cli/debug/commands.py

@@ -72,13 +72,16 @@ def run_template(
     revision: int,
 ) -> None:
     # pylint: disable=too-many-arguments
+    # Using \b for click
+    # ruff: noqa: D301
     """Run arbitrary templated command to an ANTA device.
 
     Takes a list of arguments (keys followed by a value) to build a dictionary used as template parameters.
 
-    Example:
+    \b
+    Example
     -------
-    anta debug run-template -d leaf1a -t 'show vlan {vlan_id}' vlan_id 1
+        anta debug run-template -d leaf1a -t 'show vlan {vlan_id}' vlan_id 1
 
     """
     template_params = dict(zip(params[::2], params[1::2]))

anta/cli/debug/utils.py

@@ -11,7 +11,7 @@
 
 import click
 
-from anta.cli.utils import ExitCode, inventory_options
+from anta.cli.utils import ExitCode, core_options
 
 if TYPE_CHECKING:
     from anta.inventory import AntaInventory
@@ -22,7 +22,7 @@
 def debug_options(f: Callable[..., Any]) -> Callable[..., Any]:
     """Click common options required to execute a command on a specific device."""
 
-    @inventory_options
+    @core_options
     @click.option(
         "--ofmt",
         type=click.Choice(["json", "text"]),
@@ -44,7 +44,6 @@ def wrapper(
         ctx: click.Context,
         *args: tuple[Any],
         inventory: AntaInventory,
-        tags: set[str] | None,
         device: str,
         **kwargs: Any,
     ) -> Any:

anta/cli/nrfu/__init__.py

@@ -5,14 +5,14 @@
 
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, get_args
+from typing import TYPE_CHECKING
 
 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.result_manager import ResultManager
+from anta.result_manager.models import AntaTestStatus
 
 if TYPE_CHECKING:
     from anta.catalog import AntaCatalog
@@ -49,7 +49,7 @@ 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: list[str] = list(AntaTestStatus)
 HIDE_STATUS.remove("unset")
 
 
@@ -147,3 +147,4 @@ def nrfu(
 nrfu.add_command(commands.json)
 nrfu.add_command(commands.text)
 nrfu.add_command(commands.tpl_report)
+nrfu.add_command(commands.md_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, save_to_csv
+from .utils import print_jinja, print_json, print_table, print_text, run_tests, save_markdown_report, save_to_csv
 
 logger = logging.getLogger(__name__)
 
@@ -28,7 +28,7 @@
     required=False,
 )
 def table(ctx: click.Context, group_by: Literal["device", "test"] | None) -> None:
-    """ANTA command to check network states with table result."""
+    """ANTA command to check network state with table results."""
     run_tests(ctx)
     print_table(ctx, group_by=group_by)
     exit_with_code(ctx)
@@ -42,10 +42,10 @@ def table(ctx: click.Context, group_by: Literal["device", "test"] | None) -> Non
     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 file",
+    help="Path to save report as a JSON file",
 )
 def json(ctx: click.Context, output: pathlib.Path | None) -> None:
-    """ANTA command to check network state with JSON result."""
+    """ANTA command to check network state with JSON results."""
     run_tests(ctx)
     print_json(ctx, output=output)
     exit_with_code(ctx)
@@ -54,7 +54,7 @@ def json(ctx: click.Context, output: pathlib.Path | None) -> None:
 @click.command()
 @click.pass_context
 def text(ctx: click.Context) -> None:
-    """ANTA command to check network states with text result."""
+    """ANTA command to check network state with text results."""
     run_tests(ctx)
     print_text(ctx)
     exit_with_code(ctx)
@@ -105,3 +105,19 @@ def tpl_report(ctx: click.Context, template: pathlib.Path, output: pathlib.Path
     run_tests(ctx)
     print_jinja(results=ctx.obj["result_manager"], template=template, output=output)
     exit_with_code(ctx)
+
+
+@click.command()
+@click.pass_context
+@click.option(
+    "--md-output",
+    type=click.Path(file_okay=True, dir_okay=False, exists=False, writable=True, path_type=pathlib.Path),
+    show_envvar=True,
+    required=True,
+    help="Path to save the report as a Markdown file",
+)
+def md_report(ctx: click.Context, md_output: pathlib.Path) -> None:
+    """ANTA command to check network state with Markdown report."""
+    run_tests(ctx)
+    save_markdown_report(ctx, md_output=md_output)
+    exit_with_code(ctx)

anta/cli/nrfu/utils.py

@@ -19,6 +19,7 @@
 from anta.models import AntaTest
 from anta.reporter import ReportJinja, ReportTable
 from anta.reporter.csv_reporter import ReportCsv
+from anta.reporter.md_reporter import MDReportGenerator
 from anta.runner import main
 
 if TYPE_CHECKING:
@@ -94,14 +95,21 @@ def print_table(ctx: click.Context, group_by: Literal["device", "test"] | None =
 
 
 def print_json(ctx: click.Context, output: pathlib.Path | None = None) -> None:
-    """Print result in a json format."""
+    """Print results as JSON. If output is provided, save to file instead."""
     results = _get_result_manager(ctx)
-    console.print()
-    console.print(Panel("JSON results", style="cyan"))
-    rich.print_json(results.json)
-    if output is not None:
-        with output.open(mode="w", encoding="utf-8") as fout:
-            fout.write(results.json)
+
+    if output is None:
+        console.print()
+        console.print(Panel("JSON results", style="cyan"))
+        rich.print_json(results.json)
+    else:
+        try:
+            with output.open(mode="w", encoding="utf-8") as file:
+                file.write(results.json)
+            console.print(f"JSON results saved to {output} ✅", style="cyan")
+        except OSError:
+            console.print(f"Failed to save JSON results to {output} ❌", style="cyan")
+            ctx.exit(ExitCode.USAGE_ERROR)
 
 
 def print_text(ctx: click.Context) -> None:
@@ -134,6 +142,22 @@ def save_to_csv(ctx: click.Context, csv_file: pathlib.Path) -> None:
         ctx.exit(ExitCode.USAGE_ERROR)
 
 
+def save_markdown_report(ctx: click.Context, md_output: pathlib.Path) -> None:
+    """Save the markdown report to a file.
+
+    Parameters
+    ----------
+        ctx: Click context containing the result manager.
+        md_output: Path to save the markdown report.
+    """
+    try:
+        MDReportGenerator.generate(results=_get_result_manager(ctx), md_filename=md_output)
+        console.print(f"Markdown report saved to {md_output} ✅", style="cyan")
+    except OSError:
+        console.print(f"Failed to save Markdown report to {md_output} ❌", 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/cli/utils.py

@@ -112,7 +112,7 @@ def resolve_command(self, ctx: click.Context, args: Any) -> Any:
         return cmd.name, cmd, args
 
 
-def inventory_options(f: Callable[..., Any]) -> Callable[..., Any]:
+def core_options(f: Callable[..., Any]) -> Callable[..., Any]:
     """Click common options when requiring an inventory to interact with devices."""
 
     @click.option(
@@ -190,22 +190,12 @@ def inventory_options(f: Callable[..., Any]) -> Callable[..., Any]:
         required=True,
         type=click.Path(file_okay=True, dir_okay=False, exists=True, readable=True, path_type=Path),
     )
-    @click.option(
-        "--tags",
-        help="List of tags using comma as separator: tag1,tag2,tag3.",
-        show_envvar=True,
-        envvar="ANTA_TAGS",
-        type=str,
-        required=False,
-        callback=parse_tags,
-    )
     @click.pass_context
     @functools.wraps(f)
     def wrapper(
         ctx: click.Context,
         *args: tuple[Any],
         inventory: Path,
-        tags: set[str] | None,
         username: str,
         password: str | None,
         enable_password: str | None,
@@ -219,7 +209,7 @@ def wrapper(
         # pylint: disable=too-many-arguments
         # If help is invoke somewhere, do not parse inventory
         if ctx.obj.get("_anta_help"):
-            return f(*args, inventory=None, tags=tags, **kwargs)
+            return f(*args, inventory=None, **kwargs)
         if prompt:
             # User asked for a password prompt
             if password is None:
@@ -255,7 +245,37 @@ def wrapper(
             )
         except (TypeError, ValueError, YAMLError, OSError, InventoryIncorrectSchemaError, InventoryRootKeyError):
             ctx.exit(ExitCode.USAGE_ERROR)
-        return f(*args, inventory=i, tags=tags, **kwargs)
+        return f(*args, inventory=i, **kwargs)
+
+    return wrapper
+
+
+def inventory_options(f: Callable[..., Any]) -> Callable[..., Any]:
+    """Click common options when requiring an inventory to interact with devices."""
+
+    @core_options
+    @click.option(
+        "--tags",
+        help="List of tags using comma as separator: tag1,tag2,tag3.",
+        show_envvar=True,
+        envvar="ANTA_TAGS",
+        type=str,
+        required=False,
+        callback=parse_tags,
+    )
+    @click.pass_context
+    @functools.wraps(f)
+    def wrapper(
+        ctx: click.Context,
+        *args: tuple[Any],
+        tags: set[str] | None,
+        **kwargs: dict[str, Any],
+    ) -> Any:
+        # pylint: disable=too-many-arguments
+        # If help is invoke somewhere, do not parse inventory
+        if ctx.obj.get("_anta_help"):
+            return f(*args, tags=tags, **kwargs)
+        return f(*args, tags=tags, **kwargs)
 
     return wrapper
 

anta/constants.py

@@ -0,0 +1,19 @@
+# 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.
+"""Constants used in ANTA."""
+
+from __future__ import annotations
+
+ACRONYM_CATEGORIES: set[str] = {"aaa", "mlag", "snmp", "bgp", "ospf", "vxlan", "stp", "igmp", "ip", "lldp", "ntp", "bfd", "ptp", "lanz", "stun", "vlan"}
+"""A set of network protocol or feature acronyms that should be represented in uppercase."""
+
+MD_REPORT_TOC = """**Table of Contents:**
+
+- [ANTA Report](#anta-report)
+  - [Test Results Summary](#test-results-summary)
+    - [Summary Totals](#summary-totals)
+    - [Summary Totals Device Under Test](#summary-totals-device-under-test)
+    - [Summary Totals Per Category](#summary-totals-per-category)
+  - [Test Results](#test-results)"""
+"""Table of Contents for the Markdown report."""

anta/custom_types.py

@@ -112,9 +112,6 @@ def validate_regex(value: str) -> str:
     return value
 
 
-# ANTA framework
-TestStatus = Literal["unset", "success", "failure", "error", "skipped"]
-
 # AntaTest.Input types
 AAAAuthMethod = Annotated[str, AfterValidator(aaa_group_prefix)]
 Vlan = Annotated[int, Field(ge=0, le=4094)]
@@ -198,3 +195,4 @@ def validate_regex(value: str) -> str:
     "prefixRtMembershipDroppedMaxRouteLimitViolated",
 ]
 BgpUpdateError = Literal["inUpdErrWithdraw", "inUpdErrIgnore", "inUpdErrDisableAfiSafi", "disabledAfiSafi", "lastUpdErrTime"]
+BfdProtocol = Literal["bgp", "isis", "lag", "ospf", "ospfv3", "pim", "route-input", "static-bfd", "static-route", "vrrp", "vxlan"]